runtime_integration.py #3

#!/usr/bin/env python3
"""Real p4d/systemd/SSH integration, only inside a marked disposable container.

JSON on stdout records each scenario and its exact behavioral scope. Handled
failures release owned checkpoint locks; recovery data remains for investigation.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import pwd
import shutil
import socket
import subprocess
import tempfile
import time

from fixture_inventory import FixtureInventory

BIN = Path("/p4/common/bin")


class RuntimeIntegration:
    def __init__(self, args):
        self.args = args
        self.records = []
        self.env = self.validate()
        self.home = Path(self.env["P4HOME"])
        self.logs = Path(self.env["LOGS"])
        self.counter = "sdp.disposable.runtime." + str(int(time.time()))
        self.fixture = FixtureInventory(self.env, "sdp-fixture-runtime-" + str(time.time_ns()))

    def validate(self):
        if not self.args.confirm_disposable or not Path("/run/sdp-container-test").is_file():
            raise RuntimeError("Explicit confirmation and container marker are required")
        if pwd.getpwuid(os.getuid()).pw_name != self.args.os_user:
            raise RuntimeError("Run as the marked disposable OS account")
        env = dict(os.environ, TEST_ROOT=self.args.root, INSTANCE=self.args.instance,
                   TEST_PORT=str(self.args.port), TEST_OSUSER=self.args.os_user)
        shell = '''source /p4/sdp/test/disposable_test_guard.sh || exit 2
sdp_test_validate_marker || exit 2
source /p4/common/bin/p4_vars "$INSTANCE" || exit 2
[[ "$P4PORT" == "127.0.0.1:$TEST_PORT" && "$OSUSER" == "$TEST_OSUSER" &&
   "$SERVERID" == "sdp-test-$INSTANCE" && "$SDPMAIL" == /bin/true ]] || exit 2
env -0'''
        result = subprocess.run(["bash", "-c", shell], env=env, capture_output=True, timeout=30)
        if result.returncode:
            raise RuntimeError("Disposable configuration guard failed: " + result.stderr.decode(errors="replace"))
        loaded = dict(item.decode().split("=", 1) for item in result.stdout.split(b"\0") if item)
        root = Path(loaded["P4ROOT"]).resolve()
        if not root.is_relative_to(Path(self.args.root)):
            raise RuntimeError("P4ROOT escapes the disposable deployment")
        info = subprocess.check_output([loaded["P4BIN"], "-p", loaded["P4PORT"], "-ztag", "info"], env=loaded, timeout=30).decode()
        info_tags = {fields[1].lower(): fields[2] for line in info.splitlines()
                     if len(fields := line.split(None, 2)) == 3 and fields[0] == "..."}
        if info_tags.get("serverid") != "sdp-test-" + self.args.instance:
            raise RuntimeError("Unexpected running server identity")
        return loaded

    def command(self, argv, *, check=True, timeout=900, env=None):
        result = subprocess.run([str(x) for x in argv], env=env or self.env, cwd=self.home,
                                capture_output=True, text=True, timeout=timeout)
        if check and result.returncode:
            raise RuntimeError("Command failed (%s): %s\n%s" % (result.returncode, argv[0], (result.stdout + result.stderr)[-5000:]))
        return result

    def script(self, name):
        result = self.command(["bash", BIN / name, self.args.instance], check=False)
        if result.returncode:
            detail = (result.stdout + result.stderr)[-3000:]
            for logfile in [self.logs / "checkpoint.log", self.logs / (name.removesuffix(".sh") + ".log")]:
                if logfile.is_file():
                    detail += "\n" + logfile.read_text(errors="replace")[-5000:]
            raise RuntimeError(name + " failed: " + detail)
        if (self.logs / "ckp_running.txt").exists():
            raise RuntimeError(name + " did not release its successful-run lock")

    def scenario(self, name, scripts, detail, operation):
        started = time.monotonic()
        row = {"name": name, "scripts": scripts, "detail": detail, "kind": "live"}
        try:
            evidence = operation()
            if evidence is not None:
                row["evidence"] = evidence
            row["status"] = "passed"
        except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
            row.update(status="failed", error=str(error))
        row["duration_seconds"] = round(time.monotonic() - started, 3)
        self.records.append(row)

    def checkpoint_cycle(self):
        self.fixture.create()
        self.command([self.env["P4BIN"], "counter", self.counter, "12345"])
        self.script("live_checkpoint.sh")
        self.script("daily_checkpoint.sh")
        self.script("recreate_offline_db.sh")
        marker = Path(self.env["P4HOME"]) / "offline_db/offline_db_usable.txt"
        if not marker.is_file():
            raise RuntimeError("Successful restore did not leave its usability marker")
        value = self.command([self.env["P4DBIN"], "-r", marker.parent, "-jd", "-", "db.counters"]).stdout
        if self.counter not in value or "12345" not in value:
            raise RuntimeError("Real checkpoint restore lost the integration counter")
        # Serve a copy of the actual restored offline DB, leaving its usability
        # state intact, and inspect every fixture revision/spec/shelf.
        clone = Path(tempfile.mkdtemp(prefix="sdp-restored-inventory-", dir=self.home / "tmp"))
        probe_stopped = False
        try:
            for table in marker.parent.glob("db.*"):
                shutil.copy2(table, clone / table.name)
            (clone / "server.id").write_text("sdp-test-restored-inventory\n")
            port = "127.0.0.1:20001"
            self.command([self.env["P4DBIN"], "-r", clone, "-p", port, "-J", "off", "-L", clone / "log", "--daemonsafe", "--pid-file=" + str(clone / "pid")])
            try:
                password = Path(self.env["SDP_ADMIN_PASSWORD_FILE"]).read_bytes()
                self.fixture.p4("login", "-a", data=password, port=port)
                self.fixture.assert_unchanged(port)
            finally:
                pid = int((clone / "pid").read_text())
                self.fixture.p4("admin", "stop", port=port)
                deadline = time.monotonic() + 30
                while Path("/proc/%s" % pid).exists():
                    if time.monotonic() >= deadline:
                        raise RuntimeError("Read-only restored probe did not stop before scratch cleanup")
                    time.sleep(0.05)
                probe_stopped = True
        except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
            if not probe_stopped:
                raise RuntimeError("Restored probe scratch retained at %s: %s" % (clone, error)) from error
            raise
        finally:
            if probe_stopped:
                shutil.rmtree(clone)
        (self.logs / "runtime-fixture-inventory.json").write_text(json.dumps(self.fixture.expected, indent=2) + "\n")

    def rotation(self):
        before = int(self.command([self.env["P4BIN"], "counter", "journal"]).stdout.strip())
        prefix = self.env["P4CKP_PREFIX"] + ".jnl."
        def raw_numbers():
            return sorted(int(p.name.removeprefix(prefix)) for p in Path(self.env["JOURNALS"]).glob(prefix + "*")
                          if p.name.removeprefix(prefix).isdigit())
        for _ in range(3):
            number = int(self.command([self.env["P4BIN"], "counter", "journal"]).stdout.strip())
            expected = raw_numbers() + [number]
            # Preserve the existing policy: >2 enables archiving; with 3/4
            # journals archive one, with >=5 keep the newest four.
            if len(expected) > 2:
                expected = expected[1:] if len(expected) < 5 else expected[-4:]
            self.script("rotate_journal.sh")
            if raw_numbers() != expected:
                raise RuntimeError("Journal gzip policy kept %r, expected %r" % (raw_numbers(), expected))
        after = int(self.command([self.env["P4BIN"], "counter", "journal"]).stdout.strip())
        if after - before != 3:
            raise RuntimeError("Three successful rotations did not advance the counter by three")

    def client_config(self, name):
        return {row["ServerName"]: row.get("Value", "")
                for row in self.fixture.p4("configure", "show", "allservers")
                if row.get("Name") == name and "ServerName" in row}

    def replace_client_config(self, name, values):
        # These tests are already restricted to a marked disposable deployment.
        # Preserve every original scope, including deliberately empty values.
        for scope in self.client_config(name):
            self.fixture.p4("configure", "unset", scope + "#" + name)
        for scope, value in values.items():
            self.fixture.p4("configure", "set", scope + "#" + name + "=" + value)
        if self.client_config(name) != values:
            raise RuntimeError("Server did not retain exact fixture configuration for " + name)

    @staticmethod
    def seed_sendq_tree(target, outside):
        target.mkdir(parents=True)
        nested = target / "nested directory"
        nested.mkdir()
        outside.mkdir(parents=True)
        sentinel = outside / "keep exact bytes.txt"
        sentinel.write_bytes(b"unrelated symlink target\n")
        files = [target / "queue.db", nested / "queue with spaces.db"]
        for path in files:
            path.write_bytes(b"disposable send queue\n")
        links = [target / "external file", nested / "external directory", nested / "broken link"]
        links[0].symlink_to(sentinel)
        links[1].symlink_to(outside, target_is_directory=True)
        links[2].symlink_to(outside / "absent")
        fifo = nested / "preserve fifo"
        os.mkfifo(fifo)
        return {"root": target, "directories": [target, nested], "files": files,
                "links": links, "link_targets": {path: os.readlink(path) for path in links},
                "fifo": fifo, "sentinel": sentinel,
                "directory_inodes": {path: path.stat().st_ino for path in (target, nested)}}

    @staticmethod
    def assert_sendq_tree(tree, cleaned):
        if not all(path.is_dir() and not path.is_symlink() for path in tree["directories"]):
            raise RuntimeError("Send-queue cleanup removed or replaced its root/directories")
        if any(path.stat().st_ino != inode for path, inode in tree["directory_inodes"].items()):
            raise RuntimeError("Send-queue cleanup replaced an existing directory")
        for path in tree["files"] + tree["links"]:
            if os.path.lexists(path) == cleaned:
                raise RuntimeError("Unexpected send-queue entry state: " + str(path))
        if not cleaned:
            if any(path.read_bytes() != b"disposable send queue\n" for path in tree["files"]):
                raise RuntimeError("Unselected/shared send-queue file bytes changed")
            if any(os.readlink(path) != destination for path, destination in tree["link_targets"].items()):
                raise RuntimeError("Unselected/shared send-queue symlink changed")
        if not tree["fifo"].is_fifo():
            raise RuntimeError("Send-queue cleanup removed a non-file special entry")
        if tree["sentinel"].read_bytes() != b"unrelated symlink target\n":
            raise RuntimeError("Send-queue cleanup followed a nested symlink")

    def sendq_helper(self, *, identity=None, root_override=None):
        shell = """source /p4/common/bin/p4_vars "$1" || exit 1
