#!/usr/bin/env python3 """Offline runtime improvements: no installed /p4, services, or network use.""" import json import hashlib import importlib.util import os from pathlib import Path import shutil import subprocess import sys import tempfile import time import unittest import test_maintenance_logging as maintenance_fixtures BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin" BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash")) class RuntimeImprovements(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="sdp-improvements-") self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.logs = self.root / "logs" self.checkpoints = self.root / "checkpoints" self.old = self.root / "original database" self.new = self.root / "replacement database" self.mockbin = self.root / "mockbin" for p in [self.logs, self.checkpoints, self.old, self.new, self.mockbin]: p.mkdir() (self.root / "root").symlink_to(self.old) (self.root / "offline_db").symlink_to(self.new) self.env = dict(os.environ, TEST_ROOT=str(self.root), LOGS=str(self.logs), LOGFILE=str(self.logs / "checkpoint.log"), CHECKPOINTS=str(self.checkpoints), P4CKP_PREFIX="p4_test", P4CBIN=str(self.root), P4ROOT=str(self.root / "root"), BACKUP_FUNCTIONS=str(BIN / "backup_functions.sh"), PATH=str(self.mockbin) + ":" + os.environ["PATH"], AWK="awk", KEEPCKPS="2", KEEPJNLS="1") self.prefix = '''source "$BACKUP_FUNCTIONS" OFFLINE_DB="$TEST_ROOT/offline_db" log () { printf '%s\\n' "$*" >> "$LOGFILE"; } die () { log "$*"; printf '%s\\n' "$*" >&2; exit 97; } mail_log_file () { :; } start_p4d () { echo started >> "$TEST_ROOT/services"; } compare_journal_numbers () { :; } copy_readonly_clients_dir () { return 0; } ''' # Native macOS mv lacks GNU -T. Emulate its rename semantics only for # fixture symlinks; all other moves are real moves inside this temp tree. self.write_executable(self.mockbin / "mv", "#!" + sys.executable + r''' import os, subprocess, sys from pathlib import Path args = sys.argv[1:] root = Path(os.environ["TEST_ROOT"]) needle = os.environ.get("FAIL_MOVE", "") matched = needle and needle in args[-1] if os.environ.get("FAIL_ROLLBACK") and (root / "move-failed").exists() and args[-1] == str(root / "root"): print("injected rollback failure", file=sys.stderr) sys.exit(20) if matched and (os.environ.get("FAIL_MOVE_ALWAYS") or not (root / "move-failed").exists()): (root / "move-failed").touch() print("injected move failure", file=sys.stderr) sys.exit(19) if "-Tf" in args: os.replace(args[-2], args[-1]) else: sys.exit(subprocess.call(["/bin/mv"] + args)) ''') @staticmethod def write_executable(path, text): path.write_text(text) path.chmod(0o755) def command(self, script, extra=None): return [BASH, "-c", self.prefix + script], dict(self.env, **(extra or {})) def run_bash(self, script, extra=None): args, env = self.command(script, extra) return subprocess.run(args, env=env, cwd=self.root, text=True, capture_output=True, timeout=15) def checkpoint(self, n, kind="serial", complete=True): path = self.checkpoints / ("p4_test.ckp." + str(n)) if kind == "parallel": path.mkdir() (path / "db.counters.ckp.gz").write_text("contents") elif kind == "plain": path.write_text("contents") else: path = Path(str(path) + ".gz") path.write_text("contents") if complete: Path(str(path) + ".md5").write_text("digest") return path def journal(self, n): path = self.checkpoints / ("p4_test.jnl." + str(n) + ".gz") path.write_text("journal") return path def test_checkpoint_lock_has_one_owner_under_contention(self): args, env = self.command('''ckp_running quiet echo acquired >> "$TEST_ROOT/owners" while [[ ! -f "$TEST_ROOT/release" ]]; do sleep 0.01; done ckp_complete''') owner = subprocess.Popen(args, env=env, cwd=self.root, stdout=subprocess.PIPE, stderr=subprocess.PIPE) self.addCleanup(lambda: owner.poll() is None and owner.kill()) for _ in range(200): if (self.root / "owners").exists(): break time.sleep(0.01) contenders = [subprocess.Popen(args, env=env, cwd=self.root, stdout=subprocess.PIPE, stderr=subprocess.PIPE) for _ in range(12)] for proc in contenders: stdout, stderr = proc.communicate(timeout=10) self.assertEqual(proc.returncode, 0, stderr) self.assertEqual(stdout + stderr, b"") self.assertEqual((self.root / "owners").read_text(), "acquired\n") (self.root / "release").touch() owner.communicate(timeout=10) self.assertEqual(owner.returncode, 0) self.assertFalse((self.logs / "ckp_running.txt").exists()) def test_unhandled_exit_without_die_keeps_its_lock(self): result = self.run_bash('ckp_running; exit 4') self.assertEqual(result.returncode, 4) self.assertTrue((self.logs / "ckp_running.txt").exists()) def test_nonowner_cannot_clear_replaced_or_inherited_lock(self): for script in ['ckp_complete', 'ckp_running; ( ckp_complete )', 'ckp_running; echo replacement > "$LOGS/ckp_running.txt"; ckp_complete']: with self.subTest(script=script): lock = self.logs / "ckp_running.txt" if lock.exists(): lock.unlink() if script == 'ckp_complete': lock.write_text("legacy lock\n") result = self.run_bash(script) self.assertEqual(result.returncode, 97, result.stderr) self.assertTrue(lock.exists()) def test_all_existing_checkpoint_callers_skip_before_shared_logs(self): common = self.root / "common" common.mkdir() (common / "backup_functions.sh").write_text((BIN / "backup_functions.sh").read_text()) (common / "sdp_checkpoint_lock.sh").write_text((BIN / "sdp_checkpoint_lock.sh").read_text()) (common / "p4_vars").write_text('export LOGS="$TEST_ROOT/logs" P4CBIN="$TEST_ROOT/common"\n') (self.logs / "ckp_running.txt").write_text("another owner\n") active = self.logs / "checkpoint.log" active.write_text("owner log\n") scripts = ["daily_checkpoint.sh", "rotate_journal.sh", "live_checkpoint.sh", "recreate_db_checkpoint.sh", "recreate_db_sync_replica.sh", "sync_replica.sh", "recreate_offline_db.sh", "recover_edge.sh", "journal_watch.sh", "convert_standby_to_master.sh", "convert_old_master_to_standby.sh", "replica_cleanup.sh", "edge_dump.sh"] for name in scripts: with self.subTest(script=name): target = common / name target.write_text((BIN / name).read_text().replace("/p4/common/bin", str(common))) result = subprocess.run([BASH, str(target), "1", "standby-id-or-checkpoint", "/fixture/checkpoint"], env=self.env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(active.read_text(), "owner log\n") self.assertEqual((self.logs / "ckp_running.txt").read_text(), "another owner\n") def test_retention_counts_complete_generations_across_formats(self): old = self.checkpoint(1) retained = [self.checkpoint(4, "parallel"), self.checkpoint(7, "plain")] partial = self.checkpoint(99, complete=False) journals = {n: self.journal(n) for n in range(1, 10)} result = self.run_bash('OFFLINEJNLNUM=9; remove_old_checkpoints_and_journals') self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(old.exists()) self.assertTrue(all(p.exists() for p in retained + [partial])) self.assertTrue(all(journals[n].exists() for n in range(4, 10))) self.assertFalse(any(journals[n].exists() for n in range(1, 4))) def test_retention_protects_offline_replay_and_minimum_count(self): self.checkpoint(7) journals = {n: self.journal(n) for n in range(1, 10)} result = self.run_bash('KEEPJNLS=3; OFFLINEJNLNUM=5; remove_old_checkpoints_and_journals') self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(all(journals[n].exists() for n in range(5, 10))) self.assertFalse(any(journals[n].exists() for n in range(1, 5))) def test_retention_preserves_everything_on_chain_gap(self): checkpoints = [self.checkpoint(n) for n in [1, 4, 7]] journals = [self.journal(n) for n in [1, 4, 6, 7, 8]] result = self.run_bash('OFFLINEJNLNUM=8; remove_old_checkpoints_and_journals') self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(all(p.exists() for p in checkpoints + journals)) self.assertIn("missing journal 5", (self.logs / "checkpoint.log").read_text()) def test_retention_without_complete_checkpoint_preserves_journals(self): self.checkpoint(8, complete=False) journals = [self.journal(n) for n in range(5)] result = self.run_bash('remove_old_checkpoints_and_journals') self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(all(p.exists() for p in journals)) def test_keepckps_zero_preserves_no_cleanup_policy(self): artifacts = [self.checkpoint(1), self.checkpoint(2), self.journal(0)] result = self.run_bash('KEEPCKPS=0; KEEPJNLS=0; remove_old_checkpoints_and_journals') self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(all(p.exists() for p in artifacts)) def test_missing_completion_marker_preserves_existing_offline_database(self): self.checkpoint(5, complete=False) (self.new / "db.counters").write_text("preserve") result = self.run_bash('''DoParallelCheckpoints=0 P4DBIN=unexpected_restore unexpected_restore () { echo 'restore must not run without completion marker' >&2; return 9; } recreate_offline_db_files''') self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.new / "db.counters").read_text(), "preserve") self.assertNotIn("restore must not run", (self.logs / "checkpoint.log").read_text()) def prepare_switch(self): (self.old / "db.counters").write_text("old counters") (self.old / "db.rev").write_text("old revisions") (self.new / "db.counters").write_text("new counters") (self.new / "offline_db_usable.txt").touch() (self.old / "server.id").write_text("original server") (self.new / "server.id").write_text("offline server") (self.old / "server.locks").mkdir() (self.old / "server.locks/lock").touch() (self.old / "save").mkdir() (self.old / "save/db.counters").write_text("prior saved counters") def test_database_switch_retains_old_db_until_successful_completion(self): self.prepare_switch() result = self.run_bash('switch_db_files') self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / "root").resolve(), self.new) self.assertEqual((self.root / "offline_db").resolve(), self.old) self.assertEqual((self.old / "save/db.counters").read_text(), "old counters") self.assertEqual((self.new / "server.id").read_text(), "original server") self.assertFalse((self.old / "server.locks").exists()) self.assertFalse((self.new / "offline_db_usable.txt").exists()) def test_database_switch_rolls_back_failed_second_link_and_metadata(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": str(self.root / "offline_db")}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.root / "root").resolve(), self.old) self.assertEqual((self.root / "offline_db").resolve(), self.new) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.new / "server.id").read_text(), "offline server") self.assertEqual((self.old / "save/db.counters").read_text(), "prior saved counters") self.assertEqual((self.root / "services").read_text(), "started\n") def test_database_switch_rolls_back_partial_database_move(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": "/db/"}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.new / "server.id").read_text(), "offline server") self.assertEqual((self.root / "services").read_text(), "started\n") def test_database_switch_rolls_back_first_link_failure(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": str(self.root / "root")}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.root / "root").resolve(), self.old) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.new / "server.id").read_text(), "offline server") self.assertEqual((self.root / "services").read_text(), "started\n") def test_database_switch_rolls_back_metadata_replacement(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": str(self.new) + "/"}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.new / "server.id").read_text(), "offline server") self.assertEqual((self.root / "services").read_text(), "started\n") def test_database_switch_rolls_back_previous_save_backup_failure(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": "previous-save"}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.old / "save/db.counters").read_text(), "prior saved counters") self.assertEqual((self.root / "root").resolve(), self.old) def test_database_switch_failed_rollback_stays_stopped_with_recovery_files(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": str(self.root / "offline_db"), "FAIL_ROLLBACK": "1"}) self.assertEqual(result.returncode, 97, result.stderr) self.assertFalse((self.root / "services").exists()) self.assertTrue((self.old / "P4ROOT_not_usable.txt").exists()) self.assertTrue(list(self.old.glob(".sdp-switch.*"))) self.assertIn("manual recovery", result.stderr) def test_database_switch_restores_previous_save_after_late_failure(self): self.prepare_switch() result = self.run_bash('switch_db_files', {"FAIL_MOVE": str(self.old / "save")}) self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual((self.old / "db.counters").read_text(), "old counters") self.assertEqual((self.old / "save/db.counters").read_text(), "prior saved counters") self.assertEqual((self.new / "server.id").read_text(), "offline server") self.assertEqual((self.root / "root").resolve(), self.old) def test_parallel_verify_rejects_missing_capability_before_restore(self): checkpoint = self.checkpoint(5, "parallel") result = self.run_bash('''P4DBIN=old_p4d old_p4d () { echo '-jv file'; } verify_checkpoint "$CHECKPOINTS/p4_test.ckp.5"''') self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("does not advertise -jvp", (self.logs / "checkpoint.log").read_text()) def test_database_switch_completion_uses_saved_physical_path(self): self.prepare_switch() result = self.run_bash('switch_db_files; start_p4d; complete_db_switch') self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.old / "save/db.counters").exists()) self.assertEqual((self.new / "db.counters").read_text(), "new counters") self.assertFalse(list(self.old.glob(".sdp-switch.*"))) class EdgeRecovery(unittest.TestCase): def setUp(self): self.f = RuntimeImprovements(methodName="runTest") self.f.setUp() self.addCleanup(self.f.doCleanups) f = self.f f.prepare_switch() common = f.root / "common" common.mkdir() for name in ["sdp_checkpoint_lock.sh", "backup_functions.sh"]: (common / name).write_text((BIN / name).read_text()) with (common / "backup_functions.sh").open("a") as out: out.write(''' check_vars () { :; }; set_vars () { OFFLINE_DB="$TEST_ROOT/offline_db"; } check_uid () { :; }; check_dirs () { :; }; log_init () { :; } rotate_log_file () { :; }; sdp_log_init () { :; }; mail_log_file () { :; } copy_readonly_clients_dir () { return 0; } stop_p4d () { echo stopped >> "$TEST_ROOT/services"; } start_p4d () { echo started >> "$TEST_ROOT/services"; } DoParallelCheckpoints=1 NumCheckPointThreads=2 ''') (common / "p4_vars").write_text('''export P4CBIN="$TEST_ROOT/common" export P4DBIN="$TEST_ROOT/p4d_mock" P4JOURNAL="$TEST_ROOT/logs/journal" export P4SERVER=p4_test P4HOME="$TEST_ROOT" ''') self.script = common / "recover_edge.sh" self.script.write_text((BIN / "recover_edge.sh").read_text().replace("/p4/common/bin", str(common))) f.write_executable(f.root / "p4d_mock", "#!" + sys.executable + r''' import json, os, sys from pathlib import Path args = sys.argv[1:] root = Path(os.environ["TEST_ROOT"]) if args == ["-h"]: print("-jvp verify parallel checkpoint") sys.exit(0) with (root / "p4d_calls.jsonl").open("a") as out: out.write(json.dumps(args) + "\n") source = args[-1] mode = "" if "-jr" in args or "-jrp" in args: mode = "master_restore" if "master.ckp" in source else "edge_replay" elif "-jd" in args or "-jdpm" in args: mode = "edge_dump" if ".sdp-edge-dump." in source else "checkpoint_dump" elif "-jv" in args or "-jvp" in args: mode = "verify" if mode == os.environ.get("FAIL_P4D_MODE"): print("injected p4d " + mode + " failure", file=sys.stderr) sys.exit(7) if "-jr" in args or "-jrp" in args: target = Path(args[args.index("-r") + 1]) (target / "db.counters").write_text("restored counters") (target / "db.rev").write_text("restored revisions") elif "-jd" in args or "-jdpm" in args: if source == "db.counters": print("@pv@ 0 @db.counters@ @journal@ @6@") elif source != "-": artifact = Path(source) if "-jdpm" in args: artifact.mkdir() (artifact / "db.counters.ckp.gz").write_text("dump") else: artifact.write_text("dump") Path(str(artifact) + ".md5").write_text("digest") ''') def master_checkpoint(self, parallel): source = self.f.root / ("master.ckp" if parallel else "master.ckp.gz") if parallel: source.mkdir() (source / "db.counters.ckp.gz").write_text("checkpoint") else: source.write_text("checkpoint") Path(str(source) + ".md5").write_text("digest") return source def run_edge(self, source, extra=None): return subprocess.run([BASH, str(self.script), "1", str(source)], env=dict(self.f.env, **(extra or {})), text=True, capture_output=True, timeout=15) def test_serial_master_checkpoint_is_staged_and_verified_before_restart(self): result = self.run_edge(self.master_checkpoint(False)) self.assertEqual(result.returncode, 0, result.stderr + (self.f.logs / "recover_edge.log").read_text()) calls = [json.loads(x) for x in (self.f.root / "p4d_calls.jsonl").read_text().splitlines()] master = [args for args in calls if "master.ckp.gz" in args[-1]] self.assertIn("-jv", master[0]) self.assertTrue(all("-jr" in args for args in master[1:])) self.assertEqual((self.f.root / "services").read_text(), "stopped\nstarted\n") self.assertEqual((self.f.root / "root/db.counters").read_text(), "restored counters") self.assertTrue((self.f.root / "offline_db/offline_db_usable.txt").exists()) self.assertFalse((self.f.logs / "ckp_running.txt").exists()) def test_edge_recovery_with_unwritable_metadata_parent_survives_next_switch(self): physical = self.f.root / "root-owned-metadata-parent" physical.mkdir() moved = physical / "offline" shutil.move(self.f.new, moved) self.f.new = moved (self.f.root / "offline_db").unlink() (self.f.root / "offline_db").symlink_to(moved) source = self.master_checkpoint(True) physical.chmod(0o555) self.addCleanup(physical.chmod, 0o755) result = self.run_edge(source) self.assertEqual(result.returncode, 0, result.stderr) candidate = (self.f.root / "root").resolve() self.assertTrue(candidate.is_relative_to(moved)) self.assertEqual((moved / "db.counters").read_text(), "new counters") self.assertEqual((candidate / "db.counters").read_text(), "restored counters") # A subsequent ordinary two-root switch must preserve its new active DB. result = self.f.run_bash("switch_db_files; complete_db_switch") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.f.root / "root/db.counters").read_text(), "restored counters") result = self.run_edge(source) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.f.root / "root/db.counters").read_text(), "restored counters") def test_parallel_master_checkpoint_uses_parallel_verify_and_restore(self): result = self.run_edge(self.master_checkpoint(True)) self.assertEqual(result.returncode, 0, result.stderr) calls = [json.loads(x) for x in (self.f.root / "p4d_calls.jsonl").read_text().splitlines()] master = [args for args in calls if args[-1].endswith("master.ckp")] self.assertIn("-jvp", master[0]) self.assertTrue(all("-jrp" in args for args in master[1:])) def test_failed_edge_switch_restores_both_original_database_links(self): result = self.run_edge(self.master_checkpoint(True), {"FAIL_MOVE": str(self.f.root / "offline_db")}) self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual((self.f.root / "root").resolve(), self.f.old) self.assertEqual((self.f.root / "offline_db").resolve(), self.f.new) self.assertEqual((self.f.old / "db.counters").read_text(), "old counters") self.assertEqual((self.f.new / "db.counters").read_text(), "new counters") self.assertEqual((self.f.root / "services").read_text(), "stopped\nstarted\n") def test_failed_edge_replay_preserves_live_database_and_releases_owned_lock(self): result = self.run_edge(self.master_checkpoint(True), {"FAIL_P4D_MODE": "edge_replay"}) self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual((self.f.old / "db.counters").read_text(), "old counters") self.assertEqual((self.f.new / "db.counters").read_text(), "new counters") self.assertEqual((self.f.root / "root").resolve(), self.f.old) self.assertEqual((self.f.root / "services").read_text(), "stopped\n") self.assertFalse((self.f.logs / "ckp_running.txt").exists()) class MissingTransfers(unittest.TestCase): def setUp(self): self.f = maintenance_fixtures.MaintenanceLogging(methodName="runTest") self.f.setUp() self.addCleanup(self.f.doCleanups) self.f.prepare_rsync_transport_probe() def calls(self): probe = self.f.root / "rsync-probe.jsonl" return [json.loads(x) for x in probe.read_text().splitlines()] if probe.exists() else [] def test_detailed_log_transfers_literal_paths_once_without_generated_shell(self): filename = 'path with spaces;$(touch INJECTED)`touch BACKTICK`\'".txt' (self.f.logs / "verify_output.log").write_text( f"info: //depot/{filename}#1 - MISSING!\ninfo: //depot/{filename}#2 - MISSING!\n") (self.f.logs / "p4verify.log").write_text("summary only\n") result = self.f.run_script("rsync_missing.sh", "1") self.assertEqual(result.returncode, 0, result.stderr) calls = self.calls() self.assertEqual(len(calls), 1) self.assertEqual(calls[0]["args"], ["-avs", "--", "test@master:" + str(self.f.home / "depots/depot" / filename) + ",d/", str(self.f.home / "depots/depot" / filename) + ",d/"]) self.assertFalse((self.f.root / "INJECTED").exists()) self.assertFalse((self.f.root / "BACKTICK").exists()) self.assertFalse((self.f.home / "tmp/temp_rsync.sh").exists()) def test_bad_record_prevents_all_transfers(self): for record in ["info: //depot/../escape#1 - MISSING!", "unexpected MISSING! message"]: with self.subTest(record=record): (self.f.logs / "verify_output.log").write_text("info: //depot/good#1 - MISSING!\n" + record + "\n") result = self.f.run_script("rsync_missing.sh", "1") self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual(self.calls(), []) def test_missing_detailed_log_requires_explicit_legacy_input(self): (self.f.logs / "verify_output.log").unlink(missing_ok=True) result = self.f.run_script("rsync_missing.sh", "1") self.assertEqual(result.returncode, 1) self.assertEqual(self.calls(), []) (self.f.logs / "p4verify.log").write_text("info: //depot/file#1 - MISSING!\n") result = self.f.run_script("rsync_missing.sh", "1", str(self.f.logs / "p4verify.log")) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(len(self.calls()), 1) class CoverageReporting(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="sdp-coverage-") self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.source = self.root / "source" self.results = self.root / "results" self.source.mkdir(); self.results.mkdir() spec = importlib.util.spec_from_file_location("sdp_coverage", Path(__file__).parent / "containers/coverage.py") self.coverage = importlib.util.module_from_spec(spec) spec.loader.exec_module(self.coverage) self.script = self.source / "Server/Unix/p4/common/bin/rsync_missing.sh" self.script.parent.mkdir(parents=True) self.script.write_text("#!/bin/bash\nexit 0\n") self.manifest = [{"path": self.script.relative_to(self.source).as_posix(), "sha256": hashlib.sha256(self.script.read_bytes()).hexdigest(), "symlink": None}] (self.source / ".export-manifest.json").write_text(json.dumps(self.manifest)) def test_uninventoried_stale_script_is_excluded_and_syntax_is_not_behavior(self): (self.script.parent / "deleted_in_depot.sh").write_text("#!/bin/bash\nexit 0\n") result = self.coverage.report(self.source, self.results) self.assertEqual(len(result["scripts"]), 1) self.assertEqual(result["scripts"][0]["level"], "static_only") self.assertFalse(result["source_matches_results"]) def test_live_evidence_requires_matching_source_manifest(self): (self.results / "runtime.log").write_text(json.dumps({"scenarios": [ {"name": "ssh_host_trust", "scripts": ["rsync_missing.sh"], "status": "passed", "detail": "fixture handshake"}]})) result = self.coverage.report(self.source, self.results) self.assertEqual(result["scripts"][0]["level"], "static_only") (self.results / "source-manifest.json").write_text(json.dumps(self.manifest)) result = self.coverage.report(self.source, self.results) self.assertEqual(result["scripts"][0]["level"], "live") self.script.write_text("#!/bin/bash\nexit 1\n") result = self.coverage.report(self.source, self.results) self.assertEqual(result["scripts"][0]["level"], "static_only") def test_unittest_parser_keeps_failures_and_skips_distinct(self): result = self.coverage.unittest_results("test_a (test_mod.Case.test_a) ... ok\n" "test_b (test_mod.Case) ... FAIL\n" "test_c (test_mod.Case) ... skipped 'no service'\n" "test_d (test_mod.Case) ... diagnostic output\nmore details\nok\n" " test_e (test_mod.Case.test_e) (mode='failed') ... FAIL\n") self.assertEqual([row["status"] for row in result], ["passed", "failed", "skipped", "passed", "failed"]) self.assertEqual(result[0]["test"], "test_mod.Case.test_a") def test_passing_case_never_hides_failed_harness_case(self): self.coverage.HARNESS["rsync_missing.sh"] = ["rsync_missing.sh"] (self.results / "source-manifest.json").write_text(json.dumps(self.manifest)) (self.results / "integration.log").write_text( "[PASS] rsync_missing.sh\n\x1b[31m[FAIL]\x1b[0m rsync_missing.sh -- exit=1\n") row = self.coverage.report(self.source, self.results)["scripts"][0] self.assertEqual(row["level"], "live") self.assertEqual(row["outcome"], "failed") self.assertEqual([case["status"] for case in row["live"]], ["passed", "failed"]) def test_skips_make_passing_scope_partial(self): passed = {"status": "passed"} self.assertEqual(self.coverage.aggregate_outcome(passed, [passed, {"status": "skipped"}]), "partial") self.assertEqual(self.coverage.aggregate_outcome(passed, []), "no-behavioral-evidence") def test_python_installer_uses_real_outcome_and_log(self): self.script.rename(self.script.with_name("install_sdp_python.sh")) self.manifest[0]["path"] = str(Path(self.manifest[0]["path"]).with_name("install_sdp_python.sh")) (self.source / ".export-manifest.json").write_text(json.dumps(self.manifest)) (self.results / "source-manifest.json").write_text(json.dumps(self.manifest)) (self.results / "test-outcomes.json").write_text(json.dumps({"python-installer": 0})) row = self.coverage.report(self.source, self.results)["scripts"][0] self.assertEqual(row["outcome"], "failed", "Missing log must not claim successful execution") (self.results / "python-installer.log").write_text("Built and installed\n") row = self.coverage.report(self.source, self.results)["scripts"][0] self.assertEqual(row["outcome"], "passed-for-listed-scenarios") self.assertEqual(row["level"], "live") def test_live_runner_refuses_host_without_container_marker(self): runner = Path(__file__).parent / "containers/runtime_integration.py" result = subprocess.run([sys.executable, str(runner), "--root", str(self.root), "--instance", "1", "--port", "19999", "--os-user", "perforce"], capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 2) self.assertIn("guard_error", json.loads(result.stdout)) if __name__ == "__main__": unittest.main()