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