Accept the corpus retention-payload observation in qualification receipts

scripts/ci/check-docker-workflow-isolation.py

@@ -17,7 +17,7 @@
 # explicit model update and countercontrol review before gaining coverage.
 QUALIFICATION_CLEANUP_SOURCES = {
     "scripts/ci/native_qualification.py": "65916e73b7f0c6431c612f411e27f9429a2c37dbab287d1f17b840dee0f8d185",
-    "scripts/ci/native_alias_evidence.py": "f5af7af8fa8eeeda1b205dfd9316d6d44277f814e850bc8356d85f1cac1c43c3",
+    "scripts/ci/native_alias_evidence.py": "dc213ce6d43e5ad050de2acb730140aa33e4c21825f8dc5bf86e66f69106b95d",
     "scripts/ci/native_alias_transport.py": "16e495ed6874f7d62f29f0f4dc11aaf9e1715bd7046823f48d3443b3ed370c74",
 }
 

scripts/ci/native_alias_evidence.py

@@ -452,8 +452,9 @@ def receipt(provider: Provider, artifact_id: int, digest: str, name: str,
     "actual_image_config_digest", "container_image_architecture", "container_image_os",
     "image_source_commit", "image_release_tag", "observations", "cleanup",
 }
+# Exactly the status object native-image-corpus.sh writes (Server #325).
 CORPUS_OBSERVATIONS = {"reference_records", "native_record", "compare", "service_calls",
-                       "schedule_catalog", "namespace_delete_payloads"}
+                       "schedule_catalog", "namespace_delete_payloads", "retention_payloads"}
 COMPOSE_OBSERVATIONS = {"health_serving", "ready", "cluster_version_matches_release",
                         "worker_registered", "loopback_binding", "unset_token_refused",
                         "production_default_token_refused"}

tests/Unit/Support/native_qualification_test.py

@@ -604,6 +604,36 @@ def test_diagnostic_context_does_not_query_provider_or_docker(self):
                     self.assertIs(result["eligible"], False)
         self.assertEqual(provider.commands.reads, [])
 
+    def test_closed_observation_sets_equal_the_producer_scripts_output(self):
+        # The receipt sets must match what the frozen producers actually write.
+        corpus = (ROOT / "scripts/conformance/server-parity/native-image-corpus.sh").read_text()
+        status = re.findall(r"status:\{([^}]*)\}", corpus)
+        self.assertEqual(len(status), 1)
+        self.assertEqual({item.split(":")[0].strip() for item in status[0].split(",")}, E.CORPUS_OBSERVATIONS)
+        compose = (ROOT / "scripts/smoke-published-compose.sh").read_text()
+        results = re.findall(r"results = \{(.*?)\n    \}", compose, re.S)
+        self.assertEqual(len(results), 1)
+        self.assertEqual(set(re.findall(r'^\s*"([a-z_]+)":', results[0], re.M)), E.COMPOSE_OBSERVATIONS)
+
+    def test_corpus_observations_accept_every_written_status_and_refuse_a_missing_one(self):
+        # Exact status object of qualifier run 42407065494's native-image-corpus.json.
+        real = json.loads('{"reference_records":0,"native_record":0,"compare":0,"service_calls":0,'
+                          '"schedule_catalog":0,"namespace_delete_payloads":0,"retention_payloads":0}')
+        self.assertEqual(Q.qualification_observations("corpus", None, dict(real)), real)
+        old_six = E.CORPUS_OBSERVATIONS - {"retention_payloads"}
+        with self.assertRaisesRegex(E.Refusal, "qualification_observation_shape"):
+            E.closed(dict(real), old_six, "qualification_observation_shape")
+        with self.assertRaisesRegex(E.Refusal, "qualification_observation_shape"):
+            Q.qualification_observations("corpus", None, {**real, "unexpected": 0})
+        written = {key: 0 for key in E.CORPUS_OBSERVATIONS}
+        self.assertEqual(Q.qualification_observations("corpus", None, dict(written)), written)
+        for missing in sorted(E.CORPUS_OBSERVATIONS):
+            partial = {key: 0 for key in E.CORPUS_OBSERVATIONS - {missing}}
+            with self.assertRaisesRegex(E.Refusal, "qualification_observation_shape"):
+                Q.qualification_observations("corpus", None, partial)
+        with self.assertRaisesRegex(E.Refusal, "qualification_corpus_failed"):
+            Q.qualification_observations("corpus", None, {**written, "retention_payloads": 1})
+
     def test_observed_image_and_finalized_receipt_bind_to_consumer(self):
         provider = ProducerProvider()
         with patch.dict(os.environ, CONTEXT, clear=True):