test_p4d_tracking.py #2

#!/usr/bin/env python3
"""Offline tracking/lifecycle checks; no real daemon, signals or account changes.

Execute the shipped/generated wrappers and the actual init/base script logic.
Only environment setup, logging helpers and external processes are fixtures.
These argv assertions do not substitute for a live server tracking-config check.
"""
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

SDP = Path(__file__).resolve().parents[1]
BIN = SDP / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))
MARKER = "SDP_P4D_SERVICE_START"


class TrackingFixture(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix="sdp-p4d-tracking-")
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name)
        self.record = self.root / "calls.jsonl"
        self.binary = self.root / "p4d_1_bin"
        self.env = dict(os.environ, TEST_ROOT=str(self.root),
                        TEST_RECORD=str(self.record), TEST_CHILD_RC="0")
        self.env.pop(MARKER, None)
        self.write_executable(self.binary, "#!" + sys.executable + "\n" + '''
import json, os, pathlib, sys
args = sys.argv[1:]
root = pathlib.Path(os.environ["TEST_ROOT"])
launch = "-p" in args
row = {"kind": "launch" if launch else "offline", "args": args,
       "marker": os.environ.get("SDP_P4D_SERVICE_START")}
with pathlib.Path(os.environ["TEST_RECORD"]).open("a") as output:
    output.write(json.dumps(row) + "\\n")
if "-V" in args:
    print("Rev. P4D/LINUX26AARCH64/2026.1/1234567 (2026/10/01).")
if os.environ.get("TEST_FAIL_OPERATION") in args:
    raise SystemExit(6)
status = int(os.environ.get("TEST_START_RC" if launch else "TEST_CHILD_RC", "0"))
if launch and status == 0:
    (root / "alive").write_text("fake process only\\n")
    (root / "db/server.pid").write_text("99999999\\n") if (root / "db").is_dir() else None
raise SystemExit(status)
''')

    @staticmethod
    def write_executable(path, text):
        path.write_text(text)
        path.chmod(0o755)
        return path

    def calls(self):
        return [json.loads(line) for line in self.record.read_text().splitlines()] if self.record.exists() else []

    def sample_wrapper(self):
        source = (SDP / "Server/Unix/p4/1/bin/p4d_1").read_text()
        source = source.replace("/p4/common/bin/p4d_1_bin", str(self.binary))
        return self.write_executable(self.root / "sample-wrapper", source)

    def generated_wrapper(self, case_sensitive):
        source = (SDP / "Server/Unix/setup/mkdirs.sh").read_text()
        start = source.index("if (( CASE_SENSITIVE == 1 )); then")
        end = source.index("# Plain-named 'p4'/'p4d'", start)
        directory = self.root / ("generated-" + str(case_sensitive))
        directory.mkdir(exist_ok=True)
        result = subprocess.run([BASH, "-c", source[start:end]], cwd=directory,
                                env=dict(self.env, CASE_SENSITIVE=str(case_sensitive), SDP_INSTANCE="1"),
                                capture_output=True, text=True, timeout=10)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        wrapper = directory / "p4d_1"
        self.assertTrue(os.access(wrapper, os.X_OK))
        # Keep generated behavior; relocate only its raw binary for this fixture.
        text = wrapper.read_text().replace("/p4/1/bin/p4d_1_bin", str(self.binary))
        # Nested init calls must exercise the requested Bash version too.
        text = text.replace("#!/bin/bash\n", "#!" + BASH + "\n", 1)
        self.write_executable(wrapper, text)
        return wrapper

    def invoke(self, wrapper, args, marker=None, **extra_env):
        self.record.unlink(missing_ok=True)
        env = dict(self.env, **extra_env)
        if marker is not None:
            env[MARKER] = marker
        return subprocess.run([BASH, wrapper, *args], env=env, capture_output=True,
                              text=True, timeout=10)


