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

test/integration_test.py

@@ -1547,7 +1547,7 @@ def bundle_uuids(hit: JSON) -> set[str]:
         params = {'catalog': catalog, 'filters': json.dumps(filters)}
         manifest_url = furl(url=endpoint, path='/manifest/files', args=params)
 
-        def assert_manifest(expected_bundles):
+        def test_compact_manifest(expected_bundles):
             manifest = BytesIO(self._get_url_content(PUT, manifest_url))
             manifest_rows = self._read_csv_manifest(manifest)
             all_found_bundles = set()
@@ -1563,13 +1563,37 @@ def assert_manifest(expected_bundles):
                 all_found_bundles.update(found_bundles)
             self.assertEqual(expected_bundles, all_found_bundles)
 
+        all_bundles = {public_bundle, *managed_access_bundles}
         # With authorized credentials, all bundles included in the filters
         # should be represented in the manifest
         with self._service_account_credentials:
-            assert_manifest({public_bundle, *managed_access_bundles})
+            test_compact_manifest(all_bundles)
 
         # Without credentials, only the public bundle should be represented
-        assert_manifest({public_bundle})
+        test_compact_manifest({public_bundle})
+
+        def dump_verbatim_manifest(bundles: Iterable[str]) -> set[str]:
+            manifest_url = furl(url=endpoint, path='/manifest/files', args={
+                'catalog': catalog,
+                'format': ManifestFormat.verbatim_jsonl.value,
+                'filters': json.dumps({special_fields.bundle_uuid: {'is': list(bundles)}})
+            })
+            manifest_lines = BytesIO(self._get_url_content(PUT, manifest_url)).readlines()
+            manifest_content = {
+                json.dumps(json.loads(replica), sort_keys=True)
+                for replica in manifest_lines
+            }
+            self.assertEqual(len(manifest_lines), len(manifest_content))
+            return manifest_content
+
+        for manifest_format in ManifestFormat.verbatim_jsonl, ManifestFormat.verbatim_pfb:
+            if manifest_format in metadata_plugin.manifest_formats:
+                with self.subTest(format=manifest_format):
+                    unauthorized = dump_verbatim_manifest(all_bundles)
+                    with self._service_account_credentials:
+                        authorized = dump_verbatim_manifest(all_bundles)
+                        private_only = dump_verbatim_manifest(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