test_service_lifecycle.py #1

#!/usr/bin/env python3
"""Exercise broker/proxy lifecycle failures and cache authentication offline."""
from pathlib import Path
import os
import shutil
import subprocess
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"))


class ServiceLifecycle(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-service-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.common = self.root / "common"
        self.common.mkdir()
        self.env = dict(os.environ, TEST_ROOT=str(self.root))
        (self.common / "p4_vars").write_text('''
P4CBIN="$TEST_ROOT/common"
P4CCFG="$TEST_ROOT/common"
P4SERVER=p4_1
P4MASTER_ID=master
P4BROKERBIN="$TEST_ROOT/daemon"
P4PBIN="$TEST_ROOT/daemon"
OSUSER=perforce
LOGS="$TEST_ROOT"
PROXY_TARGET=127.0.0.1:19999
PROXY_PORT=19998
PROXY_MON_LEVEL=1
P4PCACHE="$TEST_ROOT/cache"
P4SERVICEUSER=svc
GREP=grep
''')
        (self.common / "p4_1.broker.master.cfg").write_text('target = "127.0.0.1:19999";\n')
        (self.common / "ps_functions.sh").write_text('''
# Fake processes only: no real signals or accounts are used.
id () { if [[ $* == '-u' ]]; then echo 1000; else echo perforce; fi; }
get_pids () { [[ ${RUNNING:-0} == 1 ]] && echo 999999; return 0; }
kill () { echo attempted >> "$TEST_ROOT/signals"; }
sleep () { :; }
''')
        self.executable(self.root / "daemon", 'echo started >> "$TEST_ROOT/starts"\nexit "${START_RC:-0}"\n')

    def executable(self, path, body):
        path.write_text("#!" + BASH + "\n" + body)
        path.chmod(0o755)

    def fixture(self, name):
        target = self.root / name
        self.executable(target, (BIN / name).read_text().replace("/p4/common/bin/", str(self.common) + "/"))
        return target

    def run_service(self, path, operation, **env):
        return subprocess.run([BASH, str(path), "1", operation], env=dict(self.env, **env),
                              capture_output=True, text=True, timeout=10)

    def test_start_failure_is_reported_for_both_services(self):
        for name in ("p4broker_base", "p4p_base"):
            with self.subTest(name=name):
                result = self.run_service(self.fixture(name), "start", START_RC="7")
                self.assertEqual(result.returncode, 7, result.stdout + result.stderr)

    def test_stop_timeout_never_claims_success(self):
        for name in ("p4broker_base", "p4p_base"):
            with self.subTest(name=name):
                result = self.run_service(self.fixture(name), "stop", RUNNING="1")
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertIn("did not stop after 60 seconds", result.stderr)
                self.assertNotIn(" stopped.", result.stdout)

    def test_restart_never_starts_when_old_process_cannot_stop(self):
        for name in ("p4broker_base", "p4p_base"):
            with self.subTest(name=name):
                result = self.run_service(self.fixture(name), "restart", RUNNING="1")
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertFalse((self.root / "starts").exists())

    def test_restart_stopped_service_starts_and_reports_failure(self):
        for name in ("p4broker_base", "p4p_base"):
            for rc in (0, 7):
                with self.subTest(name=name, rc=rc):
                    result = self.run_service(self.fixture(name), "restart", START_RC=str(rc))
                    self.assertEqual(result.returncode, rc, result.stdout + result.stderr)
                    self.assertTrue((self.root / "starts").exists())

    def test_cache_purge_stops_after_failed_login(self):
        with (self.common / "p4_vars").open("a") as output:
            output.write('P4BIN="$TEST_ROOT/p4"\nP4PORT=127.0.0.1:19999\nP4USER=admin\nKEEPLOGS=7\n')
        (self.common / "backup_functions.sh").write_text('''
check_vars () { :; }; set_vars () { :; }; get_journalnum () { :; }
log_init () { :; }; sdp_log_init () { :; }; rotate_log_file () { :; }; remove_log_files () { :; }
die () { echo "$*" >&2; exit 1; }
''')
        self.executable(self.common / "p4login", 'echo login-denied >&2\nexit 5\n')
        self.executable(self.root / "p4", 'touch "$TEST_ROOT/purged"\n')
        result = self.run_service(self.fixture("p4cachepurge.sh"), "unused")
        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
        self.assertFalse((self.root / "purged").exists())
        self.assertIn("login-denied", (self.root / "p4cachepurge.log").read_text())
        self.assertIn("Cache purge login failed", result.stderr)


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