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