#!/usr/bin/env python3 """Test exporter/runner policy with fake Perforce/Docker processes only.""" import hashlib import io import json import marshal from pathlib import Path import tempfile import types import unittest from unittest import mock from containers import export_source, run_matrix class ExportTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name).resolve() self.source = self.root / "sdp" self.source.mkdir() def process(self, rows, rc=0): output = io.BytesIO() for row in rows: marshal.dump({key.encode(): value.encode() for key, value in row.items()}, output) output.seek(0) return types.SimpleNamespace(stdout=output, stderr=io.BytesIO(), wait=lambda: rc) def test_inventory_excludes_depot_and_pending_deletes_includes_readd(self): rows = [] for name, head, action in [("active", "edit", ""), ("stale", "delete", ""), ("pending", "edit", "delete"), ("moved", "edit", "move/delete"), ("readd", "delete", "add"), ("new", "", "add")]: rows.append(dict(clientFile=str(self.source / name), headAction=head, action=action)) with mock.patch.object(export_source.subprocess, "Popen", return_value=self.process(rows)): records = export_source.perforce_files(self.source) self.assertEqual([Path(row["clientFile"]).name for row in records], ["active", "readd", "new"]) def test_export_captures_current_unsubmitted_bytes_and_sha(self): current = self.source / "current.sh" current.write_bytes(b"unsubmitted content\n") current.chmod(0o755) (self.source / "stale.sh").write_text("must not copy") rows = [dict(clientFile=str(current), depotFile="//test/current.sh", headAction="edit", action="edit", haveRev="8")] destination = self.root / "export" with mock.patch.object(export_source, "perforce_files", return_value=rows): manifest = export_source.export(self.source, destination) self.assertEqual((destination / "current.sh").read_bytes(), current.read_bytes()) self.assertEqual(manifest[0]["sha256"], hashlib.sha256(current.read_bytes()).hexdigest()) self.assertFalse((destination / "stale.sh").exists()) self.assertEqual((destination / "current.sh").stat().st_mode & 0o777, 0o755) self.assertEqual(json.loads((destination / ".export-manifest.json").read_text())[0]["haveRev"], "8") def test_export_missing_maintained_file_fails_before_creation(self): destination = self.root / "export" with mock.patch.object(export_source, "perforce_files", return_value=[dict(clientFile=str(self.source / "missing"))]), self.assertRaises(ValueError): export_source.export(self.source, destination) self.assertFalse(destination.exists()) def test_inventory_error_cannot_silently_export_partial_tree(self): with mock.patch.object(export_source.subprocess, "Popen", return_value=self.process([{"code": "error", "data": "connection failed"}], 1)), self.assertRaises(RuntimeError): export_source.perforce_files(self.source) class RunnerTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) args = types.SimpleNamespace(results=self.root, docker="fake-docker", name_prefix="unit", source=None, release="r26.1") self.matrix = run_matrix.Matrix(args) def test_runner_refuses_unowned_existing_container(self): result = types.SimpleNamespace(returncode=0, stdout=json.dumps([{"Config": {"Labels": {}}}])) with mock.patch.object(self.matrix, "command", return_value=result), self.assertRaises(RuntimeError): self.matrix.inspect_owned("ubuntu") def test_start_has_no_host_mounts_or_published_ports(self): folder = self.root / "ubuntu" folder.mkdir() (folder / "image-inspect.json").write_text(json.dumps([{"Id": "sha256:built"}])) def result(args, *unused, **kwargs): return types.SimpleNamespace(returncode=0, stdout="running\n", stderr="") with mock.patch.object(self.matrix, "inspect_owned", side_effect=[None, {"Image": "sha256:built"}]), mock.patch.object(self.matrix, "command", side_effect=result) as command: self.matrix.start("ubuntu") run = next(call.args[0] for call in command.call_args_list if call.args[0][0] == "run") self.assertIn("--privileged", run) for flag in ("-v", "--volume", "--mount", "-p", "--publish", "--network=host"): self.assertNotIn(flag, run) self.assertEqual(run[-1], "sha256:built") self.assertTrue(json.loads((folder / "image-reproducibility.json").read_text())["matches"]) def test_start_rejects_existing_container_from_an_older_image(self): folder = self.root / "ubuntu" folder.mkdir() (folder / "image-inspect.json").write_text(json.dumps([{"Id": "sha256:built"}])) with mock.patch.object(self.matrix, "inspect_owned", return_value={"Image": "sha256:older"}), mock.patch.object(self.matrix, "command") as command, self.assertRaisesRegex(RuntimeError, "recorded built image"): self.matrix.start("ubuntu") command.assert_not_called() def test_bootstrap_requires_manifest_before_copy(self): with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command") as command, self.assertRaises(RuntimeError): self.matrix.bootstrap("ubuntu") command.assert_not_called() def test_bootstrap_requires_immutable_baseline_before_copying_inputs(self): source = self.root / "source" source.mkdir() (source / ".export-manifest.json").write_text("[]") self.matrix.args.source = source with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command") as command, self.assertRaisesRegex(RuntimeError, "baseline-source"): self.matrix.bootstrap("ubuntu") command.assert_not_called() def test_collection_preserves_runner_outcomes_over_container_snapshot(self): folder = self.root / "ubuntu" folder.mkdir() outcomes = folder / "test-outcomes.json" outcomes.write_text(json.dumps({"integration": 1, "runtime": 0})) (folder / "runtime.log").write_text("fresh runtime output") def command(arguments, *unused, **kwargs): if arguments[0] == "cp": outcomes.write_text(json.dumps({"runtime": 0})) (folder / "runtime.log").write_text("stale runtime output") (folder / "maintenance.log").write_text("stale unrun stage") return types.SimpleNamespace(returncode=0, stdout="") with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command", side_effect=command): self.matrix.collect("ubuntu") self.assertEqual(json.loads(outcomes.read_text()), {"integration": 1, "runtime": 0}) self.assertEqual((folder / "runtime.log").read_text(), "fresh runtime output") self.assertFalse((folder / "maintenance.log").exists()) def test_bootstrap_archives_old_stage_results_before_new_installation(self): source, baseline = self.root / "source", self.root / "baseline" for path in (source, baseline): path.mkdir() (path / ".export-manifest.json").write_text("[]") (baseline / ".baseline-info.json").write_text("{}") self.matrix.args.source, self.matrix.args.baseline_source = source, baseline self.matrix.args.resume = True folder = self.root / "ubuntu" folder.mkdir() (folder / "test-outcomes.json").write_text('{"offline": 0}') (folder / "offline.log").write_text("earlier run") with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command") as command: self.matrix.bootstrap("ubuntu") invalidation = next(call.args[0] for call in command.call_args_list if call.args[0][2:4] == ["rm", "-f"]) self.assertIn("/results/maintenance.log", invalidation) self.assertIn("/results/baseline-comparison.json", invalidation) self.assertNotIn("/results/python-installer-evidence.json", invalidation) self.assertNotIn("/results/python-installer.log", invalidation) self.assertFalse((folder / "test-outcomes.json").exists()) self.assertFalse((folder / "offline.log").exists()) archives = list(folder.glob("previous-validation-*")) self.assertEqual(len(archives), 1) self.assertEqual((archives[0] / "offline.log").read_text(), "earlier run") def test_selected_test_preserves_other_stage_outcome(self): folder = self.root / "ubuntu" folder.mkdir() (folder / "test-outcomes.json").write_text(json.dumps({"runtime": 1})) self.matrix.args.test_stage = ["offline"] with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command", return_value=types.SimpleNamespace(returncode=0)) as command: self.assertTrue(self.matrix.test("ubuntu")) self.assertEqual(command.call_count, 3) # select, test, restore baseline test_command = command.call_args_list[1].args[0] self.assertIn("SDP_BASELINE_SOURCE=/inputs/baseline-sdp", test_command) self.assertIn("SDP_CANDIDATE_SOURCE=/inputs/sdp", test_command) self.assertEqual(json.loads((folder / "test-outcomes.json").read_text()), {"runtime": 1, "offline": 0}) def test_test_stage_records_offline_and_integration_failures_independently(self): self.matrix.args.test_stage = list(run_matrix.PRIMARY_STAGES) def command(arguments, *unused, **kwargs): return types.SimpleNamespace(returncode=int("unittest" in arguments)) with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "command", side_effect=command): self.assertTrue(self.matrix.test("ubuntu")) self.assertEqual(json.loads((self.root / "ubuntu/test-outcomes.json").read_text()), {"offline": 1, "integration": 0, "runtime": 0, "maintenance": 0, "baseline-comparison": 0, "python-installer": 0}) def test_distro_selection_failure_is_recorded_and_baseline_restored(self): self.matrix.args.test_stage = ["distro-offline"] with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "select_python", side_effect=[RuntimeError("unavailable"), None]) as select, self.assertRaises(RuntimeError): self.matrix.test("ubuntu") self.assertEqual(select.call_args_list, [mock.call("ubuntu", "distro"), mock.call("ubuntu", "baseline")]) self.assertEqual(json.loads((self.root / "ubuntu/test-outcomes.json").read_text()), {"distro-offline": 1}) def test_distro_comparison_has_independent_result_path(self): self.matrix.args.test_stage = ["distro-baseline-comparison"] with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "select_python") as select, mock.patch.object(self.matrix, "command", return_value=types.SimpleNamespace(returncode=0)) as command: self.assertFalse(self.matrix.test("ubuntu")) self.assertEqual(command.call_args.args[0][-1], "/results/distro-baseline-comparison.json") self.assertEqual(select.call_args_list, [mock.call("ubuntu", "distro"), mock.call("ubuntu", "baseline")]) def test_runtime_restore_failure_preserves_stage_failure_evidence(self): self.matrix.args.test_stage = ["distro-offline"] with mock.patch.object(self.matrix, "inspect_owned", return_value={"owned": True}), mock.patch.object(self.matrix, "select_python", side_effect=[None, RuntimeError("restore failed")]), mock.patch.object(self.matrix, "command", return_value=types.SimpleNamespace(returncode=1)), self.assertRaisesRegex(RuntimeError, "restore the primary"): self.matrix.test("ubuntu") self.assertEqual(json.loads((self.root / "ubuntu/test-outcomes.json").read_text()), {"distro-offline": 1, "python-runtime-restore": 1}) if __name__ == "__main__": unittest.main()