"""Acceptance must reject green process statuses with missing/failed evidence.""" import hashlib import io import json from pathlib import Path import sys import tempfile import tarfile import unittest from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parent / "containers")) import release_gate as gate import export_baseline import export_archive import baseline_comparison def passed_scenario(name): row = {"name": name, "status": "passed"} if name == "sendq_cleanup": row["evidence"] = {"checks": [{"name": check, "status": "passed", "real_partitioned_have_preserved": True, "server_was_stopped": check != "running_server_refused", "required_alias_and_queue_preserved": True} for check in sorted(gate.SENDQ_CHECKS)], "regular_files_and_symlinks_removed": True, "directories_and_fifo_preserved": True, "nested_symlink_targets_preserved": True, "fixture_records_and_bytes_preserved": True} elif name == "database_switch": row["evidence"] = {"both_physical_roots_verified": True, "sendq_physical_roots_cleaned": ["/db1", "/db2"], "real_partitioned_have_preserved": True, "readonly_path_contains_space_and_equals": True} return row class ReleaseGate(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="sdp-release-gate-") self.addCleanup(self.temp.cleanup) self.folder = Path(self.temp.name) self.write("source-manifest.json", [{"path": "Server/Unix/setup/install_sdp_python.sh", "sha256": "candidate"}]) self.write("baseline-source-manifest.json", [{"path": "source", "sha256": "baseline"}]) self.candidate = gate.manifest_hash(self.folder / "source-manifest.json") self.baseline = gate.manifest_hash(self.folder / "baseline-source-manifest.json") self.write("test-outcomes.json", {stage: 0 for stage in gate.PRIMARY_STAGES + gate.DISTRO_STAGES}) self.write("container-inspect.json", {"Image": "sha256:final", "State": {"Running": True}}) self.write("image-inspect.json", [{"Id": "sha256:final"}]) self.write("journal-path.txt", "/p4/1/journals/journal\n") self.write("p4d-process-argv.json", {"argv": ["p4d", "-J", "/p4/1/journals/journal"]}) self.write("python-runtimes.json", {"distro_supported": True, "baseline": {"version": "3.14.8"}}) self.write("python-installer-evidence.json", {"source_sha256": "candidate", "exit_code": 0, "build_os_user": "perforce"}) self.write("installer-command.txt", "/source/mkdirs.sh /fixture/mkdirs.cfg\n") self.write("service-final-state.txt", "ActiveState=active\nMainPID=123\n") self.write("coverage.json", {"source_matches_results": True, "failed_stages": [], "scripts": [ {"script": "backup_functions.sh", "level": "live", "outcome": "passed-for-listed-scenarios"}]}) for prefix in ("", "distro-"): offline_rows = [name.rsplit(".", 1)[1] + " (" + name + ") ... ok" for name in sorted(gate.OFFLINE_CASES)] self.write(prefix + "offline.log", "\n".join(offline_rows) + "\nRan %d tests in 1s\nOK\n" % len(offline_rows)) self.write(prefix + "integration.log", "\n".join("[PASS] " + name for name in sorted(gate.HARNESS_CASES)) + "\ntotal=7 pass=7 fail=0 skip=0\n") self.write(prefix + "runtime.log", {"scenarios": [passed_scenario(name) for name in sorted(gate.RUNTIME_CASES)]}) self.write(prefix + "maintenance.log", {"scenarios": [{"name": name, "status": "passed"} for name in sorted(gate.MAINTENANCE_CASES)]}) self.write(prefix + "baseline-comparison.json", {"scenarios": [{"name": name, "status": "passed"} for name in sorted(gate.BASELINE_CASES)], "baseline_manifest_sha256": self.baseline, "candidate_manifest_sha256": self.candidate}) def write(self, name, value): (self.folder / name).write_text(value if isinstance(value, str) else json.dumps(value)) def check(self): return gate.platform_errors(self.folder, self.candidate, self.baseline) def test_complete_consistent_evidence_is_accepted(self): self.assertEqual(self.check(), []) def test_zero_stage_status_does_not_hide_failed_assertion(self): p = self.folder / "integration.log" p.write_text(p.read_text() + "[FAIL] fixture bytes differ\n") self.assertTrue(any("failed assertion" in e for e in self.check())) def test_green_summary_does_not_replace_required_behavioral_checks(self): self.write("integration.log", "[PASS] command exited\ntotal=1 pass=1 fail=0 skip=0\n") self.assertTrue(any("missing state/content assertion" in e for e in self.check())) def test_missing_and_failed_stage_each_reject(self): for status in (None, 7): stages = {s: 0 for s in gate.PRIMARY_STAGES + gate.DISTRO_STAGES} if status is None: stages.pop("baseline-comparison") else: stages["baseline-comparison"] = status self.write("test-outcomes.json", stages) self.assertTrue(self.check()) def test_skips_and_absent_concurrency_regression_reject(self): for log in ("Ran 3 tests in 1s\nOK (skipped=1)\n", "Ran 3 tests in 1s\nOK\n"): self.write("offline.log", log) self.assertTrue(self.check()) def test_expected_application_error_logs_are_distinct_from_unittest_failure_headers(self): original = (self.folder / "offline.log").read_text() self.write("offline.log", "ERROR:root:example.py failed (exit 7): denied\n" + original) self.assertEqual(self.check(), []) for header in ("ERROR: test_failure (test_module.TestCase)\n", "FAIL: test_failure (test_module.TestCase)\n", "ERROR: setUpClass (test_module.TestCase)\n", "FAILED (failures=1)\n"): with self.subTest(header=header): self.write("offline.log", original + header) self.assertTrue(any("failed or skipped test" in error for error in self.check())) def test_module_mentions_and_expected_failures_cannot_replace_passed_tests(self): self.write("offline.log", "test_logging_concurrency test_baseline_runtime test_baseline_maintenance\nRan 3 tests in 1s\nOK\n") self.assertTrue(self.check()) rows = [name.rsplit(".", 1)[1] + " (" + name + ") ... expected failure" for name in sorted(gate.OFFLINE_CASES)] self.write("offline.log", "\n".join(rows) + "\nRan 22 tests in 1s\nOK\n") self.assertTrue(self.check()) def test_missing_recovery_completion_regression_rejects_each_runtime(self): required = {name for name in gate.OFFLINE_CASES if name.startswith("test_recovery_completion.")} self.assertGreaterEqual(len(required), 10) for filename in ("offline.log", "distro-offline.log"): original = (self.folder / filename).read_text() for name in sorted(required): with self.subTest(filename=filename, missing=name): self.write(filename, "\n".join(line for line in original.splitlines() if name not in line) + "\n") self.assertTrue(any(name in error for error in self.check())) self.write(filename, original) def test_arbitrary_passing_case_does_not_replace_live_or_baseline_contracts(self): for report in ("maintenance.log", "baseline-comparison.json"): original = (self.folder / report).read_text() self.write(report, {"scenarios": [{"name": "anything", "status": "passed"}], "baseline_manifest_sha256": self.baseline, "candidate_manifest_sha256": self.candidate}) self.assertTrue(any("missing required scenario" in e for e in self.check())) self.write(report, original) def test_rebuilt_image_or_wrong_journal_path_reject(self): self.write("image-inspect.json", [{"Id": "sha256:new-unexercised-image"}]) self.assertTrue(any("final image" in e for e in self.check())) self.write("image-inspect.json", [{"Id": "sha256:final"}]) self.write("journal-path.txt", "/p4/1/logs/journal\n") self.assertTrue(any("journal-path" in e for e in self.check())) def test_failed_or_omitted_recovery_scenario_rejects(self): for scenario in ({"name": "database_switch", "status": "failed"}, {"name": "other", "status": "passed"}): self.write("runtime.log", {"scenarios": [scenario]}) self.assertTrue(self.check()) def test_sendq_green_label_cannot_replace_required_scope_and_preservation_checks(self): for name in ("sendq_cleanup", "database_switch"): rows = [passed_scenario(case) for case in sorted(gate.RUNTIME_CASES)] next(row for row in rows if row["name"] == name).pop("evidence") self.write("runtime.log", {"scenarios": rows}) self.assertTrue(self.check(), name) for mutation in ("missing", "failed", "client_lost", "running", "alias"): rows = [passed_scenario(case) for case in sorted(gate.RUNTIME_CASES)] checks = next(row for row in rows if row["name"] == "sendq_cleanup")["evidence"]["checks"] if mutation == "missing": checks.pop() elif mutation == "failed": checks[0]["status"] = "failed" elif mutation == "running": checks[0]["server_was_stopped"] = False elif mutation == "alias": next(check for check in checks if check["name"] == "protected_root_alias_refused")["required_alias_and_queue_preserved"] = False else: next(check for check in checks if check["name"].startswith("shared_readonly_"))["real_partitioned_have_preserved"] = False self.write("runtime.log", {"scenarios": rows}) self.assertTrue(self.check(), mutation) def test_source_and_baseline_mismatches_reject(self): self.assertTrue(gate.platform_errors(self.folder, "wrong", self.baseline)) self.assertTrue(gate.platform_errors(self.folder, self.candidate, "wrong")) def test_unexercised_interpreter_build_and_wrong_live_journal_reject(self): self.write("python-installer-evidence.json", {"source_sha256": "old", "exit_code": 0, "build_os_user": "root"}) self.assertTrue(any("Python build" in e for e in self.check())) self.write("p4d-process-argv.json", {"argv": ["p4d", "-J", "/p4/1/logs/journal"]}) self.assertTrue(any("running p4d" in e for e in self.check())) def test_coverage_failure_is_not_hidden_by_live_classification(self): self.write("coverage.json", {"source_matches_results": True, "scripts": [ {"script": "backup_functions.sh", "level": "live", "outcome": "failed"}]}) self.assertTrue(any("failed/incomplete" in e for e in self.check())) def test_topology_requires_every_named_failure_and_recovery_case(self): rows = [{"name": name, "status": "passed"} for name in gate.TOPOLOGY_CASES] self.assertEqual(gate.scenario_errors({"scenarios": rows}, "topology", gate.TOPOLOGY_CASES), []) rows[0]["status"] = "blocked" self.assertTrue(gate.scenario_errors({"scenarios": rows}, "topology", gate.TOPOLOGY_CASES)) self.assertTrue(gate.scenario_errors({"scenarios": rows[1:]}, "topology", gate.TOPOLOGY_CASES)) def test_topology_rejects_changed_driver_or_unverified_node_source_and_image(self): drivers = {name: "hash-" + name for name in ("topology_integration.py", "topology_bootstrap.sh")} manifest = [{"path": "test/containers/" + name, "sha256": digest} for name, digest in drivers.items()] document = {"scenarios": [{"name": name, "status": "passed"} for name in gate.TOPOLOGY_CASES], "source_manifest_sha256": self.candidate, "all_required_passed": True, "journal": "/p4/1/journals/journal", "shared_storage": False, "driver_sha256": drivers, "image_evidence": {"node_id": "sha256:nodes"}, "nodes": {name: {"service": daemon, "main_pid": 123, "running_image_id": "sha256:nodes", "source_manifest_sha256": self.candidate, "journal": "/p4/1/journals/journal", "command": "p4d -J /p4/1/journals/journal -r /p4/1/root"} for name, daemon in gate.TOPOLOGY_NODES.items()}} self.assertEqual(gate.topology_errors(document, self.candidate, manifest), []) drivers["topology_integration.py"] = "development-driver" self.assertTrue(any("driver differs" in error for error in gate.topology_errors(document, self.candidate, manifest))) drivers["topology_integration.py"] = "hash-topology_integration.py" for field, value in (("running_image_id", "sha256:unexercised"), ("source_manifest_sha256", "old-source"), ("main_pid", 0), ("command", "p4d -J /p4/1/logs/journal")): original = document["nodes"]["edge"][field] document["nodes"]["edge"][field] = value self.assertTrue(gate.topology_errors(document, self.candidate, manifest), field) document["nodes"]["edge"][field] = original document["nodes"].pop("proxy") self.assertTrue(gate.topology_errors(document, self.candidate, manifest)) def test_versions_require_all_fourteen_cells_and_their_data_checks(self): cases = [{"platform": platform, "release": release, "status": "passed", "source_manifest_sha256": self.candidate, "journal": "/p4/1/journals/journal", "image_id": "sha256:fixture", "architecture": "arm64", "host_architecture": "arm64", "emulated": False, "binary_builds": {binary: "vendor/" + release + "/build" for binary in ("p4", "p4d", "p4broker", "p4p")}, "checksums": {binary: "a" * 64 for binary in ("p4", "p4d", "p4broker", "p4p")}, "python": "3.12", "scenarios": [passed_scenario(name) for name in gate.VERSION_CASES]} for platform in gate.VERSION_PLATFORMS for release in gate.VERSION_RELEASES] doc = {"status": "passed", "source_manifest_sha256": self.candidate, "cases": cases} self.assertEqual(gate.version_errors(doc, self.candidate), []) cases[0]["emulated"] = True self.assertTrue(any("architecture evidence" in error for error in gate.version_errors(doc, self.candidate))) cases[0]["emulated"] = False self.assertTrue(gate.version_errors(doc, "different-source")) cases[0]["source_manifest_sha256"] = "different-source" self.assertTrue(gate.version_errors(doc, self.candidate)) cases[0]["source_manifest_sha256"] = self.candidate original = cases[0]["binary_builds"]["p4d"] cases[0]["binary_builds"]["p4d"] = "unmatched-version" self.assertTrue(gate.version_errors(doc, self.candidate)) cases[0]["binary_builds"]["p4d"] = original cases[0]["scenarios"].pop() self.assertTrue(gate.version_errors(doc, self.candidate)) cases.pop(0) self.assertTrue(gate.version_errors(doc, self.candidate)) class SubmittedExport(unittest.TestCase): def test_baseline_paths_cannot_escape_selected_depot(self): for file in ("//other/source/file", "//depot/source/../outside", "//depot/source//absolute"): with self.subTest(file=file), self.assertRaises(ValueError): export_baseline.relative_path("//depot/source", file) def test_export_uses_submitted_revisions_not_workspace_bytes(self): with tempfile.TemporaryDirectory() as temp: target = Path(temp) / "baseline" rows = [{"depotFile": "//depot/sdp/script.sh", "rev": "5", "action": "edit", "type": "text+x"}, {"depotFile": "//depot/sdp/retired", "rev": "3", "action": "delete", "type": "text"}] from subprocess import CompletedProcess def print_revision(argv, **kwargs): self.assertEqual(argv, ["p4", "print", "-q", "//depot/sdp/script.sh#5"]) return CompletedProcess(argv, 0, b"submitted content\n", b"") with patch.object(export_baseline, "records", return_value=rows), patch.object(export_baseline.subprocess, "run", side_effect=print_revision): manifest = export_baseline.export("//depot/sdp", 34003, target) self.assertEqual(len(manifest), 1) self.assertEqual((target / "script.sh").read_bytes(), b"submitted content\n") self.assertEqual(manifest[0]["sha256"], hashlib.sha256(b"submitted content\n").hexdigest()) self.assertFalse((target / "retired").exists()) self.assertTrue((target / "script.sh").stat().st_mode & 0o111) class DeploymentArchive(unittest.TestCase): def archive(self, path, members): with tarfile.open(path, "w:gz") as stream: for name, data, kind in members: info = tarfile.TarInfo(name) info.mode, info.type, info.size = 0o755, kind, len(data) stream.addfile(info, io.BytesIO(data)) def test_archive_bytes_and_unix_content_preserved_windows_excluded(self): with tempfile.TemporaryDirectory() as temp: archive, target = Path(temp) / "source.tgz", Path(temp) / "baseline" self.archive(archive, [("sdp/Server/Unix/script", b"deployed\n", tarfile.REGTYPE), ("sdp/Server/Windows/old.ps1", b"unused", tarfile.REGTYPE)]) before = archive.read_bytes() rows = export_archive.export(archive, target, confirmed=True) self.assertEqual(archive.read_bytes(), before) self.assertEqual(len(rows), 1) self.assertEqual((target / "Server/Unix/script").read_bytes(), b"deployed\n") self.assertFalse((target / "Server/Windows").exists()) info = json.loads((target / ".baseline-info.json").read_text()) self.assertTrue(info["production_baseline_confirmed"]) self.assertEqual(info["archive_sha256"], hashlib.sha256(before).hexdigest()) self.assertEqual(baseline_comparison.verify_manifest(target), gate.manifest_hash(target / ".export-manifest.json")) def test_archive_rejects_unsafe_paths_links_duplicates_before_writing(self): for members in ([("sdp/../../escape", b"x", tarfile.REGTYPE)], [("/sdp/absolute", b"x", tarfile.REGTYPE)], [("sdp/link", b"", tarfile.SYMTYPE)], [("sdp/x", b"x", tarfile.REGTYPE)] * 2): with self.subTest(members=members), tempfile.TemporaryDirectory() as temp: archive, target = Path(temp) / "source.tgz", Path(temp) / "baseline" self.archive(archive, members) with self.assertRaises(ValueError): export_archive.export(archive, target) self.assertFalse(target.exists()) def test_expected_failure_record_is_not_passing_baseline_evidence(self): class KnownFailure(unittest.TestCase): @unittest.expectedFailure def test_failure(self): self.fail("known bug") result = unittest.TextTestRunner(stream=io.StringIO(), resultclass=baseline_comparison.Results).run( unittest.defaultTestLoader.loadTestsFromTestCase(KnownFailure)) self.assertTrue(result.expectedFailures) self.assertEqual(result.scenarios[0]["status"], "failed") if __name__ == "__main__": unittest.main()