quay/data/registry_model/test/test_pre_oci_model.py

import pytest

from data import model
from data.registry_model.registry_pre_oci_model import PreOCIModel
from data.registry_model.datatypes import RepositoryReference
from test.fixtures import *

@pytest.fixture()
def pre_oci_model(initialized_db):
  return PreOCIModel()


@pytest.mark.parametrize('names, expected', [
  (['unknown'], None),
  (['latest'], {'latest'}),
  (['latest', 'prod'], {'latest', 'prod'}),
  (['latest', 'prod', 'another'], {'latest', 'prod'}),
  (['foo', 'prod'], {'prod'}),
])
def test_find_matching_tag(names, expected, pre_oci_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)
  if expected is None:
    assert found is None
  else:
    assert found.name in expected


@pytest.mark.parametrize('repo_namespace, repo_name, expected', [
  ('devtable', 'simple', {'latest'}),
  ('buynlarge', 'orgrepo', {'latest', 'prod'}),
])
def test_get_most_recent_tag(repo_namespace, repo_name, expected, pre_oci_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)
  if expected is None:
    assert found is None
  else:
    assert found.name in expected


@pytest.mark.parametrize('repo_namespace, repo_name, expected', [
  ('devtable', 'simple', True),
  ('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)
  if expected:
    assert repo_ref
  else:
    assert repo_ref is None


@pytest.mark.parametrize('repo_namespace, repo_name', [
  ('devtable', 'simple'),
  ('buynlarge', 'orgrepo'),
])
def test_lookup_manifests(repo_namespace, repo_name, pre_oci_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)
  assert found._db_id == found_manifest._db_id
  assert found.digest == found_manifest.digest


def test_create_manifest_label(pre_oci_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)

  pre_oci_model.create_manifest_label(found_manifest, 'foo', 'bar', 'internal')


@pytest.mark.parametrize('repo_namespace, repo_name', [
  ('devtable', 'simple'),
  ('devtable', 'complex'),
  ('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)
  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,
                                                 include_parents=True)

    assert found_image.docker_image_id == image.docker_image_id
    assert found_image.parents == image.parents

    # Check that the tags list can be retrieved.
    assert image.tags is not None
    found_tags.update({tag.name for tag in image.tags})

    # Check against the actual DB row.
    model_image = model.image.get_image(repository_ref._db_id, found_image.docker_image_id)
    assert model_image.id == found_image._db_id
    assert ([pid for pid in model_image.ancestor_id_list()] ==
            [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,
                                                 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)
  assert unknown is None