quay/endpoints/api/test/test_logs_models_pre_oci.py

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})]]
  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(18, 4, '1')]