test_trigger_utilities.py #1

#!/usr/bin/env python3
"""Deterministic trigger behavior using temporary forms and fake P4/RADIUS APIs."""
import contextlib
import io
import os
from pathlib import Path
import runpy
import subprocess
import sys
import tempfile
import types
import unittest
from unittest import mock

SDP = Path(__file__).resolve().parents[1]
TRIGGERS = SDP / "Server/Unix/p4/common/bin/triggers"


class TriggerUtilityTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-trigger-utilities-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)

    def script(self, name, *args):
        return subprocess.run([sys.executable, str(TRIGGERS / name), *args], cwd=self.root, text=True, capture_output=True, timeout=10)

    def form(self, content):
        path = self.root / "form.txt"
        path.write_text(content)
        return path

    def test_jobs_filter_rejects_expensive_patterns_and_allows_specific_search(self):
        for pattern in ("*", "**", "description=*", "*text*"):
            self.assertEqual(self.script("JobsCmdFilter.py", "-e", pattern).returncode, 1)
        self.assertEqual(self.script("JobsCmdFilter.py", "-e", "status=open", "-m", "10").returncode, 0)

    def test_workspace_ascii_and_name_checks(self):
        for content, expected in [("Client: valid_name\nRoot: /work/test\n", 0), ("Client: bad$name\n", 1), ("Client: valid\nDescription: caf\u00e9\n", 1)]:
            self.assertEqual(self.script("PreventWsNonAscii.py", str(self.form(content))).returncode, expected)

    def test_workspace_defaults_change_only_new_forms(self):
        for script in ("SetWsOptions.py", "SetWsOptionsAndView.py"):
            content = "Client: test\nOptions: noallwrite nomodtime normdir\nSubmitOptions: submitunchanged\nView:\n\t//depot/... //test/...\n"
            path = self.form(content)
            result = self.script(script, str(path))
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertIn(" modtime rmdir", path.read_text())
            self.assertIn("leaveunchanged", path.read_text())
            if script.endswith("AndView.py"):
                self.assertIn("//EditMe-AdjustThisPath/...", path.read_text())
            existing = content.replace("Options:", "Access: 2026/10/16 12:00:00\nOptions:")
            path.write_text(existing)
            self.assertEqual(self.script(script, str(path)).returncode, 0)
            self.assertEqual(path.read_text(), existing)

    def test_workspace_template_and_stream_views_are_preserved(self):
        content = "Client: test\nOptions: nomodtime normdir\nView:\n\t//specific/... //test/...\n"
        path = self.form(content)
        self.assertEqual(self.script("SetWsOptionsAndView.py", str(path), "-t,template,test").returncode, 0)
        self.assertEqual(path.read_text(), content)
        stream = content.replace("View:\n", "Stream: //stream/main\nView:\n")
        path.write_text(stream)
        self.assertEqual(self.script("SetWsOptionsAndView.py", str(path)).returncode, 0)
        self.assertIn("//specific/...", path.read_text())

    def invoke_main(self, name, arguments, output=None, error=None):
        namespace = runpy.run_path(str(TRIGGERS / name))
        with mock.patch.object(sys, "argv", [name, *arguments]), mock.patch("subprocess.check_output", return_value=output, side_effect=error) as command, contextlib.redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as stop:
            namespace["main"]()
        return stop.exception.code, command

    def test_tfs_job_must_be_real_job_not_description_text(self):
        valid = "Change 1\n\tdescription\nJobs fixed ...\nTFS123 fixed on today\n\tdetail\nAffected files ...\n"
        invalid = "Change 1\n\tTFS123 in description\nJobs fixed ...\nOTHER99 fixed\n\tTFS456 in job description\nAffected files ...\n"
        self.assertEqual(self.invoke_main("TFSJobCheck.py", ["1"], valid)[0], 0)
        self.assertEqual(self.invoke_main("TFSJobCheck.py", ["1"], invalid)[0], 1)
        self.assertEqual(self.invoke_main("TFSJobCheck.py", ["1"], "No linked jobs")[0], 1)
        self.assertEqual(self.invoke_main("TFSJobCheck.py", ["1"], error=OSError("offline"))[0], 1)

    def test_description_template_applies_only_to_group_members(self):
        content = "Change: new\nDescription:\n\t<enter description here>\n"
        path = self.form(content)
        self.assertEqual(self.invoke_main("submit_form_1.py", [str(path), "alice"], "other_group\n")[0], 0)
        self.assertEqual(path.read_text(), content)
        self.assertEqual(self.invoke_main("submit_form_1.py", [str(path), "alice"], "submit_form_1_users\n")[0], 0)
        self.assertIn("Incident:", path.read_text())
        self.assertIn("Reviewed by:", path.read_text())

    def test_command_block_honors_exempt_user_and_read_only_option(self):
        self.assertEqual(self.script("command_block.py", "perforce").returncode, 0)
        self.assertEqual(self.script("command_block.py", "ordinary", "-o").returncode, 0)
        self.assertEqual(self.script("command_block.py", "ordinary", "-i").returncode, 1)

    def test_checkpoint_trigger_blocks_checkpoint_and_allows_journal(self):
        for command, expected in [("checkpoint", 1), ("checkpoint -z", 1), ("journal", 0), ("stop", 0)]:
            result = subprocess.run(["/bin/bash", str(TRIGGERS / "block_checkpoint.sh"), "ordinary", *command.split()], capture_output=True, text=True)
            self.assertEqual(result.returncode, expected)

    def p4_module(self):
        module = types.ModuleType("P4")
        module.P4Exception = type("P4Exception", (Exception,), {})
        module.P4 = mock.Mock(return_value=types.SimpleNamespace(connect=lambda: None, errors=[], warnings=[]))
        def depot(path):
            obj = types.SimpleNamespace(depotFile=path, revisions=[])
            def revision():
                rev = types.SimpleNamespace(); obj.revisions.append(rev); return rev
            obj.new_revision = revision
            return obj
        module.DepotFile = depot
        return module

    def test_p4change_preserves_files_revisions_and_jobs(self):
        with mock.patch.dict(sys.modules, {"P4": self.p4_module(), "sdp_structured_log": None}):
            namespace = runpy.run_path(str(TRIGGERS / "P4Triggers.py"))
        change = namespace["P4Change"]({"change": "1", "user": "u", "client": "c", "desc": "d", "time": "0", "status": "submitted", "depotFile": ["//d/a"], "type": ["text"], "rev": ["2"], "action": ["edit"], "job": ["j1"], "jobstat": ["closed"]})
        self.assertEqual(change.files[0].depotFile, "//d/a")
        self.assertEqual(change.files[0].revisions[0].rev, "2")
        self.assertEqual(change.jobs, {"j1": "closed"})

    def test_p4trigger_unexpected_exception_returns_failure_and_logs(self):
        with mock.patch.dict(sys.modules, {"P4": self.p4_module(), "sdp_structured_log": None}):
            namespace = runpy.run_path(str(TRIGGERS / "P4Triggers.py"))
        trigger = namespace["P4Trigger"](log=str(self.root / "trigger.log"))
        self.addCleanup(trigger.log.close)
        trigger.getChange = mock.Mock(side_effect=RuntimeError("fixture exception"))
        self.assertEqual(trigger.parseChange("1"), 1)
        self.assertIn("fixture exception", (self.root / "trigger.log").read_text())

    def radius(self, replies):
        pyrad = types.ModuleType("pyrad")
        packet = types.ModuleType("pyrad.packet"); packet.AccessRequest=1; packet.AccessAccept=2
        pyrad.packet = packet
        client = types.ModuleType("pyrad.client")
        dictionary = types.ModuleType("pyrad.dictionary"); dictionary.Dictionary = lambda path: path
        request = dict(); request = type("Request", (dict,), {"PwCrypt": lambda self, text: b"encrypted"})()
        outcomes = iter(replies)
        def create(**kwargs):
            reply = next(outcomes)
            if isinstance(reply, Exception):
                raise reply
            return types.SimpleNamespace(CreateAuthPacket=lambda **kwargs: request, SendPacket=lambda req: types.SimpleNamespace(code=reply, keys=lambda: []))
        client.Client = mock.Mock(side_effect=create)
        env = {"SDP_AUTH_RADIUS_SERVERS": "one two", "SDP_AUTH_RADIUS_SECRET": "fixture"}
        with mock.patch.dict(sys.modules, {"pyrad": pyrad, "pyrad.packet": packet, "pyrad.client": client, "pyrad.dictionary": dictionary}), mock.patch.dict(os.environ, env), mock.patch.object(sys, "argv", ["radtest", "alice"]), mock.patch.object(sys, "stdin", io.StringIO("password")), contextlib.redirect_stdout(io.StringIO()):
            try:
                runpy.run_path(str(TRIGGERS / "radtest.py"), run_name="__main__")
                rc = 0
            except SystemExit as error:
                rc = error.code
        return rc, client.Client.call_count

    def test_radius_diagnostic_reports_denial_as_failure(self):
        self.assertEqual(self.radius([3, 3]), (1, 2))

    def test_radius_diagnostic_tries_next_server_after_failure(self):
        self.assertEqual(self.radius([OSError("offline"), 2]), (0, 2))


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