test_runtime_improvements.py #2

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