test_container_tools.py #1

#!/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()