test_setup_improvements.py #1

#!/usr/bin/env python3
"""Offline checks: no real server, reset, authentication, mail or downloads."""
import contextlib
import datetime
import io
import os
from pathlib import Path
import re
import runpy
import shlex
import shutil
import subprocess
import sys
import tempfile
import types
import unittest
from unittest import mock

SDP = Path(__file__).resolve().parents[1]
SETUP = SDP / "Server/Unix/setup"
TRIGGERS = SDP / "Server/Unix/p4/common/bin/triggers"
TEST = SDP / "test"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


def shell_function(path, name):
    source = path.read_text()
    match = re.search(r"^(?:function )?" + re.escape(name) + r"(?:\(\))?\s*\n?\{", source, re.M)
    if not match:
        raise AssertionError("Missing shell function " + name)
    return source[match.start():source.index("\n}", match.end()) + 2]


class SetupImprovements(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-improvement-tests-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name).resolve()

    def execute(self, code, env=None):
        return subprocess.run([BASH, "-c", code], text=True, capture_output=True, env=env, timeout=10)

    def executable(self, path, source):
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(source)
        path.chmod(0o755)
        return path

    def test_edge_daily_skips_exactly_first_saturday_all_year(self):
        text = (SDP / "Server/Unix/p4/common/etc/cron.d/template.crontab").read_text()
        line = next(line for line in text.splitlines() if not line.startswith("#") and "run_if_edge.sh" in line and "daily_checkpoint.sh" in line)
        expression = line.split("&& { ", 1)[1].split("; } &&", 1)[0].replace(r"\%", "%")
        for day in (datetime.date(2026, 1, 1) + datetime.timedelta(days=n) for n in range(365)):
            code = 'date(){ case "$1" in +%u) echo ' + str(day.isoweekday()) + ';; +%d) echo ' + day.strftime("%d") + ';; esac; }; ' + expression
            result = self.execute(code)
            self.assertEqual(result.returncode == 0, not (day.isoweekday() == 6 and day.day <= 7), str(day))

    def test_tar_versions_compared_numerically(self):
        function = shell_function(SETUP / "install_sdp.sh", "version_at_least")
        for version, passes in [("1.9", False), ("1.29", False), ("1.30", True), ("1.100", True), ("2.0", True), ("3.10.1", True), ("bsd", False), ("01.030", True)]:
            result = self.execute(function + "\nversion_at_least " + shlex.quote(version) + " 1.30")
            self.assertEqual(result.returncode == 0, passes, version)

    def proxy_install_fixture(self, existing=None, preview=False):
        root = self.root / (existing or "new")
        instance = root / "alias/1"
        instance.mkdir(parents=True)
        binaries = root / "bin"
        binaries.mkdir()
        common = root / "source"
        self.executable(common / "bin/p4p", "#!/bin/sh\necho 'Rev. P4P/LINUX26AARCH64/2026.1/1234567 (2026/10/01).'\n")
        template = common / "etc/init.d/p4p_instance_init.template"
        template.parent.mkdir(parents=True)
        template.write_text("#!/bin/sh\n# REPL_SDP_INSTANCE REPL_DNSNAME\n")
        cache = instance / "cache"
        if existing == "directory":
            cache.mkdir()
            (cache / "owned-data").write_text("preserve")
        elif existing in ("symlink", "dangling-link"):
            target = root / "configured-cache"
            if existing == "symlink":
                target.mkdir()
            cache.symlink_to(target)
        source = (SETUP / "mkdirs.sh").read_text()
        block = source[source.index('# If proxy binary is available,'):source.index('# Generate p4d init script.')]
        env = dict(os.environ, SDP_COMMON=str(common), P4DIR=str(root / "alias"), DD=str(root / "data").lstrip("/"), SDP_INSTANCE="1", P4DNSNAME="fixture", AWK="awk", TEST=str(int(preview)))
        result = subprocess.run([BASH, "-c", block], cwd=binaries, env=env, text=True, capture_output=True)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        return cache, root / "data/p4/1/cache"

    def test_proxy_install_creates_usable_cache_alias(self):
        cache, physical = self.proxy_install_fixture()
        self.assertTrue(cache.is_symlink())
        (cache / "probe").write_bytes(b"proxy cache fixture")
        self.assertEqual((physical / "probe").read_bytes(), b"proxy cache fixture")

    def test_proxy_install_preserves_existing_cache_and_preview_paths(self):
        for existing in ("directory", "symlink", "dangling-link"):
            with self.subTest(existing=existing):
                cache, physical = self.proxy_install_fixture(existing)
                if existing == "directory":
                    self.assertFalse(cache.is_symlink())
                    self.assertEqual((cache / "owned-data").read_text(), "preserve")
                else:
                    self.assertEqual(cache.readlink(), cache.parent.parent.parent / "configured-cache")
                self.assertTrue(physical.is_dir())
        cache, physical = self.proxy_install_fixture("preview", preview=True)
        self.assertFalse(cache.exists() or cache.is_symlink())
        self.assertTrue(physical.is_dir())

    def test_initial_spec_archives_use_configured_depot_root(self):
        bindir = self.root / "bin"
        state = self.root / "active-root"
        state.write_text("root")
        archive = self.root / "archives"
        self.executable(bindir / "p4", "#!" + sys.executable + '\n' + '''
import os,sys
from pathlib import Path
args=sys.argv[1:]
if args[:2]==['server','-o']:
    print('Services:\\tstandard')
elif args[:3]==['-s','depot','-i']:
    data=sys.stdin.read()
    name=next(line.split(':',1)[1].strip() for line in data.splitlines() if line.startswith('Depot:'))
    path=Path(os.environ['ARCHIVES'])/Path(os.environ['CONFIG_STATE']).read_text()/name
    path.parent.mkdir(parents=True,exist_ok=True)
    path.write_text(data)
elif '-i' in args:
    sys.stdin.read()
''')
        self.executable(bindir / "ccheck.sh", '#!/bin/sh\nprintf depots > "$CONFIG_STATE"\n')
        self.executable(bindir / "p4login", "#!/bin/sh\nexit 0\n")
        source = (SETUP / "configure_new_server.sh").read_text()
        section = source[source.index('# Configure this server\'s spec as a commit-server'):source.index('if [[ "$DoCheckpoint" -eq 1 ]]')]
        env = dict(os.environ, PATH=str(bindir) + os.pathsep + os.environ["PATH"], P4BIN=str(bindir / "p4"), P4CBIN=str(bindir), P4MASTER_ID="commit", INSTANCE="1", CONFIG_STATE=str(state), ARCHIVES=str(archive))
        code = 'ErrorCount=0; WarningCount=0; msg(){ :; }; errmsg(){ ErrorCount=$((ErrorCount+1)); }; warnmsg(){ WarningCount=$((WarningCount+1)); }; stop_p4d(){ :; }; start_p4d(){ :; };\n' + section + '\nexit "$ErrorCount"'
        result = subprocess.run([BASH, "-c", code, str(SETUP / "configure_new_server.sh")], env=env, text=True, capture_output=True)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertTrue((archive / "depots/spec").is_file())
        self.assertTrue((archive / "depots/unload").is_file())
        self.assertFalse((archive / "root").exists(), "First specification archive was written before the archive root was configured")

    def test_runtime_plan_has_no_dependency_or_filesystem_side_effects(self):
        target = self.root / "python"
        result = subprocess.run([BASH, str(SETUP / "install_sdp_python.sh"), "-n", "-r", str(target), "-R", "r26.1"], text=True, capture_output=True)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("3.14.8", result.stdout)
        self.assertIn("2026.1", result.stdout)
        self.assertFalse(target.exists())

    def test_runtime_rejects_unsupported_pair_and_non_ssl(self):
        for args in [("-P", "3.6.2"), ("-R", "r17.1"), ("-no_ssl",), ("-w",)]:
            result = subprocess.run([BASH, str(SETUP / "install_sdp_python.sh"), "-n", *args], text=True, capture_output=True)
            self.assertNotEqual(result.returncode, 0, args)

    def installer_env(self, fail_download=False):
        bindir = self.root / "bin"
        self.executable(bindir / "uname", '#!/bin/sh\ncase "$1" in -s) echo Linux;; -m) echo x86_64;; esac\n')
        for tool in ("gcc", "g++", "openssl"):
            self.executable(bindir / tool, "#!/bin/sh\nexit 0\n")
        self.executable(bindir / "wget", '#!/bin/sh\n' + ('exit 28\n' if fail_download else 'printf fake > "$3"\n'))
        self.executable(bindir / "tar", '#!/bin/sh\ncase "$1" in -tzf) echo Python-3.14.8/;; -xzf) mkdir Python-3.14.8; printf "#!/bin/sh\\nexit 0\\n" > Python-3.14.8/configure; chmod +x Python-3.14.8/configure;; esac\n')
        self.executable(bindir / "make", '#!/bin/sh\nif [ "$1" = install ]; then\n mkdir -p "$FAKE_TARGET/bin"\n printf \'#!/bin/sh\\nprintf "%%s\\n" "$*" >> "$FAKE_PYTHON_CALLS"\\nexit 0\\n\' > "$FAKE_TARGET/bin/python3"\n chmod +x "$FAKE_TARGET/bin/python3"\nfi\n')
        return dict(os.environ, PATH=str(bindir) + os.pathsep + os.environ["PATH"], FAKE_TARGET=str(self.root / "python"), FAKE_PYTHON_CALLS=str(self.root / "python_calls"))

    def test_runtime_download_failure_preserves_existing_interpreter(self):
        env = self.installer_env(fail_download=True)
        old = self.executable(self.root / "python/bin/python3", "#!/bin/sh\necho old\n")
        result = subprocess.run([BASH, str(SETUP / "install_sdp_python.sh"), "-f", "-r", str(self.root / "python"), "-w", str(self.root / "work"), "-L", "off"], env=env, text=True, capture_output=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("echo old", old.read_text())

    def test_runtime_failure_replays_structured_logger_diagnostics(self):
        env = self.installer_env(fail_download=True)
        env["TMPDIR"] = str(self.root)
        script = self.root / "installer.sh"
        helper = SDP / "Server/Unix/p4/common/bin/sdp_structured_log.sh"
        script.write_text((SETUP / "install_sdp_python.sh").read_text().replace("/p4/common/bin/sdp_structured_log.sh", str(helper)))
        result = subprocess.run([BASH, str(script), "-r", str(self.root / "python"), "-w", str(self.root / "work"), "-L", str(self.root / "installer.log")], env=env, text=True, capture_output=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("Python source download failed", result.stdout + result.stderr)
        self.assertFalse(list(self.root.glob("sdp-stderr.*")))

    def test_runtime_mock_build_installs_only_compatible_wheel(self):
        env = self.installer_env()
        result = subprocess.run([BASH, str(SETUP / "install_sdp_python.sh"), "-r", str(self.root / "python"), "-w", str(self.root / "work"), "-L", "off"], env=env, text=True, capture_output=True)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        calls = (self.root / "python_calls").read_text()
        self.assertIn("--only-binary=:all: p4python==2026.1.*", calls)
        self.assertIn("import P4", calls)
        self.assertFalse((self.root / "work").exists())

    def test_tools_refuse_implicit_target_before_any_server_calls(self):
        for name in ("reset.sh", "sdp_test_harness.sh"):
            result = subprocess.run([BASH, str(TEST / name)], text=True, capture_output=True)
            self.assertEqual(result.returncode, 2, result.stdout + result.stderr)
            self.assertIn("--confirm-disposable", result.stderr)

    def guard_script(self, matching=True):
        target = self.root / "sdp-test-fixture"
        target.mkdir(exist_ok=True)
        (target / "p4").mkdir(exist_ok=True)
        alias = self.root / "virtual-p4"
        if not alias.exists():
            alias.symlink_to(target / "p4")
        # Redirect only the guard's hardcoded /p4 check into an isolated fixture.
        guard = (TEST / "disposable_test_guard.sh").read_text().replace('/p4', str(alias))
        # The target itself must still be root/p4, not root/<alias>.
        guard = guard.replace('$TEST_ROOT' + str(alias), '$TEST_ROOT/p4')
        values = 'TEST_ROOT=%s; INSTANCE=unit; TEST_PORT=19999; TEST_OSUSER=perforce;\n' % shlex.quote(str(target))
        (target / ".sdp-disposable-test").write_text("SDP_DISPOSABLE_V1\n%s\n%s\n19999\nperforce\nsdp-test-unit\n" % (target, "unit" if matching else "wrong"))
        return guard + "\n" + values + "sdp_test_validate_marker\n"

    def test_guard_accepts_matching_marked_fixture(self):
        result = self.execute(self.guard_script())
        self.assertEqual(result.returncode, 0, result.stderr)

    def test_guard_rejects_marker_instance_mismatch(self):
        result = self.execute(self.guard_script(matching=False))
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("does not match", result.stderr)

    def test_guard_rejects_unsafe_root_and_invalid_port(self):
        guard = (TEST / "disposable_test_guard.sh").read_text()
        for root, port in [("/", "19999"), ("/tmp/../sdp-test-demo", "19999"), ("/tmp/sdp-test-demo", "0"), ("/tmp/sdp-test-demo", "65536")]:
            result = self.execute(guard + '\nTEST_ROOT=%s; INSTANCE=1; TEST_PORT=%s; TEST_OSUSER=perforce; sdp_test_validate_args' % (shlex.quote(root), shlex.quote(port)))
            self.assertNotEqual(result.returncode, 0)

    def assertion_script(self, name, output, rc):
        source = (TEST / "sdp_test_harness.sh").read_text()
        function = shell_function(TEST / "sdp_test_harness.sh", name)
        helpers = '_pass(){ echo pass; }; _fail(){ echo fail; }; _exec(){ _OUT=%s; _RC=%d; };\n' % (shlex.quote(output), rc)
        return helpers + function + '\n' + name + ' example marker unused\n'

    def test_harness_output_assertions_fail_on_command_failure(self):
        for name, output in [("t_has", "marker"), ("t_hasnot", "")]:
            result = self.execute(self.assertion_script(name, output, 1))
            self.assertEqual(result.stdout.strip(), "fail")

    def test_harness_fixture_failure_stops_before_submit(self):
        calls = self.root / "calls"
        code = 'WORKDIR=%s; CROOT=$WORKDIR/ws; DEPOT=unit; CLIENT=unit;\np4(){ echo "$*" >> %s; return 1; };\n' % (shlex.quote(str(self.root)), shlex.quote(str(calls)))
        code += shell_function(TEST / "sdp_test_harness.sh", "create_fixture") + '\ncreate_fixture\n'
        result = self.execute(code)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(calls.read_text().strip(), "depot -o unit")

    def test_harness_accepts_exact_server_identity_with_actual_tag_case(self):
        function = shell_function(TEST / "sdp_test_harness.sh", "server_info_matches")
        for tag, identity, expected in (("ServerID", "sdp-test-1", 0), ("serverID", "sdp-test-1", 0), ("ServerID", "sdp-test-10", 1)):
            info = "... %s %s\n... serverRoot /p4/1/root" % (tag, identity)
            result = self.execute('INSTANCE=1; P4ROOT=/p4/1/root; SERVER_INFO=%s;\n' % shlex.quote(info) + function + "\nserver_info_matches")
            self.assertEqual(result.returncode, expected, result.stderr)

    def test_configure_starts_only_a_stopped_server(self):
        source = (SETUP / "configure_new_server.sh").read_text()
        block = source[source.index('if ! is_server_up p4d; then'):source.index('\nsleep 1')]
        for running, expected in ((0, ""), (1, "started")):
            result = self.execute('is_server_up(){ return %d; }; start_p4d(){ echo started; };\n' % running + block)
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertEqual(result.stdout.strip(), expected)

    def test_configure_preserves_authenticated_admin_password(self):
        source = (SETUP / "configure_new_server.sh").read_text()
        begin = source.index('         if ! p4 login -s >/dev/null 2>&1; then')
        end = source.index('\n         "$P4CBIN"/p4login -v', begin)
        block = source[begin:end]
        calls = self.root / "p4-calls"
        result = self.execute('p4(){ printf "%s\\n" "$*" >> %s; };\n' % ('%s', shlex.quote(str(calls))) + block)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(calls.read_text().strip(), "login -s")

    def test_changed_shells_parse(self):
        for path in [SETUP / "install_sdp_python.sh", SETUP / "install_sdp.sh", TEST / "reset.sh", TEST / "sdp_test_harness.sh", TEST / "disposable_test_guard.sh"]:
            result = subprocess.run([BASH, "-n", str(path)], capture_output=True, text=True)
            self.assertEqual(result.returncode, 0, result.stderr)


class TriggerImprovements(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-trigger-improvements-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.modules = {"ldap": types.ModuleType("ldap")}

    def load(self, name, env=None):
        with mock.patch.dict(sys.modules, self.modules), mock.patch.dict(os.environ, env or {}):
            return runpy.run_path(str(TRIGGERS / name))

    def test_radius_hosts_and_secret_file_configuration(self):
        pyrad = types.ModuleType("pyrad")
        client = types.ModuleType("pyrad.client"); client.Client = mock.Mock()
        dictionary = types.ModuleType("pyrad.dictionary"); dictionary.Dictionary = mock.Mock()
        packet = types.ModuleType("pyrad.packet")
        self.modules.update({"pyrad": pyrad, "pyrad.client": client, "pyrad.dictionary": dictionary, "pyrad.packet": packet})
        secret = self.root / "secret"; secret.write_bytes(b"test-secret\n")
        namespace = self.load("rad_authcheck.py", {"SDP_AUTH_RADIUS_SERVERS": "one two", "SDP_AUTH_RADIUS_SECRET_FILE": str(secret), "SDP_AUTH_AD_HOSTS": "ldaps://a ldaps://b"})
        self.assertEqual(namespace["RAD_SERVERS"], ["one", "two"])
        self.assertEqual(namespace["RAD_SHARED_SECRET"], b"test-secret")
        self.assertEqual(namespace["AD_HOSTS"], ["ldaps://a", "ldaps://b"])
        client.Client.assert_not_called()

    def test_incident_trigger_instance_does_not_consume_existing_user_arg(self):
        namespace = self.load("submit_form_1_in.py")
        for args, instance in [(["123", "alice", "--instance", "blue"], "blue"), (["123", "alice"], "1")]:
            with mock.patch.dict(os.environ, {}, clear=True), mock.patch.object(sys, "argv", ["trigger", *args]), mock.patch("subprocess.check_output", return_value="Incident: 123") as command, self.assertRaises(SystemExit) as stop:
                namespace["main"]()
            self.assertEqual(stop.exception.code, 0)
            self.assertEqual(command.call_args.args[0][1:3], [instance, "/p4/{0}/bin/p4_{0}".format(instance)])


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