897a091692
### Description of Changes ran yapf for the branch [TESTING->locally using docker compose] Issue: https://coreosdev.atlassian.net/browse/QUAY-632 ## Reviewer Checklist - [ ] It works! - [ ] Comments provide sufficient explanations for the next contributor - [ ] Tests cover changes and corner cases - [ ] Follows Quay syntax patterns and format
344 lines
12 KiB
Python
344 lines
12 KiB
Python
import pytest
|
|
|
|
from data.model import DataModelException, InvalidImageException
|
|
from endpoints.api.tag_models_interface import RepositoryTagHistory, Tag, Repository
|
|
from mock import Mock, call
|
|
|
|
from data import model
|
|
from endpoints.api.tag_models_pre_oci import pre_oci_model
|
|
from util.morecollections import AttrDict
|
|
|
|
EMPTY_REPOSITORY = 'empty_repository'
|
|
EMPTY_NAMESPACE = 'empty_namespace'
|
|
BAD_REPOSITORY_NAME = 'bad_repository_name'
|
|
BAD_NAMESPACE_NAME = 'bad_namespace_name'
|
|
|
|
|
|
@pytest.fixture
|
|
def get_monkeypatch(monkeypatch):
|
|
return monkeypatch
|
|
|
|
|
|
def mock_out_get_repository(monkeypatch, namespace_name, repository_name):
|
|
def return_none(namespace_name, repository_name):
|
|
return None
|
|
|
|
def return_repository(namespace_name, repository_name):
|
|
return 'repository'
|
|
|
|
if namespace_name == BAD_NAMESPACE_NAME or repository_name == BAD_REPOSITORY_NAME:
|
|
return_function = return_none
|
|
else:
|
|
return_function = return_repository
|
|
|
|
monkeypatch.setattr(model.repository, 'get_repository', return_function)
|
|
|
|
|
|
def create_mock_tag(name, reversion, lifetime_start_ts, lifetime_end_ts, mock_id, docker_image_id,
|
|
manifest_list):
|
|
tag_mock = Mock()
|
|
tag_mock.name = name
|
|
image_mock = Mock()
|
|
image_mock.docker_image_id = docker_image_id
|
|
tag_mock.image = image_mock
|
|
tag_mock.reversion = reversion
|
|
tag_mock.lifetime_start_ts = lifetime_start_ts
|
|
tag_mock.lifetime_end_ts = lifetime_end_ts
|
|
tag_mock.id = mock_id
|
|
tag_mock.manifest_list = manifest_list
|
|
tag = Tag(name=name, reversion=reversion, image=image_mock, docker_image_id=docker_image_id,
|
|
lifetime_start_ts=lifetime_start_ts, lifetime_end_ts=lifetime_end_ts,
|
|
manifest_list=manifest_list)
|
|
return tag_mock, tag
|
|
|
|
|
|
first_mock, first_tag = create_mock_tag('tag1', 'rev1', 'start1', 'end1', 'id1',
|
|
'docker_image_id1', [])
|
|
second_mock, second_tag = create_mock_tag('tag2', 'rev2', 'start2', 'end2', 'id2',
|
|
'docker_image_id2', ['manifest'])
|
|
|
|
|
|
def mock_out_list_repository_tag_history(monkeypatch, namespace_name, repository_name, page, size,
|
|
specific_tag):
|
|
def list_empty_tag_history(repository, page, size, specific_tag):
|
|
return [], {}, False
|
|
|
|
def list_filled_tag_history(repository, page, size, specific_tag):
|
|
tags = [first_mock, second_mock]
|
|
return tags, {
|
|
first_mock.id: first_mock.manifest_list,
|
|
second_mock.id: second_mock.manifest_list
|
|
}, len(tags) > size
|
|
|
|
def list_only_second_tag(repository, page, size, specific_tag):
|
|
tags = [second_mock]
|
|
return tags, {second_mock.id: second_mock.manifest_list}, len(tags) > size
|
|
|
|
if namespace_name == EMPTY_NAMESPACE or repository_name == EMPTY_REPOSITORY:
|
|
return_function = list_empty_tag_history
|
|
else:
|
|
if specific_tag == 'tag2':
|
|
return_function = list_only_second_tag
|
|
else:
|
|
return_function = list_filled_tag_history
|
|
|
|
monkeypatch.setattr(model.tag, 'list_repository_tag_history', return_function)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'expected, namespace_name, repository_name, page, size, specific_tag', [
|
|
(None, BAD_NAMESPACE_NAME, 'repository_name', 1, 100, None),
|
|
(None, 'namespace_name', BAD_REPOSITORY_NAME, 1, 100, None),
|
|
(RepositoryTagHistory(tags=[], more=False), EMPTY_NAMESPACE, EMPTY_REPOSITORY, 1, 100, None),
|
|
(RepositoryTagHistory(tags=[first_tag, second_tag], more=False), 'namespace', 'repository', 1,
|
|
100, None),
|
|
(RepositoryTagHistory(tags=[first_tag, second_tag], more=True), 'namespace', 'repository', 1,
|
|
1, None),
|
|
(RepositoryTagHistory(tags=[second_tag], more=False), 'namespace', 'repository', 1, 100,
|
|
'tag2'),
|
|
])
|
|
def test_list_repository_tag_history(expected, namespace_name, repository_name, page, size,
|
|
specific_tag, get_monkeypatch):
|
|
mock_out_get_repository(get_monkeypatch, namespace_name, repository_name)
|
|
mock_out_list_repository_tag_history(get_monkeypatch, namespace_name, repository_name, page,
|
|
size, specific_tag)
|
|
assert pre_oci_model.list_repository_tag_history(namespace_name, repository_name, page, size,
|
|
specific_tag) == expected
|
|
|
|
|
|
def get_repo_image_mock(monkeypatch, return_value):
|
|
def return_return_value(namespace_name, repository_name, image_id):
|
|
return return_value
|
|
|
|
monkeypatch.setattr(model.image, 'get_repo_image', return_return_value)
|
|
|
|
|
|
def test_get_repo_not_exists(get_monkeypatch):
|
|
namespace_name = 'namespace_name'
|
|
repository_name = 'repository_name'
|
|
image_id = 'image_id'
|
|
get_repo_image_mock(get_monkeypatch, None)
|
|
|
|
repo = pre_oci_model.get_repo(namespace_name, repository_name, image_id)
|
|
|
|
assert repo is None
|
|
|
|
|
|
def test_get_repo_exists(get_monkeypatch):
|
|
namespace_name = 'namespace_name'
|
|
repository_name = 'repository_name'
|
|
image_id = 'image_id'
|
|
mock = Mock()
|
|
mock.namespace_user = namespace_name
|
|
mock.name = repository_name
|
|
mock.repository = mock
|
|
get_repo_image_mock(get_monkeypatch, mock)
|
|
|
|
repo = pre_oci_model.get_repo(namespace_name, repository_name, image_id)
|
|
|
|
assert repo is not None
|
|
assert repo.repository_name == repository_name
|
|
assert repo.namespace_name == namespace_name
|
|
|
|
|
|
def get_repository_mock(monkeypatch, return_value):
|
|
def return_return_value(namespace_name, repository_name, kind_filter=None):
|
|
return return_value
|
|
|
|
monkeypatch.setattr(model.repository, 'get_repository', return_return_value)
|
|
|
|
|
|
def get_repo_tag_image_mock(monkeypatch, return_value):
|
|
def return_return_value(repo, tag_name, include_storage=False):
|
|
return return_value
|
|
|
|
monkeypatch.setattr(model.tag, 'get_repo_tag_image', return_return_value)
|
|
|
|
|
|
def test_get_repo_tag_image_with_repo_and_repo_tag(get_monkeypatch):
|
|
mock_storage = Mock()
|
|
mock_image = Mock()
|
|
mock_image.docker_image_id = 'some docker image id'
|
|
mock_image.created = 1235
|
|
mock_image.comment = 'some comment'
|
|
mock_image.command = 'some command'
|
|
mock_image.storage = mock_storage
|
|
mock_image.ancestors = []
|
|
|
|
get_repository_mock(get_monkeypatch, mock_image)
|
|
get_repo_tag_image_mock(get_monkeypatch, mock_image)
|
|
|
|
image = pre_oci_model.get_repo_tag_image(
|
|
Repository('namespace_name', 'repository_name'), 'tag_name')
|
|
|
|
assert image is not None
|
|
assert image.docker_image_id == 'some docker image id'
|
|
|
|
|
|
def test_get_repo_tag_image_without_repo(get_monkeypatch):
|
|
get_repository_mock(get_monkeypatch, None)
|
|
|
|
image = pre_oci_model.get_repo_tag_image(
|
|
Repository('namespace_name', 'repository_name'), 'tag_name')
|
|
|
|
assert image is None
|
|
|
|
|
|
def test_get_repo_tag_image_without_repo_tag_image(get_monkeypatch):
|
|
mock = Mock()
|
|
mock.docker_image_id = 'some docker image id'
|
|
get_repository_mock(get_monkeypatch, mock)
|
|
|
|
def raise_exception(repo, tag_name, include_storage=False):
|
|
raise DataModelException()
|
|
|
|
get_monkeypatch.setattr(model.tag, 'get_repo_tag_image', raise_exception)
|
|
|
|
image = pre_oci_model.get_repo_tag_image(
|
|
Repository('namespace_name', 'repository_name'), 'tag_name')
|
|
|
|
assert image is None
|
|
|
|
|
|
def test_create_or_update_tag(get_monkeypatch):
|
|
mock = Mock()
|
|
get_monkeypatch.setattr(model.tag, 'create_or_update_tag', mock)
|
|
|
|
pre_oci_model.create_or_update_tag('namespace_name', 'repository_name', 'tag_name',
|
|
'docker_image_id')
|
|
|
|
assert mock.call_count == 1
|
|
assert mock.call_args == call('namespace_name', 'repository_name', 'tag_name', 'docker_image_id')
|
|
|
|
|
|
def test_delete_tag(get_monkeypatch):
|
|
mock = Mock()
|
|
get_monkeypatch.setattr(model.tag, 'delete_tag', mock)
|
|
|
|
pre_oci_model.delete_tag('namespace_name', 'repository_name', 'tag_name')
|
|
|
|
assert mock.call_count == 1
|
|
assert mock.call_args == call('namespace_name', 'repository_name', 'tag_name')
|
|
|
|
|
|
def test_get_parent_images_with_exception(get_monkeypatch):
|
|
mock = Mock(side_effect=InvalidImageException)
|
|
|
|
get_monkeypatch.setattr(model.image, 'get_image_by_id', mock)
|
|
|
|
images = pre_oci_model.get_parent_images('namespace_name', 'repository_name', 'tag_name')
|
|
assert images == []
|
|
|
|
|
|
def test_get_parent_images_empty_parent_images(get_monkeypatch):
|
|
get_image_by_id_mock = Mock()
|
|
get_monkeypatch.setattr(model.image, 'get_image_by_id', get_image_by_id_mock)
|
|
|
|
get_parent_images_mock = Mock(return_value=[])
|
|
get_monkeypatch.setattr(model.image, 'get_parent_images', get_parent_images_mock)
|
|
|
|
images = pre_oci_model.get_parent_images('namespace_name', 'repository_name', 'tag_name')
|
|
assert images == []
|
|
|
|
|
|
def compare_images(parent_image, attr_dict):
|
|
for field in parent_image._fields:
|
|
assert getattr(parent_image, field) == getattr(attr_dict, field)
|
|
|
|
|
|
def test_get_parent_images(get_monkeypatch):
|
|
get_image_by_id_mock = Mock()
|
|
get_monkeypatch.setattr(model.image, 'get_image_by_id', get_image_by_id_mock)
|
|
|
|
def fake_list():
|
|
return []
|
|
|
|
image_one = AttrDict({
|
|
'docker_image_id': 'docker_image_id',
|
|
'created': 'created_one',
|
|
'comment': 'comment_one',
|
|
'command': 'command_one',
|
|
'storage': AttrDict({
|
|
'image_size': 'image_size_one',
|
|
'uploading': 'uploading_one'
|
|
}),
|
|
'ancestors': 'one/two/three',
|
|
'ancestor_id_list': fake_list
|
|
})
|
|
image_two = AttrDict({
|
|
'docker_image_id': 'docker_image_id_two',
|
|
'created': 'created',
|
|
'comment': 'comment',
|
|
'command': 'command',
|
|
'storage': AttrDict({
|
|
'image_size': 'image_size',
|
|
'uploading': 'uploading'
|
|
}),
|
|
'ancestors': 'four/five/six',
|
|
'ancestor_id_list': fake_list
|
|
})
|
|
|
|
get_parent_images_mock = Mock(return_value=[image_one, image_two])
|
|
get_monkeypatch.setattr(model.image, 'get_parent_images', get_parent_images_mock)
|
|
get_monkeypatch.setattr(model.image, 'get_image_by_id', Mock())
|
|
|
|
images = pre_oci_model.get_parent_images('namespace_name', 'repository_name', 'tag_name')
|
|
image_one.ancestor_id_list = []
|
|
image_one.storage_image_size = 'image_size_one'
|
|
image_one.storage_uploading = 'uploading_one'
|
|
image_one.ancestor_length = 13
|
|
image_two.ancestor_id_list = []
|
|
image_two.storage_uploading = 'uploading'
|
|
image_two.storage_image_size = 'image_size'
|
|
image_two.ancestor_length = 13
|
|
compare_images(images[0], image_one)
|
|
compare_images(images[1], image_two)
|
|
|
|
|
|
def test_list_repository_tags(get_monkeypatch):
|
|
mock = Mock(return_value=[])
|
|
|
|
get_monkeypatch.setattr(model.tag, 'list_repository_tags', mock)
|
|
|
|
pre_oci_model.list_repository_tags('namespace_name', 'repository_name')
|
|
|
|
mock.assert_called_once_with('namespace_name', 'repository_name')
|
|
|
|
|
|
def test_get_repository(get_monkeypatch):
|
|
mock = Mock()
|
|
|
|
get_monkeypatch.setattr(model.repository, 'get_repository', mock)
|
|
|
|
pre_oci_model.get_repository('namespace_name', 'repository_name')
|
|
|
|
mock.assert_called_once_with('namespace_name', 'repository_name')
|
|
|
|
|
|
def test_tag_to_manifest(get_monkeypatch):
|
|
repo_mock = Mock()
|
|
restore_tag_mock = Mock(return_value=None)
|
|
get_repository_mock = Mock(return_value=repo_mock)
|
|
|
|
get_monkeypatch.setattr(model.tag, 'restore_tag_to_manifest', restore_tag_mock)
|
|
get_monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock)
|
|
|
|
pre_oci_model.restore_tag_to_manifest(
|
|
Repository('namespace', 'repository'), 'tag_name', 'manifest_digest')
|
|
|
|
get_repository_mock.assert_called_once_with('namespace', 'repository')
|
|
restore_tag_mock.assert_called_once_with(repo_mock, 'tag_name', 'manifest_digest')
|
|
|
|
|
|
def test__tag_to_image(get_monkeypatch):
|
|
repo_mock = Mock()
|
|
restore_tag_mock = Mock(return_value=None)
|
|
get_repository_mock = Mock(return_value=repo_mock)
|
|
|
|
get_monkeypatch.setattr(model.tag, 'restore_tag_to_image', restore_tag_mock)
|
|
get_monkeypatch.setattr(model.repository, 'get_repository', get_repository_mock)
|
|
|
|
pre_oci_model.restore_tag_to_image(Repository('namespace', 'repository'), 'tag_name', 'image_id')
|
|
|
|
get_repository_mock.assert_called_once_with('namespace', 'repository')
|
|
restore_tag_mock.assert_called_once_with(repo_mock, 'tag_name', 'image_id')
|