bug: Adding allow hidden flag while looking up for manifests (PROJQUAY-8536) (#3722)

When an image is pulled by digest, a temp tag is created to prevent the manifest from being garbage collected. This is true when a manifest list is pulled by tag as well. However, if this temporary tag expires (default is 1 day for proxied organizations) and the same manifest is pulled again by digest, the system attempts to create the manifest again, leading to an integrity error because the manifest already exists in the database.

---------

Co-authored-by: shudeshp <shudeshp@redhat.com>

data/registry_model/registry_proxy_model.py

@@ -319,6 +319,7 @@ class ProxyModel(OCIModel):
             repository_ref,
             manifest_digest,
             allow_dead=True,
+            allow_hidden=True,
             require_available=False,
             raise_on_error=True,
         )

data/registry_model/test/test_registry_proxy_model.py

@@ -1,5 +1,5 @@
 import json
-from datetime import timedelta
+from datetime import datetime, timedelta
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -471,6 +471,7 @@ class TestRegistryProxyModelLookupManifestByDigest:
     orgname = "quayio-cache"
     repository = f"{orgname}/{upstream_repository}"
     digest = UBI8_8_4_DIGEST
+    tag = "8.4"
 
     @pytest.fixture(autouse=True)
     def setup(self, app):
@@ -545,6 +546,40 @@ class TestRegistryProxyModelLookupManifestByDigest:
         tag = oci.tag.get_tag_by_manifest_id(repo_ref.id, manifest.id)
         assert tag.lifetime_end_ms > first_tag.lifetime_end_ms
 
+    @patch("data.registry_model.registry_proxy_model.Proxy", MagicMock())
+    def test_returns_existing_manifest_when_temp_tag_expires(self, create_repo):
+        repo_ref = create_repo(self.orgname, self.upstream_repository, self.user)
+        input_manifest = parse_manifest_from_bytes(
+            Bytes.for_string_or_unicode(UBI8_8_4_MANIFEST_SCHEMA2),
+            DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE,
+        )
+        proxy_model = ProxyModel(
+            self.orgname,
+            self.upstream_repository,
+            self.user,
+        )
+
+        proxy_model._proxy.get_manifest = MagicMock(
+            return_value=(UBI8_8_4_MANIFEST_SCHEMA2, DOCKER_SCHEMA2_MANIFEST_CONTENT_TYPE)
+        )
+
+        manifest, tag = proxy_model._create_manifest_and_retarget_tag(
+            repo_ref, input_manifest, self.tag
+        )
+        assert manifest is not None
+        assert tag is not None
+
+        expired_time = datetime.now() - timedelta(days=2)
+        tag.lifetime_end_ms = int(expired_time.timestamp() * 1000)
+        with patch.object(proxy_model._proxy, "manifest_exists", return_value=UBI8_8_4_DIGEST):
+            retrieved_manifest = proxy_model.lookup_manifest_by_digest(
+                repo_ref, UBI8_8_4_DIGEST, allow_hidden=True
+            )
+
+        assert retrieved_manifest is not None
+        assert retrieved_manifest.digest == UBI8_8_4_DIGEST
+        assert retrieved_manifest.internal_manifest_bytes.as_unicode() == UBI8_8_4_MANIFEST_SCHEMA2
+
     def test_recreate_tag_when_manifest_is_cached_and_exists_upstream(
         self, create_repo, proxy_manifest_response
     ):