6a94eba1a2
The migration to actually drop the table will go in after this code has been pushed to production
129 lines
5.5 KiB
Python
129 lines
5.5 KiB
Python
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})]]
|
|
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(6, 4, '1')]
|