class WrapperTracking(TrackingFixture):
    def test_sample_wrapper_defaults_quiet_and_preserves_case_argv_and_status(self):
        args = ["-r", "path with spaces", "-jr", "$(literal); 'quoted'", "-vtrack=3"]
        for marker in (None, "", "0", "yes"):
            with self.subTest(marker=marker):
                result = self.invoke(self.sample_wrapper(), args, marker, TEST_CHILD_RC="9")
                self.assertEqual(result.returncode, 9, result.stdout + result.stderr)
                self.assertEqual(self.calls(), [{"kind": "offline", "args": ["-C1", "-vtrack=0", *args], "marker": None}])

    def test_sample_service_mode_omits_only_default_tracking_and_consumes_marker(self):
        args = ["-r", "space path", "-v", "track=2", "-V"]
        result = self.invoke(self.sample_wrapper(), args, "1", TEST_CHILD_RC="7")
        self.assertEqual(result.returncode, 7, result.stdout + result.stderr)
        self.assertEqual(self.calls(), [{"kind": "offline", "args": ["-C1", *args], "marker": None}])

    def test_generated_wrappers_preserve_case_mode_and_tracking_selection(self):
        args = ["-r", "database with spaces", "-jd", "-", "literal $(no-command)"]
        for case_sensitive in (0, 1):
            wrapper = self.generated_wrapper(case_sensitive)
            for marker in (None, "1"):
                with self.subTest(case_sensitive=case_sensitive, marker=marker):
                    result = self.invoke(wrapper, args, marker, TEST_CHILD_RC="8")
                    self.assertEqual(result.returncode, 8, result.stdout + result.stderr)
                    prefix = ([] if case_sensitive else ["-C1"]) + ([] if marker == "1" else ["-vtrack=0"])
                    self.assertEqual(self.calls(), [{"kind": "offline", "args": prefix + args, "marker": None}])

    def test_generated_wrappers_leave_explicit_debug_arguments_unchanged(self):
        for case_sensitive in (0, 1):
            wrapper = self.generated_wrapper(case_sensitive)
            for marker in (None, "1"):
                for args in (["-vtrack=4", "-V"], ["-v", "track=5,server=3", "-V"]):
                    with self.subTest(case_sensitive=case_sensitive, marker=marker, args=args):
                        result = self.invoke(wrapper, args, marker)
                        self.assertEqual(result.returncode, 0, result.stderr)
                        prefix = ([] if case_sensitive else ["-C1"]) + ([] if marker == "1" else ["-vtrack=0"])
                        self.assertEqual(self.calls()[0]["args"], prefix + args)
                        self.assertIsNone(self.calls()[0]["marker"])

    def test_systemd_environment_alone_does_not_disable_quiet_defaults(self):
        for wrapper in (self.sample_wrapper(), self.generated_wrapper(0), self.generated_wrapper(1)):
            with self.subTest(wrapper=wrapper):
                result = self.invoke(wrapper, [], OS_INIT_MECHANISM="systemd")
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("-vtrack=0", self.calls()[0]["args"])
                self.assertIsNone(self.calls()[0]["marker"])


