Change tags API endpoint to use new registry model interface
This commit is contained in:
Joseph Schorr
parent
8225c61a1f
commit
bc99dd7963
7 changed files with 148 additions and 841 deletions
|
|
@ -788,11 +788,11 @@ SECURITY_TESTS = [
|
|||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'public/publicrepo'}, {u'image': 'WXNG'}, 'freshuser', 403),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'public/publicrepo'}, {u'image': 'WXNG'}, 'reader', 403),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'devtable/shared'}, {u'image': 'WXNG'}, None, 401),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'devtable/shared'}, {u'image': 'WXNG'}, 'devtable', 400),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'devtable/shared'}, {u'image': 'WXNG'}, 'devtable', 404),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'devtable/shared'}, {u'image': 'WXNG'}, 'freshuser', 403),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'devtable/shared'}, {u'image': 'WXNG'}, 'reader', 403),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'buynlarge/orgrepo'}, {u'image': 'WXNG'}, None, 401),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'buynlarge/orgrepo'}, {u'image': 'WXNG'}, 'devtable', 400),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'buynlarge/orgrepo'}, {u'image': 'WXNG'}, 'devtable', 404),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'buynlarge/orgrepo'}, {u'image': 'WXNG'}, 'freshuser', 403),
|
||||
(RestoreTag, 'POST', {'tag': 'HP8R', 'repository': 'buynlarge/orgrepo'}, {u'image': 'WXNG'}, 'reader', 403),
|
||||
|
||||
|
|
|
|||
|
|
@ -1,129 +1,13 @@
|
|||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from mock import patch, Mock, MagicMock, call
|
||||
from data.registry_model import registry_model
|
||||
|
||||
from data.model import DataModelException
|
||||
from endpoints.api.tag_models_interface import RepositoryTagHistory, Tag
|
||||
from endpoints.api.test.shared import conduct_api_call
|
||||
from endpoints.test.shared import client_with_identity
|
||||
from endpoints.api.tag import RepositoryTag, RestoreTag, ListRepositoryTags, RepositoryTagImages
|
||||
|
||||
from features import FeatureNameValue
|
||||
|
||||
from test.fixtures import *
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def get_repo_image():
|
||||
def mock_callable(namespace, repository, image_id):
|
||||
mock = Mock(namespace_user='devtable')
|
||||
mock.name = 'simple'
|
||||
img = Mock(repository=mock, docker_image_id=12) if image_id == 'image1' else None
|
||||
return img
|
||||
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.image.get_repo_image',
|
||||
side_effect=mock_callable) as mk:
|
||||
yield mk
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def get_repository():
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.image.get_repo_image',
|
||||
return_value='mock_repo') as mk:
|
||||
yield mk
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def get_repo_tag_image():
|
||||
def mock_get_repo_tag_image(repository, tag):
|
||||
storage_mock = Mock(image_size=1234, uploading='uploading')
|
||||
|
||||
def fake_ancestor_id_list():
|
||||
return []
|
||||
|
||||
if tag == 'existing-tag':
|
||||
return Mock(docker_image_id='mock_docker_image_id', created=12345, comment='comment',
|
||||
command='command', storage=storage_mock, ancestors=[],
|
||||
ancestor_id_list=fake_ancestor_id_list)
|
||||
else:
|
||||
raise DataModelException('Unable to find image for tag.')
|
||||
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.tag.get_repo_tag_image',
|
||||
side_effect=mock_get_repo_tag_image):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def restore_tag_to_manifest():
|
||||
def mock_restore_tag_to_manifest(repository, tag, manifest_digest):
|
||||
tag_img = Mock(docker_image_id='mock_docker_image_id') if tag == 'existing-tag' else None
|
||||
return tag_img
|
||||
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.tag.restore_tag_to_manifest',
|
||||
side_effect=mock_restore_tag_to_manifest):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def restore_tag_to_image():
|
||||
def mock_restore_tag_to_image(repository, tag, image_id):
|
||||
tag_img = Mock(docker_image_id='mock_docker_image_id') if tag == 'existing-tag' else None
|
||||
return tag_img
|
||||
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.tag.restore_tag_to_image',
|
||||
side_effect=mock_restore_tag_to_image):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def create_or_update_tag():
|
||||
with patch('endpoints.api.tag_models_pre_oci.model.tag.create_or_update_tag') as mk:
|
||||
yield mk
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def generate_manifest():
|
||||
def mock_callable(namespace, repository, tag):
|
||||
if tag == 'generatemanifestfail':
|
||||
raise Exception('test_failure')
|
||||
|
||||
with patch('endpoints.api.tag._generate_and_store_manifest', side_effect=mock_callable) as mk:
|
||||
yield mk
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def authd_client(client):
|
||||
with client_with_identity('devtable', client) as cl:
|
||||
yield cl
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def list_repository_tag_history():
|
||||
def list_repository_tag_history(namespace_name, repository_name, page, size, specific_tag):
|
||||
return RepositoryTagHistory(tags=[
|
||||
Tag(name='First Tag', image='image', reversion=False, lifetime_start_ts=0, lifetime_end_ts=0,
|
||||
manifest_list=[], docker_image_id='first docker image id'),
|
||||
Tag(name='Second Tag', image='second image', reversion=True, lifetime_start_ts=10,
|
||||
lifetime_end_ts=100, manifest_list=[], docker_image_id='second docker image id')
|
||||
], more=False)
|
||||
|
||||
with patch('endpoints.api.tag.model.list_repository_tag_history',
|
||||
side_effect=list_repository_tag_history):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def find_no_repo_tag_history():
|
||||
def list_repository_tag_history(namespace_name, repository_name, page, size, specific_tag):
|
||||
return None
|
||||
|
||||
with patch('endpoints.api.tag.model.list_repository_tag_history',
|
||||
side_effect=list_repository_tag_history):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.mark.parametrize('expiration_time, expected_status', [
|
||||
(None, 201),
|
||||
('aksdjhasd', 400),
|
||||
|
|
@ -161,126 +45,50 @@ def test_change_tag_expiration(client, app):
|
|||
assert tag.lifetime_end_ts == updated_expiration
|
||||
|
||||
|
||||
@pytest.mark.parametrize('test_image,test_tag,expected_status', [
|
||||
('image1', '-INVALID-TAG-NAME', 400),
|
||||
('image1', '.INVALID-TAG-NAME', 400),
|
||||
('image1',
|
||||
@pytest.mark.parametrize('image_exists,test_tag,expected_status', [
|
||||
(True, '-INVALID-TAG-NAME', 400),
|
||||
(True, '.INVALID-TAG-NAME', 400),
|
||||
(True,
|
||||
'INVALID-TAG_NAME-BECAUSE-THIS-IS-WAY-WAY-TOO-LOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOONG',
|
||||
400),
|
||||
('nonexistantimage', 'newtag', 404),
|
||||
('image1', 'generatemanifestfail', None),
|
||||
('image1', 'existing-tag', 201),
|
||||
('image1', 'newtag', 201),
|
||||
(False, 'newtag', 404),
|
||||
(True, 'generatemanifestfail', None),
|
||||
(True, 'latest', 201),
|
||||
(True, 'newtag', 201),
|
||||
])
|
||||
def test_move_tag(test_image, test_tag, expected_status, get_repo_image, get_repo_tag_image,
|
||||
create_or_update_tag, generate_manifest, authd_client):
|
||||
params = {'repository': 'devtable/simple', 'tag': test_tag}
|
||||
request_body = {'image': test_image}
|
||||
if expected_status is None:
|
||||
with pytest.raises(Exception):
|
||||
conduct_api_call(authd_client, RepositoryTag, 'put', params, request_body, expected_status)
|
||||
else:
|
||||
conduct_api_call(authd_client, RepositoryTag, 'put', params, request_body, expected_status)
|
||||
def test_move_tag(image_exists, test_tag, expected_status, client, app):
|
||||
with client_with_identity('devtable', client) as cl:
|
||||
test_image = 'unknown'
|
||||
if image_exists:
|
||||
repo_ref = registry_model.lookup_repository('devtable', 'simple')
|
||||
tag_ref = registry_model.get_repo_tag(repo_ref, 'latest', include_legacy_image=True)
|
||||
assert tag_ref
|
||||
|
||||
test_image = tag_ref.legacy_image.docker_image_id
|
||||
|
||||
params = {'repository': 'devtable/simple', 'tag': test_tag}
|
||||
request_body = {'image': test_image}
|
||||
if expected_status is None:
|
||||
with pytest.raises(Exception):
|
||||
conduct_api_call(cl, RepositoryTag, 'put', params, request_body, expected_status)
|
||||
else:
|
||||
conduct_api_call(cl, RepositoryTag, 'put', params, request_body, expected_status)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'namespace, repository, specific_tag, page, limit, expected_response_code, expected', [
|
||||
('devtable', 'simple', None, 1, 10, 200, {
|
||||
'has_additional': False
|
||||
}),
|
||||
('devtable', 'simple', None, 1, 10, 200, {
|
||||
'page': 1
|
||||
}),
|
||||
('devtable', 'simple', None, 1, 10, 200, {
|
||||
'tags': [{
|
||||
'docker_image_id': 'first docker image id',
|
||||
'name': 'First Tag',
|
||||
'reversion': False
|
||||
}, {
|
||||
'docker_image_id': 'second docker image id',
|
||||
'end_ts': 100,
|
||||
'name': 'Second Tag',
|
||||
'reversion': True,
|
||||
'start_ts': 10
|
||||
}]
|
||||
}),
|
||||
])
|
||||
def test_list_repository_tags_view_is_correct(namespace, repository, specific_tag, page, limit,
|
||||
list_repository_tag_history, expected_response_code,
|
||||
expected, authd_client):
|
||||
params = {
|
||||
'repository': namespace + '/' + repository,
|
||||
'specificTag': specific_tag,
|
||||
'page': page,
|
||||
'limit': limit
|
||||
}
|
||||
response = conduct_api_call(authd_client, ListRepositoryTags, 'get', params,
|
||||
expected_code=expected_response_code)
|
||||
compare_list_history_tags_response(expected, response.json)
|
||||
|
||||
|
||||
def compare_list_history_tags_response(expected, actual):
|
||||
if 'has_additional' in expected:
|
||||
assert expected['has_additional'] == actual['has_additional']
|
||||
|
||||
if 'page' in expected:
|
||||
assert expected['page'] == actual['page']
|
||||
|
||||
if 'tags' in expected:
|
||||
assert expected['tags'] == actual['tags']
|
||||
|
||||
|
||||
def test_no_repo_tag_history(find_no_repo_tag_history, authd_client):
|
||||
params = {'repository': 'devtable/simple', 'specificTag': None, 'page': 1, 'limit': 10}
|
||||
conduct_api_call(authd_client, ListRepositoryTags, 'get', params, expected_code=404)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'specific_tag, page, limit, expected_specific_tag, expected_page, expected_limit', [
|
||||
(None, None, None, None, 1, 50),
|
||||
('specific_tag', 12, 13, 'specific_tag', 12, 13),
|
||||
('specific_tag', -1, 101, 'specific_tag', 1, 100),
|
||||
('specific_tag', 0, 0, 'specific_tag', 1, 1),
|
||||
])
|
||||
def test_repo_tag_history_param_parse(specific_tag, page, limit, expected_specific_tag,
|
||||
expected_page, expected_limit, authd_client):
|
||||
mock = MagicMock()
|
||||
mock.return_value = RepositoryTagHistory(tags=[], more=False)
|
||||
|
||||
with patch('endpoints.api.tag.model.list_repository_tag_history', side_effect=mock):
|
||||
params = {
|
||||
'repository': 'devtable/simple',
|
||||
'specificTag': specific_tag,
|
||||
'page': page,
|
||||
'limit': limit
|
||||
}
|
||||
conduct_api_call(authd_client, ListRepositoryTags, 'get', params)
|
||||
|
||||
assert mock.call_args == call(namespace_name='devtable', repository_name='simple',
|
||||
page=expected_page, size=expected_limit,
|
||||
specific_tag=expected_specific_tag)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('test_manifest,test_tag,manifest_generated,expected_status', [
|
||||
(None, 'newtag', True, 200),
|
||||
(None, 'generatemanifestfail', True, None),
|
||||
('manifest1', 'newtag', False, 200),
|
||||
@pytest.mark.parametrize('repo_namespace, repo_name', [
|
||||
('devtable', 'simple'),
|
||||
('devtable', 'history'),
|
||||
('devtable', 'complex'),
|
||||
('buynlarge', 'orgrepo'),
|
||||
])
|
||||
def test_restore_tag(test_manifest, test_tag, manifest_generated, expected_status, get_repository,
|
||||
restore_tag_to_manifest, restore_tag_to_image, generate_manifest,
|
||||
authd_client):
|
||||
params = {'repository': 'devtable/simple', 'tag': test_tag}
|
||||
request_body = {'image': 'image1'}
|
||||
if test_manifest is not None:
|
||||
request_body['manifest_digest'] = test_manifest
|
||||
if expected_status is None:
|
||||
with pytest.raises(Exception):
|
||||
conduct_api_call(authd_client, RestoreTag, 'post', params, request_body, expected_status)
|
||||
else:
|
||||
conduct_api_call(authd_client, RestoreTag, 'post', params, request_body, expected_status)
|
||||
def test_list_repo_tags(repo_namespace, repo_name, client, app):
|
||||
params = {'repository': repo_namespace + '/' + repo_name}
|
||||
with client_with_identity('devtable', client) as cl:
|
||||
tags = conduct_api_call(cl, ListRepositoryTags, 'get', params).json['tags']
|
||||
repo_ref = registry_model.lookup_repository(repo_namespace, repo_name)
|
||||
history, _ = registry_model.list_repository_tag_history(repo_ref)
|
||||
assert len(tags) == len(history)
|
||||
|
||||
if manifest_generated:
|
||||
generate_manifest.assert_called_with('devtable', 'simple', test_tag)
|
||||
|
||||
@pytest.mark.parametrize('repository, tag, owned, expect_images', [
|
||||
('devtable/simple', 'prod', False, True),
|
||||
|
|
@ -291,7 +99,8 @@ def test_restore_tag(test_manifest, test_tag, manifest_generated, expected_statu
|
|||
('devtable/complex', 'prod', False, True),
|
||||
('devtable/complex', 'prod', True, True),
|
||||
])
|
||||
def test_list_tag_images(repository, tag, owned, expect_images, authd_client):
|
||||
params = {'repository': repository, 'tag': tag, 'owned': owned}
|
||||
result = conduct_api_call(authd_client, RepositoryTagImages, 'get', params, None, 200).json
|
||||
assert bool(result['images']) == expect_images
|
||||
def test_list_tag_images(repository, tag, owned, expect_images, client, app):
|
||||
with client_with_identity('devtable', client) as cl:
|
||||
params = {'repository': repository, 'tag': tag, 'owned': owned}
|
||||
result = conduct_api_call(cl, RepositoryTagImages, 'get', params, None, 200).json
|
||||
assert bool(result['images']) == expect_images
|
||||
|
|
|
|||
|
|
@ -1,215 +0,0 @@
|
|||
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 get_repo_mock(monkeypatch, return_value):
|
||||
def return_return_value(namespace_name, repository_name):
|
||||
return return_value
|
||||
|
||||
monkeypatch.setattr(model.repository, 'get_repository', return_return_value)
|
||||
|
||||
|
||||
def test_get_repo_not_exists(get_monkeypatch):
|
||||
namespace_name = 'namespace_name'
|
||||
repository_name = 'repository_name'
|
||||
get_repo_mock(get_monkeypatch, None)
|
||||
repo = pre_oci_model.get_repo(namespace_name, repository_name)
|
||||
|
||||
assert repo is None
|
||||
|
||||
|
||||
def test_get_repo_exists(get_monkeypatch):
|
||||
namespace_name = 'namespace_name'
|
||||
repository_name = 'repository_name'
|
||||
mock = Mock()
|
||||
mock.namespace_user = namespace_name
|
||||
mock.name = repository_name
|
||||
mock.repository = mock
|
||||
get_repo_mock(get_monkeypatch, mock)
|
||||
|
||||
repo = pre_oci_model.get_repo(namespace_name, repository_name)
|
||||
|
||||
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 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')
|
||||
Reference in a new issue