Implement tag functions in new registry model interface

data/registry_model/test/test_pre_oci_model.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timedelta
+
 import pytest
 
 from data import model
@@ -103,7 +105,7 @@ def test_legacy_images(repo_namespace, repo_name, pre_oci_model):
     # 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()] ==
+    assert ([pid for pid in reversed(model_image.ancestor_id_list())] ==
             [p._db_id for p in found_image.parents])
 
     # Try without parents and ensure it raises an exception.
@@ -145,3 +147,149 @@ def test_manifest_labels(pre_oci_model):
   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)
+
+
+@pytest.mark.parametrize('repo_namespace, repo_name', [
+  ('devtable', 'simple'),
+  ('devtable', 'complex'),
+  ('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)
+  tags = pre_oci_model.list_repository_tags(repository_ref)
+  assert len(tags)
+
+  for tag in tags:
+    found_tag = pre_oci_model.get_repo_tag(repository_ref, tag.name, include_legacy_image=True)
+    assert found_tag == tag
+
+    if found_tag.legacy_image is None:
+      continue
+
+    found_image = pre_oci_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')
+  history, has_more = pre_oci_model.list_repository_tag_history(repository_ref)
+  assert not has_more
+  assert len(history) == 2
+
+
+@pytest.mark.parametrize('repo_namespace, repo_name', [
+  ('devtable', 'simple'),
+  ('devtable', 'complex'),
+  ('devtable', 'history'),
+  ('buynlarge', 'orgrepo'),
+])
+def test_delete_tags(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)
+  assert len(tags)
+
+  # Save history before the deletions.
+  previous_history, _ = pre_oci_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:
+    assert pre_oci_model.delete_tag(repository_ref, tag.name)
+
+    # Make sure the tag is no longer found.
+    found_tag = pre_oci_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)
+  assert not len(tags)
+
+  # Ensure that the tags all live in history.
+  history, _ = pre_oci_model.list_repository_tag_history(repository_ref, size=1000)
+  assert len(history) == len(previous_history)
+
+
+@pytest.mark.parametrize('use_manifest', [
+  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)
+
+  if use_manifest:
+    manifest_or_legacy_image = pre_oci_model.lookup_manifest_by_digest(repository_ref,
+                                                                       history[1].manifest_digest,
+                                                                       allow_dead=True)
+  else:
+    manifest_or_legacy_image = history[1].legacy_image
+
+  # Retarget the tag.
+  assert manifest_or_legacy_image
+  updated_tag = pre_oci_model.retarget_tag(repository_ref, 'latest', manifest_or_legacy_image,
+                                           is_reversion=True)
+
+  # Ensure the tag has changed targets.
+  if use_manifest:
+    assert updated_tag.manifest_digest == manifest_or_legacy_image.digest
+  else:
+    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)
+  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)
+
+  prod_tag = pre_oci_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)
+
+  # 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)
+  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')
+  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)
+
+  assert okay
+  assert previous is None
+
+  tag = pre_oci_model.get_repo_tag(repository_ref, 'latest')
+  assert tag.lifetime_end_ts is not None
+
+
+@pytest.mark.parametrize('repo_namespace, repo_name, expected_non_empty', [
+  ('devtable', 'simple', []),
+  ('devtable', 'complex', ['prod', 'v2.0']),
+  ('devtable', 'history', ['latest']),
+  ('buynlarge', 'orgrepo', []),
+  ('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)
+  assert len(tags)
+
+  non_empty = set()
+  for tag in tags:
+    if pre_oci_model.get_legacy_images_owned_by_tag(tag):
+      non_empty.add(tag.name)
+
+  assert non_empty == set(expected_non_empty)