Add managed-access integration test for verbatim manifests (#6098)

test/integration_test.py

@@ -126,6 +126,9 @@
     IndexExistsAndDiffersException,
     IndexService,
 )
+from azul.json_freeze import (
+    freeze,
+)
 from azul.logging import (
     configure_test_logging,
     get_test_logger,
@@ -1592,6 +1595,52 @@ def test_compact_manifest(expected_bundles):
         # Without credentials, only the public bundle should be represented
         test_compact_manifest({public_bundle})
 
+        def read_verbatim_jsonl_manifest(manifest: IO) -> set[JSON]:
+            manifest_lines = manifest.readlines()
+            manifest_content = {
+                freeze(json.loads(replica))
+                for replica in manifest_lines
+            }
+            self.assertEqual(len(manifest_lines), len(manifest_content))
+            return manifest_content
+
+        def read_verbatim_pfb_manifest(manifest: IO) -> set[str]:
+            entities = list(fastavro.reader(manifest))
+            manifest_content = {
+                # We can't assert the full contents of each entity because the
+                # schema changes depending on the filters used.
+                entity['id']
+                for entity in entities
+                # The special "Metadata" entity is always present. Dropping it
+                # from the result streamlines the set logic used in the
+                # assertion below.
+                if entity['name'] != 'Metadata'
+            }
+            return manifest_content
+
+        def get_verbatim_manifest(format: ManifestFormat,
+                                  bundles: Iterable[str],
+                                  ) -> set:
+            manifest_url = furl(url=endpoint, path='/manifest/files', args={
+                'catalog': catalog,
+                'format': format.value,
+                'filters': json.dumps({special_fields.bundle_uuid: {'is': list(bundles)}})
+            })
+            content = BytesIO(self._get_url_content(PUT, manifest_url))
+            return {
+                ManifestFormat.verbatim_jsonl: read_verbatim_jsonl_manifest,
+                ManifestFormat.verbatim_pfb: read_verbatim_pfb_manifest
+            }[format](content)
+
+        for format in ManifestFormat.verbatim_jsonl, ManifestFormat.verbatim_pfb:
+            if format in metadata_plugin.manifest_formats:
+                with self.subTest(format=format):
+                    unauthorized = get_verbatim_manifest(format, all_bundles)
+                    with self._service_account_credentials:
+                        authorized = get_verbatim_manifest(format, all_bundles)
+                        private_only = get_verbatim_manifest(format, managed_access_bundles)
+                    self.assertSetEqual(private_only, authorized - unauthorized)
+
         if ManifestFormat.curl in metadata_plugin.manifest_formats:
             # Create a single-file curl manifest and verify that the OAuth2
             # token is present on the command line