source /p4/common/bin/backup_functions.sh || exit 1
set_vars || exit 1
export LOGFILE="$2"
if [[ "$3" != inherit ]]; then export P4NAME="$3"; fi
if [[ "$4" != inherit ]]; then export P4D_FLAGS="$4"; fi
if [[ "$5" != inherit ]]; then export P4ROOT="$5"; fi
clean_client_sendq_dir
"""
        identity = identity or {}
        return self.command(["bash", "-c", shell, "sendq-live", self.args.instance,
                             self.logs / "sendq-cleanup-live.log", identity.get("P4NAME", "inherit"),
                             identity.get("P4D_FLAGS", "inherit"), root_override or "inherit"], check=False)

    def shared_sendq_checks(self, token):
        original_sendq = self.client_config("client.sendq.dir")
        original_readonly = self.client_config("client.readonly.dir")
        service = "p4d_" + self.args.instance + ".service"
        root = Path(self.env["P4ROOT"]).resolve()
        checks = []
        for name, readonly_suffix, sendq_suffix in (
                ("shared_readonly_equal_preserved", "", ""),
                ("shared_readonly_child_preserved", "/persistent", ""),
                ("shared_readonly_parent_preserved", "", "/nested directory"),
                ("shared_readonly_nested_alias_preserved", "/nested directory/external directory/persistent", ""),
                ("shared_readonly_alias_chain_preserved", None, "")):
            relative = token + "/" + name
            target = root / relative
            outside = self.home / "tmp" / (token + "-" + name)
            tree = self.seed_sendq_tree(target, outside)
            # A client named sendq.* creates genuinely persistent have data
            # with a queue-looking name. Prefix matching must not remove it.
            client = "sendq." + self.fixture.name + "-" + name
            workspace = outside / "workspace"
            workspace.mkdir()
            if "alias" in name:
                (outside / "persistent").mkdir()
            if readonly_suffix is None:
                alias = outside / "outer alias"
                alias.symlink_to(target / "nested directory/external directory/persistent", target_is_directory=True)
                readonly_value = str(alias)
            else:
                readonly_value = relative + readonly_suffix
            created = False
            try:
                self.replace_client_config("client.readonly.dir", {self.env["SERVERID"]: readonly_value})
                self.replace_client_config("client.sendq.dir", {self.env["SERVERID"]: relative + sendq_suffix})
                # Older releases cache the external client-directory setting;
                # restart after configuration before creating persistent state.
                self.command(["sudo", "-n", "systemctl", "restart", service])
                self.command([BIN / "p4login", self.args.instance])
                self.fixture.form("client", {"Client": client, "Owner": self.env["P4USER"],
                                  "Root": str(workspace), "Type": "partitioned", "LineEnd": "unix",
                                  "View0": "//%s/... //%s/..." % (self.fixture.name, client)})
                created = True
                effective = self.fixture.p4("configure", "show", "client.readonly.dir")
                form = self.fixture.p4("client", "-o", client)
                (self.logs / "sendq-shared-fixture-debug.json").write_text(json.dumps({"effective_readonly": effective, "client": form, "target": str(target)}, indent=2) + "\n")
                self.command([self.env["P4BIN"], "-c", client, "sync", "//" + self.fixture.name + "/..."])
                before_have = self.command([self.env["P4BIN"], "-c", client, "-ztag", "have", "//" + self.fixture.name + "/..."]).stdout
                if "... haveRev " not in before_have or str(workspace) not in before_have:
                    raise RuntimeError("Real partitioned client did not record synced revisions")
                self.command(["sudo", "-n", "systemctl", "stop", service])
                try:
                    readonly_root = (Path(readonly_value) if readonly_value.startswith("/") else root / readonly_value).resolve()
                    def hashes():
                        # Include externally stored readonly data reached through
                        # an alias; rglob deliberately does not follow queue links.
                        return {str(path.resolve()): hashlib.sha256(path.read_bytes()).hexdigest()
                                for folder in (target, readonly_root) for path in folder.rglob("*")
                                if path.is_file() and not path.is_symlink()}
                    before = hashes()
                    if not any(Path(path).name.startswith("db.sendq.") for path in before):
                        raise RuntimeError("Partitioned client did not create the queue-like persistent filename; observed=" + repr(sorted(before)))
                    result = self.sendq_helper()
                    if result.returncode:
                        raise RuntimeError("Shared readonly cleanup failed instead of preserving the directory")
                    self.assert_sendq_tree(tree, False)
                    if hashes() != before:
                        raise RuntimeError("Shared readonly cleanup changed persistent file bytes")
                finally:
                    self.command(["sudo", "-n", "systemctl", "start", service])
                    self.command([BIN / "p4login", self.args.instance])
                if self.command([self.env["P4BIN"], "-c", client, "-ztag", "have", "//" + self.fixture.name + "/..."]).stdout != before_have:
                    raise RuntimeError("Shared readonly cleanup changed real partitioned-client have state")
                self.fixture.assert_unchanged()
                checks.append({"name": name, "status": "passed", "server_was_stopped": True,
                               "persistent_files_checked": len(before), "real_partitioned_have_preserved": True})
            finally:
                if created:
                    self.fixture.p4("client", "-d", "-f", client)
                self.replace_client_config("client.readonly.dir", original_readonly)
                self.replace_client_config("client.sendq.dir", original_sendq)
                self.command(["sudo", "-n", "systemctl", "restart", service])
                self.command([BIN / "p4login", self.args.instance])
            shutil.rmtree(target)
            shutil.rmtree(outside)
        return checks

    def refused_sendq_alias_checks(self, token):
        original = self.client_config("client.sendq.dir")
        service = "p4d_" + self.args.instance + ".service"
        root = Path(self.env["P4ROOT"]).resolve()
        checks = []
        for name in ("protected_root_alias_refused", "configured_sendq_alias_refused"):
            scratch = self.home / "tmp" / (token + "-" + name)
            target = scratch / "queue"
            tree = self.seed_sendq_tree(target, scratch / "outside")
            required = target / "required alias"
            root_override = None
            if name == "protected_root_alias_refused":
                # Only this helper invocation uses the alias. The installed
                # service and its actual DB links retain their original paths.
                required.symlink_to(root, target_is_directory=True)
                root_override = str(required)
                sendq_value = str(target)
            else:
                required.symlink_to(target, target_is_directory=True)
                outer = scratch / "outside configured alias"
                outer.symlink_to(required, target_is_directory=True)
                sendq_value = str(outer)
            tree["links"].append(required)
            tree["link_targets"][required] = os.readlink(required)
            try:
                self.replace_client_config("client.sendq.dir", {self.env["SERVERID"]: sendq_value})
                self.command(["sudo", "-n", "systemctl", "stop", service])
                try:
                    if self.command(["sudo", "-n", "systemctl", "is-active", service], check=False).returncode == 0:
                        raise RuntimeError("Alias refusal fixture did not stop the installed service")
                    result = self.sendq_helper(root_override=root_override)
                    if result.returncode != 1:
                        raise RuntimeError("Required alias cleanup was not refused: " + name + "\n" + result.stdout + result.stderr)
                    self.assert_sendq_tree(tree, False)
                    if Path(sendq_value).resolve() != target.resolve():
                        raise RuntimeError("Refusal changed the configured queue resolution")
                finally:
                    self.command(["sudo", "-n", "systemctl", "start", service])
                    self.command([BIN / "p4login", self.args.instance])
                self.fixture.assert_unchanged()
                checks.append({"name": name, "status": "passed", "server_was_stopped": True,
                               "required_alias_and_queue_preserved": True})
            finally:
                self.replace_client_config("client.sendq.dir", original)
            shutil.rmtree(scratch)
        return checks

    def sendq_cleanup(self):
        original = self.client_config("client.sendq.dir")
        root = Path(self.env["P4ROOT"]).resolve()
        token = ".sdp-sendq-live-" + str(time.time_ns())
        scratch = self.home / "tmp" / token
        scratch.mkdir()
        relative_a, relative_b = token + "/global path", token + "/named path"
        absolute = str(scratch / "absolute queue")
        server, other, third = self.env["SERVERID"], "sdp-sendq-other", "sdp-sendq-third"
        cases = [
            ("unset_noop", {}, None, {}, False),
            ("any_relative", {"any": relative_a}, "global", {}, False),
            ("any_absolute", {"any": absolute}, "absolute", {}, False),
            ("configured_root_symlink", {"any": token + "/root link"}, "absolute", {}, False),
            ("missing_directory_noop", {"any": token + "/missing"}, None, {}, False),
            ("configured_file_refused", {"any": relative_a + "/queue.db"}, None, {"EXPECTED_STATUS": 1}, False),
            ("local_relative_overrides_any", {"any": absolute, server: relative_b}, "named", {}, False),
            ("local_absolute_overrides_any", {"any": relative_a, server: absolute}, "absolute", {}, False),
            ("other_server_ignored", {"any": relative_a, other: relative_b}, "global", {}, False),
            ("named_empty_disables_any", {"any": relative_a, server: ""}, None, {}, False),
            ("p4name_selects_scope", {"any": relative_a, server: relative_b, other: absolute}, "absolute", {"P4NAME": other}, False),
            ("startup_In_selects_scope", {"any": relative_a, server: relative_b, other: absolute}, "absolute", {"P4NAME": server, "P4D_FLAGS": "-In " + other}, False),
            ("database_p4name_one_hop", {server: relative_a, other: relative_b, third: absolute}, "named", {"DB_P4NAME": {server: other, other: third}}, False),
            ("global_database_p4name_redirect", {server: relative_a, other: relative_b, third: absolute}, "named", {"DB_P4NAME": {"any": other, other: third}}, False),
            ("named_p4name_overrides_global_redirect", {server: relative_a, other: relative_b, third: absolute}, "absolute", {"DB_P4NAME": {"any": other, server: third}}, False),
            ("startup_In_bypasses_database_redirect", {server: relative_a, other: relative_b}, "global", {"P4D_FLAGS": "-In " + server, "DB_P4NAME": {server: other}}, False),
            ("running_server_refused", {server: relative_b}, None, {}, True),
        ]
        checks = []
        service = "p4d_" + self.args.instance + ".service"
        try:
            for name, settings, selected, identity, running in cases:
                self.replace_client_config("client.sendq.dir", settings)
                trees = {key: self.seed_sendq_tree(path, scratch / ("outside-" + key))
                         for key, path in {"global": root / relative_a, "named": root / relative_b,
                                           "absolute": Path(absolute)}.items()}
                alias = root / token / "root link"
                alias.symlink_to(absolute, target_is_directory=True)
                db_names = self.client_config("P4NAME")
                changed_names = []
                stopped = False
                try:
                    if not running:
                        self.command(["sudo", "-n", "systemctl", "stop", service])
                        stopped = True
                        if self.command(["systemctl", "is-active", "--quiet", service], check=False).returncode == 0:
                            raise RuntimeError("Service remained active before send-queue cleanup")
                    for scope, value in identity.get("DB_P4NAME", {}).items():
                        self.command([self.env["P4DBIN"], "-r", self.env["P4ROOT"], "-J", "off",
                                      "-cset " + scope + "#P4NAME=" + value])
                        changed_names.append(scope)
                    result = self.sendq_helper(identity=identity)
                    expected = identity.get("EXPECTED_STATUS", 1 if running else 0)
                    if result.returncode != expected:
                        log = self.logs / "sendq-cleanup-live.log"
                        detail = log.read_text(errors="replace")[-5000:] if log.exists() else ""
                        raise RuntimeError(name + ": cleanup exit %s, expected %s\n%s" % (result.returncode, expected, detail))
                    for key, tree in trees.items():
                        self.assert_sendq_tree(tree, key == selected)
                finally:
                    for scope in changed_names:
                        option = ("-cset " + scope + "#P4NAME=" + db_names[scope]
                                  if scope in db_names else "-cunset " + scope + "#P4NAME")
                        self.command([self.env["P4DBIN"], "-r", self.env["P4ROOT"], "-J", "off", option])
                    if stopped:
                        self.command(["sudo", "-n", "systemctl", "start", service])
                        self.command([BIN / "p4login", self.args.instance])
                self.fixture.assert_unchanged()
                checks.append({"name": name, "status": "passed", "server_was_stopped": not running})
                (self.logs / "sendq-cleanup-evidence.json").write_text(json.dumps({"status": "incomplete", "checks": checks}, indent=2) + "\n")
                alias.unlink()
                for tree in trees.values():
                    shutil.rmtree(tree["root"])
                for key in trees:
                    shutil.rmtree(scratch / ("outside-" + key))
            checks.extend(self.shared_sendq_checks(token))
            checks.extend(self.refused_sendq_alias_checks(token))
            evidence = {"checks": checks, "regular_files_and_symlinks_removed": True,
                        "directories_and_fifo_preserved": True, "nested_symlink_targets_preserved": True,
                        "fixture_records_and_bytes_preserved": True}
            (self.logs / "sendq-cleanup-evidence.json").write_text(json.dumps(evidence, indent=2) + "\n")
            return evidence
        finally:
            self.replace_client_config("client.sendq.dir", original)
            # Retain failed-case fixtures and logs. Only empty successful scratch
            # roots are removed; a populated path is valuable failure evidence.
            for path in (root / token, scratch):
                if path.is_dir() and not any(path.iterdir()):
                    path.rmdir()

    def recovery_switch(self):
        if not self.records or self.records[0]["status"] != "passed":
            raise RuntimeError("Checkpoint preparation failed; recovery switch was not attempted")
        before = Path(self.env["P4ROOT"]).resolve()
        replacement = (self.home / "offline_db").resolve()
        original = self.client_config("client.sendq.dir")
        original_readonly = self.client_config("client.readonly.dir")
        relative = ".sdp-sendq-switch-" + str(time.time_ns())
        outside = self.home / "tmp" / relative
        trees = [self.seed_sendq_tree(path / relative, outside / str(number))
                 for number, path in enumerate((before, replacement))]
        readonly = relative + "-readonly/readonly = clients"
        client = "sendq." + self.fixture.name + "-switch"
        workspace = outside / "partitioned workspace"
        workspace.mkdir(parents=True)
        created = False
        try:
            self.replace_client_config("client.sendq.dir", {self.env["SERVERID"]: relative})
            self.replace_client_config("client.readonly.dir", {self.env["SERVERID"]: readonly})
            self.fixture.form("client", {"Client": client, "Owner": self.env["P4USER"],
                              "Root": str(workspace), "Type": "partitioned", "LineEnd": "unix",
                              "View0": "//%s/... //%s/..." % (self.fixture.name, client)})
            created = True
            self.command([self.env["P4BIN"], "-c", client, "sync", "//" + self.fixture.name + "/..."])
            have_command = [self.env["P4BIN"], "-c", client, "-ztag", "have", "//" + self.fixture.name + "/..."]
            before_have = self.command(have_command).stdout
            if "... haveRev " not in before_have or str(workspace) not in before_have:
                raise RuntimeError("Recovery fixture has no real partitioned-client have state")
            if not any((before / readonly).rglob("db.sendq.*")):
                raise RuntimeError("Partitioned have table is not under the configured readonly path")
            self.script("recreate_db_checkpoint.sh")
            if self.command(have_command).stdout != before_have:
                raise RuntimeError("Database switch lost real partitioned-client have revisions")
            for tree in trees:
                self.assert_sendq_tree(tree, True)
        finally:
            # Failed recovery may leave the service stopped and recovery data.
            # Restore through the API only when the service is actually running.
            if self.command(["systemctl", "is-active", "--quiet", "p4d_" + self.args.instance + ".service"], check=False).returncode == 0:
                if created:
                    self.fixture.p4("client", "-d", "-f", client)
                self.replace_client_config("client.readonly.dir", original_readonly)
                self.replace_client_config("client.sendq.dir", original)
        after = Path(self.env["P4ROOT"]).resolve()
        if before == after:
            raise RuntimeError("Recovery did not switch physical database roots")
        value = self.command([self.env["P4BIN"], "counter", self.counter]).stdout.strip()
        if value != "12345":
            raise RuntimeError("Recovery switch did not preserve the live integration counter")
        self.command(["systemctl", "is-active", "--quiet", "p4d_" + self.args.instance + ".service"])
        self.fixture.assert_unchanged()
        self.fixture.post_recovery_write()
        if list(before.glob(".sdp-switch.*")):
            raise RuntimeError("Successful switch left an unfinished transaction directory")
        for tree in trees:
            shutil.rmtree(tree["root"])
        shutil.rmtree(outside)
        return {"sendq_physical_roots_cleaned": [str(before), str(replacement)],
                "both_physical_roots_verified": True, "real_partitioned_have_preserved": True,
                "readonly_path_contains_space_and_equals": True, "fixture_records_and_bytes_preserved": True}

    def ssh_rsync(self):
        sshd = shutil.which("sshd") or "/usr/sbin/sshd"
        if not Path(sshd).is_file():
            raise RuntimeError("openssh-server is required for real host-trust testing")
        with tempfile.TemporaryDirectory(prefix="sdp-ssh-", dir=self.home / "tmp") as temp:
            work = Path(temp)
            for name in ["host", "client", "changed-host"]:
                self.command(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", work / name])
            config = work / "sshd_config"
            config.write_text("\n".join([
                "Port 2222", "ListenAddress 127.0.0.1", "HostKey " + str(work / "host"),
                "PidFile " + str(work / "sshd.pid"), "AuthorizedKeysFile " + str(work / "client.pub"),
                "PasswordAuthentication no", "KbdInteractiveAuthentication no", "UsePAM no",
                "StrictModes yes", "AllowUsers " + self.args.os_user, "LogLevel VERBOSE", ""]))
            ssh_folder = Path(pwd.getpwuid(os.getuid()).pw_dir) / ".ssh"
            ssh_folder.mkdir(mode=0o700, exist_ok=True)
            client_config = ssh_folder / "config"
            original_config = client_config.read_bytes() if client_config.exists() else None
            original_mode = client_config.stat().st_mode & 0o777 if client_config.exists() else None
            instance_vars = Path("/p4/common/config/p4_" + self.args.instance + ".vars")
            original_vars = instance_vars.read_bytes()
            known_hosts = work / "known_hosts"
            daemon = None
            ssh_log = None
            try:
                client_config.write_text("Host 127.0.0.1\n  Port 2222\n  IdentityFile %s\n  UserKnownHostsFile %s\n  IdentitiesOnly yes\n  BatchMode yes\n" % (work / "client", known_hosts))
                client_config.chmod(0o600)
                with instance_vars.open("a") as output:
                    output.write("\nexport P4MASTERHOST=127.0.0.1\n")
                ssh_log = (work / "sshd.log").open("w")
                daemon = subprocess.Popen([sshd, "-D", "-e", "-f", str(config)], stdout=ssh_log, stderr=subprocess.STDOUT)
                for _ in range(60):
                    if daemon.poll() is not None:
                        raise RuntimeError("Fixture sshd failed: " + (work / "sshd.log").read_text())
                    try:
                        with socket.create_connection(("127.0.0.1", 2222), timeout=0.1):
                            break
                    except OSError:
                        time.sleep(0.1)
                else:
                    raise RuntimeError("Fixture sshd did not become ready")
                filename = 'sdp runtime $(touch forbidden)`touch forbidden2` quotes\'".txt'
                archive = self.home / "depots/depot" / (filename + ",d")
                archive.mkdir(parents=True, exist_ok=True)
                (archive / "1.1").write_text("literal archive payload\n")
                verify = work / "verify_output.log"
                verify.write_text("info: //depot/" + filename + "#1 - MISSING!\n")
                env = dict(self.env)
                env.pop("RSYNC_RSH", None)
                argv = ["bash", BIN / "rsync_missing.sh", self.args.instance, verify]
                self.command(argv, env=env)
                if not known_hosts.is_file() or not known_hosts.read_text().strip():
                    raise RuntimeError("First SSH connection did not record the host key")
                if (self.home / "forbidden").exists() or (self.home / "forbidden2").exists():
                    raise RuntimeError("Filename content was interpreted by a shell")
                host = known_hosts.read_text().split()[0]
                changed_key = (work / "changed-host.pub").read_text().split()
                known_hosts.write_text(host + " " + " ".join(changed_key[:2]) + "\n")
                rejected = self.command(argv, check=False, env=env)
                if rejected.returncode == 0:
                    raise RuntimeError("Changed SSH host key was incorrectly accepted")
                if (archive / "1.1").read_text() != "literal archive payload\n":
                    raise RuntimeError("Host-key failure changed archive data")
            finally:
                if daemon is not None:
                    daemon.terminate()
                    try:
                        daemon.wait(timeout=10)
                    except subprocess.TimeoutExpired:
                        daemon.kill()
                        daemon.wait()
                if ssh_log is not None:
                    ssh_log.close()
                instance_vars.write_bytes(original_vars)
                if original_config is None:
                    client_config.unlink(missing_ok=True)
                else:
                    client_config.write_bytes(original_config)
                    client_config.chmod(original_mode)

    def run(self):
        self.scenario("checkpoint_restore", ["live_checkpoint.sh", "daily_checkpoint.sh", "recreate_offline_db.sh", "backup_functions.sh"],
                      "Real checkpoint/offline replay preserve nine revision records (including delete/move history) and seven exact content hashes, client/group/label specs, label members and shelf bytes, inspected through a separately served restored database.", self.checkpoint_cycle)
        self.scenario("journal_rotation", ["rotate_journal.sh", "backup_functions.sh"],
                      "Three real rotations advance the counter and preserve the existing >2 gzip threshold and newest-journal window.", self.rotation)
        self.scenario("sendq_cleanup", ["backup_functions.sh"],
                      "Actual local db.config settings select dedicated relative/absolute queues; stopped cleanup preserves directories, special entries, nested symlink targets and exact fixture bytes, and refuses the running server.", self.sendq_cleanup)
        self.scenario("database_switch", ["recreate_db_checkpoint.sh", "backup_functions.sh", "p4d_base"],
                      "Real systemd stop/start and database-root swap clean queues in both physical roots, preserve exact fixture records and bytes, and allow a new post-recovery submit; staging is cleaned.", self.recovery_switch)
        self.scenario("ssh_host_trust", ["rsync_missing.sh"],
                      "Real localhost SSH/rsync records a first host key, rejects a changed key, and treats archive names literally; no multi-host replica topology.", self.ssh_rsync)
        print(json.dumps({"scenarios": self.records}, indent=2))
        return int(any(record["status"] != "passed" for record in self.records))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", required=True)
    parser.add_argument("--instance", required=True)
    parser.add_argument("--port", required=True, type=int)
    parser.add_argument("--os-user", required=True)
    parser.add_argument("--confirm-disposable", action="store_true")
    try:
        return RuntimeIntegration(parser.parse_args()).run()
    except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
        print(json.dumps({"guard_error": str(error), "scenarios": []}, indent=2))
        return 2


if __name__ == "__main__":
    raise SystemExit(main())