vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Rename pre oci model tests in preparation for the new model (the tests will be applicable to both models)

Browse source
This commit is contained in:
Joseph Schorr 2018-10-29 14:31:59 -04:00
parent 6c29baf3fc
commit 053d918d67
1 changed files with 181 additions and 180 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

361
data/registry_model/test/test_pre_oci_model.py → data/registry_model/test/test_interface.py
View file

@ -21,9 +21,10 @@ from image.docker.schema1 import DockerSchema1ManifestBuilder
from test.fixtures import *
@pytest.fixture()
def pre_oci_model(initialized_db):
return PreOCIModel()
@pytest.fixture(params=[PreOCIModel])
def registry_model(request, initialized_db):
return request.param()
@pytest.mark.parametrize('names, expected', [
@ -33,10 +34,10 @@ def pre_oci_model(initialized_db):
(['latest', 'prod', 'another'], {'latest', 'prod'}),
(['foo', 'prod'], {'prod'}),
])
def test_find_matching_tag(names, expected, pre_oci_model):
def test_find_matching_tag(names, expected, registry_model):
repo = model.repository.get_repository('devtable', 'simple')
repository_ref = RepositoryReference.for_repo_obj(repo)
found = pre_oci_model.find_matching_tag(repository_ref, names)
found = registry_model.find_matching_tag(repository_ref, names)
if expected is None:
assert found is None
else:
@ -49,10 +50,10 @@ def test_find_matching_tag(names, expected, pre_oci_model):
('devtable', 'simple', {'latest', 'prod'}),
('buynlarge', 'orgrepo', {'latest', 'prod'}),
])
def test_get_most_recent_tag(repo_namespace, repo_name, expected, pre_oci_model):
def test_get_most_recent_tag(repo_namespace, repo_name, expected, registry_model):
repo = model.repository.get_repository(repo_namespace, repo_name)
repository_ref = RepositoryReference.for_repo_obj(repo)
found = pre_oci_model.get_most_recent_tag(repository_ref)
found = registry_model.get_most_recent_tag(repository_ref)
if expected is None:
assert found is None
else:
@ -64,8 +65,8 @@ def test_get_most_recent_tag(repo_namespace, repo_name, expected, pre_oci_model)
('buynlarge', 'orgrepo', True),
('buynlarge', 'unknownrepo', False),
])
def test_lookup_repository(repo_namespace, repo_name, expected, pre_oci_model):
repo_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
def test_lookup_repository(repo_namespace, repo_name, expected, registry_model):
repo_ref = registry_model.lookup_repository(repo_namespace, repo_name)
if expected:
assert repo_ref
else:
@ -76,22 +77,22 @@ def test_lookup_repository(repo_namespace, repo_name, expected, pre_oci_model):
('devtable', 'simple'),
('buynlarge', 'orgrepo'),
])
def test_lookup_manifests(repo_namespace, repo_name, pre_oci_model):
def test_lookup_manifests(repo_namespace, repo_name, registry_model):
repo = model.repository.get_repository(repo_namespace, repo_name)
repository_ref = RepositoryReference.for_repo_obj(repo)
found_tag = pre_oci_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = pre_oci_model.get_manifest_for_tag(found_tag)
found = pre_oci_model.lookup_manifest_by_digest(repository_ref, found_manifest.digest,
found_tag = registry_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = registry_model.get_manifest_for_tag(found_tag)
found = registry_model.lookup_manifest_by_digest(repository_ref, found_manifest.digest,
include_legacy_image=True)
assert found._db_id == found_manifest._db_id
assert found.digest == found_manifest.digest
assert found.legacy_image
def test_lookup_unknown_manifest(pre_oci_model):
def test_lookup_unknown_manifest(registry_model):
repo = model.repository.get_repository('devtable', 'simple')
repository_ref = RepositoryReference.for_repo_obj(repo)
found = pre_oci_model.lookup_manifest_by_digest(repository_ref, 'sha256:deadbeef')
found = registry_model.lookup_manifest_by_digest(repository_ref, 'sha256:deadbeef')
assert found is None
@ -101,18 +102,18 @@ def test_lookup_unknown_manifest(pre_oci_model):
('devtable', 'history'),
('buynlarge', 'orgrepo'),
])
def test_legacy_images(repo_namespace, repo_name, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
legacy_images = pre_oci_model.get_legacy_images(repository_ref)
def test_legacy_images(repo_namespace, repo_name, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
legacy_images = registry_model.get_legacy_images(repository_ref)
assert len(legacy_images)
found_tags = set()
for image in legacy_images:
found_image = pre_oci_model.get_legacy_image(repository_ref, image.docker_image_id,
found_image = registry_model.get_legacy_image(repository_ref, image.docker_image_id,
include_parents=True)
with assert_query_count(5 if found_image.parents else 4):
found_image = pre_oci_model.get_legacy_image(repository_ref, image.docker_image_id,
found_image = registry_model.get_legacy_image(repository_ref, image.docker_image_id,
include_parents=True, include_blob=True)
assert found_image.docker_image_id == image.docker_image_id
assert found_image.parents == image.parents
@ -130,76 +131,76 @@ def test_legacy_images(repo_namespace, repo_name, pre_oci_model):
[p._db_id for p in found_image.parents])
# Try without parents and ensure it raises an exception.
found_image = pre_oci_model.get_legacy_image(repository_ref, image.docker_image_id,
found_image = registry_model.get_legacy_image(repository_ref, image.docker_image_id,
include_parents=False)
with pytest.raises(Exception):
assert not found_image.parents
assert found_tags
unknown = pre_oci_model.get_legacy_image(repository_ref, 'unknown', include_parents=True)
unknown = registry_model.get_legacy_image(repository_ref, 'unknown', include_parents=True)
assert unknown is None
def test_manifest_labels(pre_oci_model):
def test_manifest_labels(registry_model):
repo = model.repository.get_repository('devtable', 'simple')
repository_ref = RepositoryReference.for_repo_obj(repo)
found_tag = pre_oci_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = pre_oci_model.get_manifest_for_tag(found_tag)
found_tag = registry_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = registry_model.get_manifest_for_tag(found_tag)
# Create a new label.
created = pre_oci_model.create_manifest_label(found_manifest, 'foo', 'bar', 'api')
created = registry_model.create_manifest_label(found_manifest, 'foo', 'bar', 'api')
assert created.key == 'foo'
assert created.value == 'bar'
assert created.source_type_name == 'api'
assert created.media_type_name == 'text/plain'
# Ensure we can look it up.
assert pre_oci_model.get_manifest_label(found_manifest, created.uuid) == created
assert registry_model.get_manifest_label(found_manifest, created.uuid) == created
# Ensure it is in our list of labels.
assert created in pre_oci_model.list_manifest_labels(found_manifest)
assert created in pre_oci_model.list_manifest_labels(found_manifest, key_prefix='fo')
assert created in registry_model.list_manifest_labels(found_manifest)
assert created in registry_model.list_manifest_labels(found_manifest, key_prefix='fo')
# Ensure it is *not* in our filtered list.
assert created not in pre_oci_model.list_manifest_labels(found_manifest, key_prefix='ba')
assert created not in registry_model.list_manifest_labels(found_manifest, key_prefix='ba')
# Delete the label and ensure it is gone.
assert pre_oci_model.delete_manifest_label(found_manifest, created.uuid)
assert pre_oci_model.get_manifest_label(found_manifest, created.uuid) is None
assert created not in pre_oci_model.list_manifest_labels(found_manifest)
assert registry_model.delete_manifest_label(found_manifest, created.uuid)
assert registry_model.get_manifest_label(found_manifest, created.uuid) is None
assert created not in registry_model.list_manifest_labels(found_manifest)
def test_manifest_label_handlers(pre_oci_model):
def test_manifest_label_handlers(registry_model):
repo = model.repository.get_repository('devtable', 'simple')
repository_ref = RepositoryReference.for_repo_obj(repo)
found_tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
found_manifest = pre_oci_model.get_manifest_for_tag(found_tag)
found_tag = registry_model.get_repo_tag(repository_ref, 'latest')
found_manifest = registry_model.get_manifest_for_tag(found_tag)
# Ensure the tag has no expiration.
assert found_tag.lifetime_end_ts is None
# Create a new label with an expires-after.
pre_oci_model.create_manifest_label(found_manifest, 'quay.expires-after', '2h', 'api')
registry_model.create_manifest_label(found_manifest, 'quay.expires-after', '2h', 'api')
# Ensure the tag now has an expiration.
updated_tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
updated_tag = registry_model.get_repo_tag(repository_ref, 'latest')
assert updated_tag.lifetime_end_ts == (updated_tag.lifetime_start_ts + (60 * 60 * 2))
def test_batch_labels(pre_oci_model):
def test_batch_labels(registry_model):
repo = model.repository.get_repository('devtable', 'history')
repository_ref = RepositoryReference.for_repo_obj(repo)
found_tag = pre_oci_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = pre_oci_model.get_manifest_for_tag(found_tag)
found_tag = registry_model.find_matching_tag(repository_ref, ['latest'])
found_manifest = registry_model.get_manifest_for_tag(found_tag)
with pre_oci_model.batch_create_manifest_labels(found_manifest) as add_label:
with registry_model.batch_create_manifest_labels(found_manifest) as add_label:
add_label('foo', '1', 'api')
add_label('bar', '2', 'api')
add_label('baz', '3', 'api')
# Ensure we can look them up.
assert len(pre_oci_model.list_manifest_labels(found_manifest)) == 3
assert len(registry_model.list_manifest_labels(found_manifest)) == 3
@pytest.mark.parametrize('repo_namespace, repo_name', [
@ -208,38 +209,38 @@ def test_batch_labels(pre_oci_model):
('devtable', 'history'),
('buynlarge', 'orgrepo'),
])
def test_repository_tags(repo_namespace, repo_name, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
def test_repository_tags(repo_namespace, repo_name, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
with assert_query_count(1):
tags = pre_oci_model.list_repository_tags(repository_ref, include_legacy_images=True)
tags = registry_model.list_repository_tags(repository_ref, include_legacy_images=True)
assert len(tags)
for tag in tags:
with assert_query_count(2):
found_tag = pre_oci_model.get_repo_tag(repository_ref, tag.name, include_legacy_image=True)
found_tag = registry_model.get_repo_tag(repository_ref, tag.name, include_legacy_image=True)
assert found_tag == tag
if found_tag.legacy_image is None:
continue
with assert_query_count(2):
found_image = pre_oci_model.get_legacy_image(repository_ref,
found_image = registry_model.get_legacy_image(repository_ref,
found_tag.legacy_image.docker_image_id)
assert found_image == found_tag.legacy_image
def test_repository_tag_history(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'history')
def test_repository_tag_history(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'history')
with assert_query_count(2):
history, has_more = pre_oci_model.list_repository_tag_history(repository_ref)
history, has_more = registry_model.list_repository_tag_history(repository_ref)
assert not has_more
assert len(history) == 2
# Ensure the latest tag is marked expired, since there is an expired one.
with assert_query_count(1):
assert pre_oci_model.has_expired_tag(repository_ref, 'latest')
assert registry_model.has_expired_tag(repository_ref, 'latest')
@pytest.mark.parametrize('repo_namespace, repo_name', [
@ -252,35 +253,35 @@ def test_repository_tag_history(pre_oci_model):
False,
True,
])
def test_delete_tags(repo_namespace, repo_name, via_manifest, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
tags = pre_oci_model.list_repository_tags(repository_ref)
def test_delete_tags(repo_namespace, repo_name, via_manifest, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
tags = registry_model.list_repository_tags(repository_ref)
assert len(tags)
# Save history before the deletions.
previous_history, _ = pre_oci_model.list_repository_tag_history(repository_ref, size=1000)
previous_history, _ = registry_model.list_repository_tag_history(repository_ref, size=1000)
assert len(previous_history) >= len(tags)
# Delete every tag in the repository.
for tag in tags:
if via_manifest:
assert pre_oci_model.delete_tag(repository_ref, tag.name)
assert registry_model.delete_tag(repository_ref, tag.name)
else:
manifest = pre_oci_model.get_manifest_for_tag(tag)
manifest = registry_model.get_manifest_for_tag(tag)
if manifest is not None:
assert pre_oci_model.delete_tags_for_manifest(manifest)
assert registry_model.delete_tags_for_manifest(manifest)
# Make sure the tag is no longer found.
with assert_query_count(1):
found_tag = pre_oci_model.get_repo_tag(repository_ref, tag.name, include_legacy_image=True)
found_tag = registry_model.get_repo_tag(repository_ref, tag.name, include_legacy_image=True)
assert found_tag is None
# Ensure all tags have been deleted.
tags = pre_oci_model.list_repository_tags(repository_ref)
tags = registry_model.list_repository_tags(repository_ref)
assert not len(tags)
# Ensure that the tags all live in history.
history, _ = pre_oci_model.list_repository_tag_history(repository_ref, size=1000)
history, _ = registry_model.list_repository_tag_history(repository_ref, size=1000)
assert len(history) == len(previous_history)
@ -288,12 +289,12 @@ def test_delete_tags(repo_namespace, repo_name, via_manifest, pre_oci_model):
True,
False,
])
def test_retarget_tag_history(use_manifest, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'history')
history, _ = pre_oci_model.list_repository_tag_history(repository_ref)
def test_retarget_tag_history(use_manifest, registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'history')
history, _ = registry_model.list_repository_tag_history(repository_ref)
if use_manifest:
manifest_or_legacy_image = pre_oci_model.lookup_manifest_by_digest(repository_ref,
manifest_or_legacy_image = registry_model.lookup_manifest_by_digest(repository_ref,
history[1].manifest_digest,
allow_dead=True)
else:
@ -301,7 +302,7 @@ def test_retarget_tag_history(use_manifest, pre_oci_model):
# Retarget the tag.
assert manifest_or_legacy_image
updated_tag = pre_oci_model.retarget_tag(repository_ref, 'latest', manifest_or_legacy_image,
updated_tag = registry_model.retarget_tag(repository_ref, 'latest', manifest_or_legacy_image,
is_reversion=True)
# Ensure the tag has changed targets.
@ -311,39 +312,39 @@ def test_retarget_tag_history(use_manifest, pre_oci_model):
assert updated_tag.legacy_image == manifest_or_legacy_image
# Ensure history has been updated.
new_history, _ = pre_oci_model.list_repository_tag_history(repository_ref)
new_history, _ = registry_model.list_repository_tag_history(repository_ref)
assert len(new_history) == len(history) + 1
def test_retarget_tag(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'complex')
history, _ = pre_oci_model.list_repository_tag_history(repository_ref)
def test_retarget_tag(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'complex')
history, _ = registry_model.list_repository_tag_history(repository_ref)
prod_tag = pre_oci_model.get_repo_tag(repository_ref, 'prod', include_legacy_image=True)
prod_tag = registry_model.get_repo_tag(repository_ref, 'prod', include_legacy_image=True)
# Retarget the tag.
updated_tag = pre_oci_model.retarget_tag(repository_ref, 'latest', prod_tag.legacy_image)
updated_tag = registry_model.retarget_tag(repository_ref, 'latest', prod_tag.legacy_image)
# Ensure the tag has changed targets.
assert updated_tag.legacy_image == prod_tag.legacy_image
# Ensure history has been updated.
new_history, _ = pre_oci_model.list_repository_tag_history(repository_ref)
new_history, _ = registry_model.list_repository_tag_history(repository_ref)
assert len(new_history) == len(history) + 1
def test_change_repository_tag_expiration(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
def test_change_repository_tag_expiration(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
tag = registry_model.get_repo_tag(repository_ref, 'latest')
assert tag.lifetime_end_ts is None
new_datetime = datetime.utcnow() + timedelta(days=2)
previous, okay = pre_oci_model.change_repository_tag_expiration(tag, new_datetime)
previous, okay = registry_model.change_repository_tag_expiration(tag, new_datetime)
assert okay
assert previous is None
tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
tag = registry_model.get_repo_tag(repository_ref, 'latest')
assert tag.lifetime_end_ts is not None
@ -355,26 +356,26 @@ def test_change_repository_tag_expiration(pre_oci_model):
('devtable', 'gargantuan', ['v2.0', 'v3.0', 'v4.0', 'v5.0', 'v6.0']),
])
def test_get_legacy_images_owned_by_tag(repo_namespace, repo_name, expected_non_empty,
pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
tags = pre_oci_model.list_repository_tags(repository_ref)
registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
tags = registry_model.list_repository_tags(repository_ref)
assert len(tags)
non_empty = set()
for tag in tags:
if pre_oci_model.get_legacy_images_owned_by_tag(tag):
if registry_model.get_legacy_images_owned_by_tag(tag):
non_empty.add(tag.name)
assert non_empty == set(expected_non_empty)
def test_get_security_status(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
tags = pre_oci_model.list_repository_tags(repository_ref, include_legacy_images=True)
def test_get_security_status(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
tags = registry_model.list_repository_tags(repository_ref, include_legacy_images=True)
assert len(tags)
for tag in tags:
assert pre_oci_model.get_security_status(tag.legacy_image)
assert registry_model.get_security_status(tag.legacy_image)
@pytest.fixture()
@ -396,24 +397,24 @@ def clear_rows(initialized_db):
('devtable', 'history'),
('buynlarge', 'orgrepo'),
])
def test_backfill_manifest_for_tag(repo_namespace, repo_name, clear_rows, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
tags = pre_oci_model.list_repository_tags(repository_ref)
def test_backfill_manifest_for_tag(repo_namespace, repo_name, clear_rows, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
tags = registry_model.list_repository_tags(repository_ref)
assert tags
for tag in tags:
assert not tag.manifest_digest
assert pre_oci_model.backfill_manifest_for_tag(tag)
assert registry_model.backfill_manifest_for_tag(tag)
tags = pre_oci_model.list_repository_tags(repository_ref, include_legacy_images=True)
tags = registry_model.list_repository_tags(repository_ref, include_legacy_images=True)
assert tags
for tag in tags:
assert tag.manifest_digest
manifest = pre_oci_model.get_manifest_for_tag(tag)
manifest = registry_model.get_manifest_for_tag(tag)
assert manifest
legacy_image = pre_oci_model.get_legacy_image(repository_ref, tag.legacy_image.docker_image_id,
legacy_image = registry_model.get_legacy_image(repository_ref, tag.legacy_image.docker_image_id,
include_parents=True)
parsed_manifest = manifest.get_parsed_manifest()
@ -427,19 +428,19 @@ def test_backfill_manifest_for_tag(repo_namespace, repo_name, clear_rows, pre_oc
('devtable', 'history'),
('buynlarge', 'orgrepo'),
])
def test_backfill_manifest_on_lookup(repo_namespace, repo_name, clear_rows, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
tags = pre_oci_model.list_repository_tags(repository_ref)
def test_backfill_manifest_on_lookup(repo_namespace, repo_name, clear_rows, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
tags = registry_model.list_repository_tags(repository_ref)
assert tags
for tag in tags:
assert not tag.manifest_digest
assert not pre_oci_model.get_manifest_for_tag(tag)
assert not registry_model.get_manifest_for_tag(tag)
manifest = pre_oci_model.get_manifest_for_tag(tag, backfill_if_necessary=True)
manifest = registry_model.get_manifest_for_tag(tag, backfill_if_necessary=True)
assert manifest
updated_tag = pre_oci_model.get_repo_tag(repository_ref, tag.name)
updated_tag = registry_model.get_repo_tag(repository_ref, tag.name)
assert updated_tag.manifest_digest == manifest.digest
@ -449,8 +450,8 @@ def test_backfill_manifest_on_lookup(repo_namespace, repo_name, clear_rows, pre_
('disabled', False),
])
def test_is_namespace_enabled(namespace, expect_enabled, pre_oci_model):
assert pre_oci_model.is_namespace_enabled(namespace) == expect_enabled
def test_is_namespace_enabled(namespace, expect_enabled, registry_model):
assert registry_model.is_namespace_enabled(namespace) == expect_enabled
@pytest.mark.parametrize('repo_namespace, repo_name', [
@ -459,20 +460,20 @@ def test_is_namespace_enabled(namespace, expect_enabled, pre_oci_model):
('devtable', 'history'),
('buynlarge', 'orgrepo'),
])
def test_list_manifest_layers(repo_namespace, repo_name, pre_oci_model):
repository_ref = pre_oci_model.lookup_repository(repo_namespace, repo_name)
tags = pre_oci_model.list_repository_tags(repository_ref)
def test_list_manifest_layers(repo_namespace, repo_name, registry_model):
repository_ref = registry_model.lookup_repository(repo_namespace, repo_name)
tags = registry_model.list_repository_tags(repository_ref)
assert tags
for tag in tags:
manifest = pre_oci_model.get_manifest_for_tag(tag)
manifest = registry_model.get_manifest_for_tag(tag)
assert manifest
with assert_query_count(4):
layers = pre_oci_model.list_manifest_layers(manifest)
layers = registry_model.list_manifest_layers(manifest)
assert layers
layers = pre_oci_model.list_manifest_layers(manifest, include_placements=True)
layers = registry_model.list_manifest_layers(manifest, include_placements=True)
assert layers
parsed_layers = list(manifest.get_parsed_manifest().layers)
@ -484,88 +485,88 @@ def test_list_manifest_layers(repo_namespace, repo_name, pre_oci_model):
assert manifest_layer.blob.storage_path
assert manifest_layer.blob.placements
repo_blob = pre_oci_model.get_repo_blob_by_digest(repository_ref, manifest_layer.blob.digest)
repo_blob = registry_model.get_repo_blob_by_digest(repository_ref, manifest_layer.blob.digest)
assert repo_blob.digest == manifest_layer.blob.digest
assert manifest_layer.estimated_size(1) is not None
def test_derived_image(pre_oci_model):
def test_derived_image(registry_model):
# Clear all existing derived storage.
DerivedStorageForImage.delete().execute()
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
manifest = pre_oci_model.get_manifest_for_tag(tag)
repository_ref = registry_model.lookup_repository('devtable', 'simple')
tag = registry_model.get_repo_tag(repository_ref, 'latest')
manifest = registry_model.get_manifest_for_tag(tag)
# Ensure the squashed image doesn't exist.
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {}) is None
assert registry_model.lookup_derived_image(manifest, 'squash', {}) is None
# Create a new one.
squashed = pre_oci_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {})
assert pre_oci_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {}) == squashed
squashed = registry_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {})
assert registry_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {}) == squashed
assert squashed.unique_id
# Check and set the size.
assert squashed.blob.compressed_size is None
pre_oci_model.set_derived_image_size(squashed, 1234)
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {}).blob.compressed_size == 1234
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {}).unique_id == squashed.unique_id
registry_model.set_derived_image_size(squashed, 1234)
assert registry_model.lookup_derived_image(manifest, 'squash', {}).blob.compressed_size == 1234
assert registry_model.lookup_derived_image(manifest, 'squash', {}).unique_id == squashed.unique_id
# Ensure its returned now.
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {}) == squashed
assert registry_model.lookup_derived_image(manifest, 'squash', {}) == squashed
# Ensure different metadata results in a different derived image.
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {'foo': 'bar'}) is None
assert registry_model.lookup_derived_image(manifest, 'squash', {'foo': 'bar'}) is None
squashed_foo = pre_oci_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us',
squashed_foo = registry_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us',
{'foo': 'bar'})
assert squashed_foo != squashed
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {'foo': 'bar'}) == squashed_foo
assert registry_model.lookup_derived_image(manifest, 'squash', {'foo': 'bar'}) == squashed_foo
assert squashed.unique_id != squashed_foo.unique_id
# Lookup with placements.
squashed = pre_oci_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {},
squashed = registry_model.lookup_or_create_derived_image(manifest, 'squash', 'local_us', {},
include_placements=True)
assert squashed.blob.placements
# Delete the derived image.
pre_oci_model.delete_derived_image(squashed)
assert pre_oci_model.lookup_derived_image(manifest, 'squash', {}) is None
registry_model.delete_derived_image(squashed)
assert registry_model.lookup_derived_image(manifest, 'squash', {}) is None
def test_derived_image_signatures(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
manifest = pre_oci_model.get_manifest_for_tag(tag)
def test_derived_image_signatures(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
tag = registry_model.get_repo_tag(repository_ref, 'latest')
manifest = registry_model.get_manifest_for_tag(tag)
derived = pre_oci_model.lookup_derived_image(manifest, 'squash', {})
derived = registry_model.lookup_derived_image(manifest, 'squash', {})
assert derived
signature = pre_oci_model.get_derived_image_signature(derived, 'gpg2')
signature = registry_model.get_derived_image_signature(derived, 'gpg2')
assert signature is None
pre_oci_model.set_derived_image_signature(derived, 'gpg2', 'foo')
assert pre_oci_model.get_derived_image_signature(derived, 'gpg2') == 'foo'
registry_model.set_derived_image_signature(derived, 'gpg2', 'foo')
assert registry_model.get_derived_image_signature(derived, 'gpg2') == 'foo'
def test_torrent_info(pre_oci_model):
def test_torrent_info(registry_model):
# Remove all existing info.
TorrentInfo.delete().execute()
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
manifest = pre_oci_model.get_manifest_for_tag(tag)
repository_ref = registry_model.lookup_repository('devtable', 'simple')
tag = registry_model.get_repo_tag(repository_ref, 'latest')
manifest = registry_model.get_manifest_for_tag(tag)
layers = pre_oci_model.list_manifest_layers(manifest)
layers = registry_model.list_manifest_layers(manifest)
assert layers
assert pre_oci_model.get_torrent_info(layers[0].blob) is None
pre_oci_model.set_torrent_info(layers[0].blob, 2, 'foo')
assert registry_model.get_torrent_info(layers[0].blob) is None
registry_model.set_torrent_info(layers[0].blob, 2, 'foo')
# Set it again exactly, which should be a no-op.
pre_oci_model.set_torrent_info(layers[0].blob, 2, 'foo')
registry_model.set_torrent_info(layers[0].blob, 2, 'foo')
# Check the information we've set.
torrent_info = pre_oci_model.get_torrent_info(layers[0].blob)
@ -574,33 +575,33 @@ def test_torrent_info(pre_oci_model):
assert torrent_info.pieces == 'foo'
# Try setting it again. Nothing should happen.
pre_oci_model.set_torrent_info(layers[0].blob, 3, 'bar')
registry_model.set_torrent_info(layers[0].blob, 3, 'bar')
torrent_info = pre_oci_model.get_torrent_info(layers[0].blob)
torrent_info = registry_model.get_torrent_info(layers[0].blob)
assert torrent_info is not None
assert torrent_info.piece_length == 2
assert torrent_info.pieces == 'foo'
def test_blob_uploads(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
def test_blob_uploads(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
blob_upload = pre_oci_model.create_blob_upload(repository_ref, str(uuid.uuid4()),
blob_upload = registry_model.create_blob_upload(repository_ref, str(uuid.uuid4()),
'local_us', {'some': 'metadata'})
assert blob_upload
assert blob_upload.storage_metadata == {'some': 'metadata'}
assert blob_upload.location_name == 'local_us'
# Ensure we can find the blob upload.
assert pre_oci_model.lookup_blob_upload(repository_ref, blob_upload.upload_id) == blob_upload
assert registry_model.lookup_blob_upload(repository_ref, blob_upload.upload_id) == blob_upload
# Update and ensure the changes are saved.
assert pre_oci_model.update_blob_upload(blob_upload, 1, 'the-pieces_hash',
assert registry_model.update_blob_upload(blob_upload, 1, 'the-pieces_hash',
blob_upload.piece_sha_state,
{'new': 'metadata'}, 2, 3,
blob_upload.sha_state)
updated = pre_oci_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
updated = registry_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
assert updated
assert updated.uncompressed_byte_count == 1
assert updated.piece_hashes == 'the-pieces_hash'
@ -609,45 +610,45 @@ def test_blob_uploads(pre_oci_model):
assert updated.chunk_count == 3
# Delete the upload.
pre_oci_model.delete_blob_upload(blob_upload)
registry_model.delete_blob_upload(blob_upload)
# Ensure it can no longer be found.
assert not pre_oci_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
assert not registry_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
def test_commit_blob_upload(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
blob_upload = pre_oci_model.create_blob_upload(repository_ref, str(uuid.uuid4()),
def test_commit_blob_upload(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
blob_upload = registry_model.create_blob_upload(repository_ref, str(uuid.uuid4()),
'local_us', {'some': 'metadata'})
# Commit the blob upload and make sure it is written as a blob.
digest = 'sha256:' + hashlib.sha256('hello').hexdigest()
blob = pre_oci_model.commit_blob_upload(blob_upload, digest, 60)
blob = registry_model.commit_blob_upload(blob_upload, digest, 60)
assert blob.digest == digest
# Ensure the upload can no longer be found.
assert not pre_oci_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
assert not registry_model.lookup_blob_upload(repository_ref, blob_upload.upload_id)
def test_mount_blob_into_repository(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
latest_tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
manifest = pre_oci_model.get_manifest_for_tag(latest_tag)
def test_mount_blob_into_repository(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
latest_tag = registry_model.get_repo_tag(repository_ref, 'latest')
manifest = registry_model.get_manifest_for_tag(latest_tag)
target_repository_ref = pre_oci_model.lookup_repository('devtable', 'complex')
target_repository_ref = registry_model.lookup_repository('devtable', 'complex')
layers = pre_oci_model.list_manifest_layers(manifest, include_placements=True)
layers = registry_model.list_manifest_layers(manifest, include_placements=True)
assert layers
for layer in layers:
# Ensure the blob doesn't exist under the repository.
assert not pre_oci_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest)
assert not registry_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest)
# Mount the blob into the repository.
assert pre_oci_model.mount_blob_into_repository(layer.blob, target_repository_ref, 60)
assert registry_model.mount_blob_into_repository(layer.blob, target_repository_ref, 60)
# Ensure it now exists.
found = pre_oci_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest)
found = registry_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest)
assert found == layer.blob
@ -655,20 +656,20 @@ class SomeException(Exception):
pass
def test_get_cached_repo_blob(pre_oci_model):
def test_get_cached_repo_blob(registry_model):
model_cache = InMemoryDataModelCache()
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
latest_tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
manifest = pre_oci_model.get_manifest_for_tag(latest_tag)
repository_ref = registry_model.lookup_repository('devtable', 'simple')
latest_tag = registry_model.get_repo_tag(repository_ref, 'latest')
manifest = registry_model.get_manifest_for_tag(latest_tag)
layers = pre_oci_model.list_manifest_layers(manifest, include_placements=True)
layers = registry_model.list_manifest_layers(manifest, include_placements=True)
assert layers
blob = layers[0].blob
# Load a blob to add it to the cache.
found = pre_oci_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', blob.digest)
found = registry_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', blob.digest)
assert found.digest == blob.digest
assert found.uuid == blob.uuid
assert found.compressed_size == blob.compressed_size
@ -683,7 +684,7 @@ def test_get_cached_repo_blob(pre_oci_model):
with patch('data.registry_model.registry_pre_oci_model.model.blob.get_repository_blob_by_digest',
fail):
# Make sure we can load again, which should hit the cache.
cached = pre_oci_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', blob.digest)
cached = registry_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', blob.digest)
assert cached.digest == blob.digest
assert cached.uuid == blob.uuid
assert cached.compressed_size == blob.compressed_size
@ -694,13 +695,13 @@ def test_get_cached_repo_blob(pre_oci_model):
# Try another blob, which should fail since the DB is not connected and the cache
# does not contain the blob.
with pytest.raises(SomeException):
pre_oci_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', 'some other digest')
registry_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', 'some other digest')
def test_create_manifest_and_retarget_tag(pre_oci_model):
repository_ref = pre_oci_model.lookup_repository('devtable', 'simple')
latest_tag = pre_oci_model.get_repo_tag(repository_ref, 'latest', include_legacy_image=True)
manifest = pre_oci_model.get_manifest_for_tag(latest_tag).get_parsed_manifest()
def test_create_manifest_and_retarget_tag(registry_model):
repository_ref = registry_model.lookup_repository('devtable', 'simple')
latest_tag = registry_model.get_repo_tag(repository_ref, 'latest', include_legacy_image=True)
manifest = registry_model.get_manifest_for_tag(latest_tag).get_parsed_manifest()
builder = DockerSchema1ManifestBuilder('devtable', 'simple', 'anothertag')
builder.add_layer(manifest.blob_digests[0],
@ -708,7 +709,7 @@ def test_create_manifest_and_retarget_tag(pre_oci_model):
sample_manifest = builder.build(docker_v2_signing_key)
assert sample_manifest is not None
another_manifest, tag = pre_oci_model.create_manifest_and_retarget_tag(repository_ref,
another_manifest, tag = registry_model.create_manifest_and_retarget_tag(repository_ref,
sample_manifest,
'anothertag')
assert another_manifest is not None
@ -717,5 +718,5 @@ def test_create_manifest_and_retarget_tag(pre_oci_model):
assert tag.name == 'anothertag'
assert another_manifest.get_parsed_manifest().manifest_dict == sample_manifest.manifest_dict
layers = pre_oci_model.list_manifest_layers(another_manifest)
layers = registry_model.list_manifest_layers(another_manifest)
assert len(layers) == 1