diff --git a/data/registry_model/test/test_pre_oci_model.py b/data/registry_model/test/test_interface.py similarity index 54% rename from data/registry_model/test/test_pre_oci_model.py rename to data/registry_model/test/test_interface.py index addadf6cc..695ecc3ec 100644 --- a/data/registry_model/test/test_pre_oci_model.py +++ b/data/registry_model/test/test_interface.py @@ -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