import pytest from mock import Mock from data import model, database from endpoints.api.logs_models_interface import LogEntry, LogEntryPage, AggregatedLogEntry from endpoints.api.logs_models_pre_oci import pre_oci_model from util.morecollections import AttrDict def test_get_logs_query(monkeypatch): get_repository_mock = Mock() monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) get_user_mock = Mock() monkeypatch.setattr(model.user, 'get_user', get_user_mock) get_logs_query_mock = Mock() monkeypatch.setattr(model.log, 'get_logs_query', get_logs_query_mock) paginate_mock = Mock() paginate_mock.return_value = ([], {}) monkeypatch.setattr(model.modelutil, 'paginate', paginate_mock) assert pre_oci_model.get_logs_query('start_time', 'end_time', 'preformer_namne', 'repository_name', 'namespace_name', set(), None) == LogEntryPage([], {}) def test_get_logs_query_returns_list_log_entries(monkeypatch): get_repository_mock = Mock() monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) get_user_mock = Mock() monkeypatch.setattr(model.user, 'get_user', get_user_mock) get_logs_query_mock = Mock() monkeypatch.setattr(model.log, 'get_logs_query', get_logs_query_mock) paginate_mock = Mock() paginate_mock.return_value = ([AttrDict({'kind': 1, 'datetime': 'datetime', 'ip': 'ip', 'metadata_json': '{}', 'account': AttrDict( {'username': 'account_username', 'email': 'account_email', 'robot': False, 'organization': False}), 'performer': AttrDict( {'email': 'performer_email', 'username': 'performer_username', 'robot': False}), 'kind_id': 1})], {'key': 'value'}) monkeypatch.setattr(model.modelutil, 'paginate', paginate_mock) assert pre_oci_model.get_logs_query('start_time', 'end_time', 'performer_username', 'repository_name', 'namespace_name', set(), {'start_id': 1}) == LogEntryPage([ LogEntry('{}', 'ip', 'datetime', 'performer_email', 'performer_username', False, False, 'account_username', 'account_email', False, 1)], {'key': 'value'}) @pytest.mark.skip('Turned off until we move back to a single LogEntry table') def test_get_logs_query_calls_get_repository(monkeypatch): repo_mock = Mock() performer_mock = Mock() query_mock = Mock() get_repository_mock = Mock() get_repository_mock.return_value = repo_mock monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) get_user_mock = Mock() get_user_mock.return_value = performer_mock monkeypatch.setattr(model.user, 'get_user', get_user_mock) get_logs_query_mock = Mock() get_logs_query_mock.return_value = query_mock monkeypatch.setattr(model.log, 'get_logs_query', get_logs_query_mock) paginate_mock = Mock() page_token = {} paginate_mock.return_value = ([], page_token) monkeypatch.setattr(model.modelutil, 'paginate', paginate_mock) ignore = set() pre_oci_model.get_logs_query('start_time', 'end_time', 'performer_username', 'repository_name', 'namespace_name', ignore, page_token) get_repository_mock.assert_called_once_with('namespace_name', 'repository_name') get_user_mock.assert_called_once_with('performer_username') get_logs_query_mock.assert_called_once_with('start_time', 'end_time', performer=performer_mock, repository=repo_mock, namespace='namespace_name', ignore=ignore) paginate_mock.assert_called_once_with(query_mock, database.LogEntry, descending=True, page_token=page_token, limit=20) def test_get_log_entry_kinds(monkeypatch): get_log_entry_kinds_mock = Mock() monkeypatch.setattr(model.log, 'get_log_entry_kinds', get_log_entry_kinds_mock) pre_oci_model.get_log_entry_kinds() get_log_entry_kinds_mock.assert_called_once_with() def test_does_repo_exist_returns_false(monkeypatch): get_repository_mock = Mock() get_repository_mock.return_value = None monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) assert pre_oci_model.repo_exists('namespace_name', 'repository_name') is False def test_does_repo_exist_returns_true(monkeypatch): get_repository_mock = Mock() get_repository_mock.return_value = True monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) assert pre_oci_model.repo_exists('namespace_name', 'repository_name') is True def test_get_aggregated_logs(monkeypatch): get_aggregated_logs_mock = Mock() get_aggregated_logs_mock.side_effect = [[AttrDict({'day': '1', 'kind_id': 4, 'count': 6})], [AttrDict({'day': '1', 'kind_id': 4, 'count': 12})], [AttrDict({'day': '1', 'kind_id': 4, 'count': 3})]] monkeypatch.setattr(model.log, 'get_aggregated_logs', get_aggregated_logs_mock) repo_mock = Mock() get_repository_mock = Mock() get_repository_mock.return_value = repo_mock monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock) performer_mock = Mock() get_user_mock = Mock() get_user_mock.return_value = performer_mock monkeypatch.setattr(model.user, 'get_user', get_user_mock) actual = pre_oci_model.get_aggregated_logs('start_time', 'end_time', 'performer_name', 'repository_name', 'namespace_name', set()) assert actual == [AggregatedLogEntry(21, 4, '1')]