class ServiceLaunchTracking(TrackingFixture):
    def setUp(self):
        super().setUp()
        self.common = self.root / "common"
        self.common.mkdir()
        self.db = self.root / "db"
        self.db.mkdir()
        (self.db / "db.domain").write_text("fixture database: never opened by p4d\n")
        (self.root / "journal").write_text("@pv@ 0 @db.counters@ @journal@ 1\n")
        self.wrapper = self.generated_wrapper(0)
        (self.common / "p4_vars").write_text('''
export P4CBIN="$TEST_ROOT/common" P4ROOT="$TEST_ROOT/db" LOGS="$TEST_ROOT"
export P4DBIN="$TEST_WRAPPER" P4BIN="$TEST_ROOT/common/p4" P4TMP="$TEST_ROOT"
export P4JOURNAL="$TEST_ROOT/journal" P4D_VERSION=2026.1 P4REPLICA=FALSE
export P4PORT=127.0.0.1:19999 P4USER=fixture OSUSER=fixture
export PS="$TEST_ROOT/common/ps" SDP_START_DELAY=0
export P4D_FLAGS="-p $P4PORT -r $P4ROOT -J $P4JOURNAL -L $LOGS/log -q -d --pid-file ${TEST_EXPLICIT_TRACK:-}"
# A real p4_vars load probes this same wrapper before the service branch.
"$P4DBIN" -V >/dev/null || return 1
''')
        (self.common / "backup_functions.sh").write_text('''
set_vars () { :; }
sdp_log_init () { :; }
id () {
    if [[ $* == '-u' ]]; then
        if [[ ${TEST_ROOT_ENTRY:-0} == 1 && ${TEST_DROPPED:-0} != 1 ]]; then echo 0; else echo 1000; fi
    else echo fixture; fi
}
sleep () { :; }
kill () {
    printf '%s\\n' "$*" >> "$TEST_ROOT/signals"
    if [[ ${TEST_IGNORE_SIGNAL:-0} != 1 ]]; then
        rm -f "$TEST_ROOT/alive"
    fi
}
tail () {
    if [[ $1 == --version ]]; then printf 'fixture version capability\\n'; return 0; fi
    command tail "$@"
}
''')
        # p4d_base uses `exec su`, which requires an executable PATH stub,
        # not a shell function. Never invoke the host account-switch command.
        self.write_executable(self.common / "su", "#!" + BASH + '''
[[ $1 == - && $2 == fixture && $3 == -c ]] || exit 88
printf 'simulated login-shell account change\\n' >> "$TEST_ROOT/account-change"
unset SDP_P4D_SERVICE_START
export TEST_DROPPED=1
exec "$BASH" -c "$4"
''')
        self.write_executable(self.common / "ps", "#!" + sys.executable + "\n" + '''
import os, pathlib
root = pathlib.Path(os.environ["TEST_ROOT"])
if (root / "signals").exists():
    counter = root / "polls-after-signal"
    count = (int(counter.read_text()) if counter.exists() else 0) + 1
    counter.write_text(str(count))
    stop_at = int(os.environ.get("TEST_EXIT_AFTER_POLLS", "0"))
    if stop_at and count >= stop_at:
        (root / "alive").unlink(missing_ok=True)
raise SystemExit(0 if (root / "alive").exists() else 1)
''')
        self.write_executable(self.common / "p4login", "#!" + BASH + "\nexit 0\n")
        self.write_executable(self.common / "p4", "#!" + sys.executable + "\n" + '''
import json, os, pathlib, sys
root = pathlib.Path(os.environ["TEST_ROOT"])
args = sys.argv[1:]
with pathlib.Path(os.environ["TEST_RECORD"]).open("a") as output:
    output.write(json.dumps({"kind": "client", "args": args, "marker": os.environ.get("SDP_P4D_SERVICE_START")}) + "\\n")
if args[-2:] == ["admin", "stop"]:
    status = int(os.environ.get("TEST_ADMIN_STOP_RC", "0"))
    if status:
        raise SystemExit(status)
    (root / "alive").unlink(missing_ok=True)
elif "info" in args:
    raise SystemExit(0 if (root / "alive").exists() else 1)
''')
        # The override permits the identical regressions to run against the
        # captured submitted script as well as the candidate implementation.
        base_path = Path(os.environ.get("SDP_TEST_P4D_BASE", BIN / "p4d_base"))
        base = base_path.read_text().replace("/p4/common/bin/", str(self.common) + "/")
        base = base.replace("#!/bin/bash\n", "#!" + BASH + "\n", 1)
        self.write_executable(self.common / "p4d_base", base)
        template = (SDP / "Server/Unix/p4/common/etc/init.d/p4d_instance_init.template").read_text()
        template = template.replace("REPL_SDP_INSTANCE", "1").replace("/p4/common/bin/", str(self.common) + "/")
        self.init = self.write_executable(self.root / "p4d_1_init", template)
        self.env["TEST_WRAPPER"] = str(self.wrapper)
        self.env["PATH"] = str(self.common) + os.pathsep + os.environ["PATH"]

    def run_service(self, operation, **env):
        self.record.unlink(missing_ok=True)
        return subprocess.run([BASH, self.init, operation], env=dict(self.env, **env),
                              capture_output=True, text=True, timeout=15)

    def assert_selection(self, expected_launches=1):
        calls = self.calls()
        launches = [call for call in calls if call["kind"] == "launch"]
        self.assertEqual(len(launches), expected_launches, calls)
        for call in calls:
            self.assertIsNone(call["marker"], call)
            if call["kind"] == "client":
                continue
            self.assertEqual(call["args"][0], "-C1", call)
            if call["kind"] == "launch":
                self.assertNotIn("-vtrack=0", call["args"], call)
            else:
                self.assertEqual(call["args"][1], "-vtrack=0", call)
        return calls

    def mark_running(self):
        (self.root / "alive").write_text("fake process only\n")
        (self.db / "server.pid").write_text("99999999\n")

    def test_start_marks_only_daemon_and_keeps_all_offline_preflights_quiet(self):
        result = self.run_service("start", TEST_EXPLICIT_TRACK="-vtrack=2")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        calls = self.assert_selection()
        args = [call["args"] for call in calls if call["kind"] == "offline"]
        for flag in ("-V", "-xvU", "-jr", "-xu"):
            self.assertTrue(any(flag in entry for entry in args), (flag, args))
        self.assertTrue(any(any(arg.startswith("-cset ") for arg in entry) for entry in args), args)
        self.assertIn("-vtrack=2", next(call for call in calls if call["kind"] == "launch")["args"])

    def test_force_start_uses_service_mode_without_integrity_preflights(self):
        result = self.run_service("force_start")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        calls = self.assert_selection()
        self.assertEqual([call["args"] for call in calls if call["kind"] == "offline"], [["-C1", "-vtrack=0", "-V"]])

    def test_restart_preserves_stop_path_and_marks_only_new_daemon(self):
        self.mark_running()
        result = self.run_service("restart")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assert_selection()
        self.assertEqual((self.root / "signals").read_text().strip(), "99999999")
        self.assertTrue((self.root / "alive").exists())

    def test_inherited_service_marker_cannot_bypass_preflight_tracking(self):
        result = self.run_service("start", SDP_P4D_SERVICE_START="1", OS_INIT_MECHANISM="systemd")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        calls = self.assert_selection()
        self.assertEqual(calls[0]["args"], ["-C1", "-vtrack=0", "-V"])

    def test_root_reentry_sets_service_marker_after_login_shell_environment_reset(self):
        result = self.run_service("start", SDP_P4D_SERVICE_START="1", TEST_ROOT_ENTRY="1")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertTrue((self.root / "account-change").exists())
        calls = self.assert_selection()
        self.assertEqual(sum(call["args"][-1:] == ["-V"] for call in calls), 2)

    def test_service_launch_failure_retains_child_exit_status(self):
        for operation in ("start", "force_start"):
            with self.subTest(operation=operation):
                result = self.run_service(operation, TEST_START_RC="7")
                self.assertEqual(result.returncode, 7, result.stdout + result.stderr)
                self.assert_selection()
                self.assertFalse((self.root / "alive").exists())

    def test_restart_returns_exact_start_failure_after_confirmed_stop(self):
        self.mark_running()
        result = self.run_service("restart", TEST_START_RC="7",
                                  SDP_P4D_SERVICE_START="1")
        self.assertEqual(result.returncode, 7, result.stdout + result.stderr)
        self.assert_selection()
        self.assertIn("Confirmed shutdown", result.stdout)
        self.assertEqual((self.root / "signals").read_text().strip(), "99999999")
        self.assertFalse((self.root / "alive").exists())

    def test_stale_pid_stop_failure_prevents_restart(self):
        (self.db / "server.pid").write_text("99999999\n")
        result = self.run_service("restart", SDP_P4D_SERVICE_START="1")
        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
        calls = self.assert_selection(expected_launches=0)
        self.assertTrue(all(call["args"] == ["-C1", "-vtrack=0", "-V"]
                            for call in calls), calls)
        self.assertFalse((self.root / "alive").exists())
        self.assertFalse((self.root / "signals").exists())

    def test_stop_timeout_fails_and_never_confirms_shutdown_or_restarts(self):
        for operation in ("stop", "restart"):
            with self.subTest(operation=operation):
                self.mark_running()
                (self.root / "polls-after-signal").unlink(missing_ok=True)
                (self.root / "p4d_init.log").unlink(missing_ok=True)
                result = self.run_service(operation, TEST_IGNORE_SIGNAL="1",
                                          SDP_P4D_SERVICE_START="1")
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assert_selection(expected_launches=0)
                self.assertTrue((self.root / "alive").exists())
                self.assertEqual((self.db / "server.pid").read_text(), "99999999\n")
                self.assertGreaterEqual(int((self.root / "polls-after-signal").read_text()), 61)
                self.assertNotIn("Confirmed shutdown", result.stdout + result.stderr)
                self.assertNotIn("Confirmed shutdown", (self.root / "p4d_init.log").read_text())

    def test_stop_succeeds_when_process_exits_on_last_wait_poll(self):
        self.mark_running()
        # One immediate post-signal check, then sixty wait-loop checks.
        result = self.run_service("stop", TEST_IGNORE_SIGNAL="1", TEST_EXIT_AFTER_POLLS="61")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assert_selection(expected_launches=0)
        self.assertFalse((self.root / "alive").exists())
        self.assertEqual((self.root / "polls-after-signal").read_text(), "61")
        self.assertIn("Confirmed shutdown", result.stdout)
        self.assertNotIn("did not stop", result.stdout + result.stderr)

    def test_missing_pid_admin_stop_failure_propagates_and_blocks_restart(self):
        for operation in ("stop", "restart"):
            with self.subTest(operation=operation):
                self.mark_running()
                (self.db / "server.pid").unlink()
                result = self.run_service(operation, TEST_ADMIN_STOP_RC="9",
                                          SDP_P4D_SERVICE_START="1")
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                calls = self.assert_selection(expected_launches=0)
                admin_calls = [call for call in calls if call["args"][-2:] == ["admin", "stop"]]
                self.assertEqual(len(admin_calls), 1, calls)
                self.assertTrue((self.root / "alive").exists())
                self.assertFalse((self.root / "signals").exists())
                self.assertNotIn("Confirmed shutdown", result.stdout + result.stderr)

    def test_already_stopped_restart_succeeds_and_preserves_tracking_scope(self):
        result = self.run_service("restart", SDP_P4D_SERVICE_START="1")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        calls = self.assert_selection()
        self.assertTrue((self.root / "alive").exists())
        self.assertFalse((self.root / "signals").exists())
        self.assertFalse(any(call["args"][-2:] == ["admin", "stop"] for call in calls))

    def test_failed_offline_preflight_never_launches_daemon(self):
        result = self.run_service("start", TEST_FAIL_OPERATION="-xvU")
        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
        self.assert_selection(expected_launches=0)
        self.assertFalse((self.root / "alive").exists())

    def test_stop_and_admin_stop_never_start_a_p4d_process(self):
        for operation in ("stop", "admin_stop"):
            with self.subTest(operation=operation):
                self.mark_running()
                result = self.run_service(operation, SDP_P4D_SERVICE_START="1")
                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
                self.assert_selection(expected_launches=0)
                self.assertFalse((self.root / "alive").exists())


if __name__ == "__main__":
    unittest.main()