🧹 Refactor verify_pull_request to extract helper functions

.gitignore

@@ -22,3 +22,4 @@ evaluations/results/
 # Installed locally from an exact source lock. Upstream redistribution permission is unresolved.
 skills/alibabacloud-resourcecenter-search/
 venv/
+.venv/

src/titmas_action_gate/pr_gate.py

@@ -2024,25 +2024,42 @@ def _evaluate_pull_request(
     return receipt, result
 
 
-def verify_pull_request(
-    *,
-    task_path: str | Path,
-    evidence_path: str | Path,
-    policy_path: str | Path,
-    test_command: str,
-    approval_path: str | Path | None = None,
-    output_directory: str | Path = DEFAULT_OUTPUT_DIRECTORY,
-    repository: str | None = None,
-    pull_request: int | None = None,
-    head_sha: str | None = None,
-    execution_identity: str | None = None,
-    environment: Mapping[str, str] | None = None,
-    workspace: str | Path | None = None,
-    action_configuration_path: str | Path | None = None,
-    action_configuration_root: str | Path | None = None,
-) -> dict[str, Any]:
-    """Run one bounded PR verification and create its receipt and summary once."""
+def _abort_reserved(reserved: ExclusiveOutput | None) -> None:
+    if reserved is not None:
+        reserved.__exit__(RuntimeError, RuntimeError("OUTPUT_RESERVATION_ABORTED"), None)
+
+
+def _reserve_outputs(
+    directory: Path,
+    expected_ancestor_identities: tuple[tuple[str, int, int], ...] | None = None,
+) -> tuple[ExclusiveOutput, ExclusiveOutput]:
+    first = ExclusiveOutput(
+        directory / "receipt.json",
+        expected_ancestor_identities=expected_ancestor_identities,
+    )
+    try:
+        second = ExclusiveOutput(
+            directory / "summary.md",
+            expected_ancestor_identities=expected_ancestor_identities,
+        )
+    except Exception:
+        _abort_reserved(first)
+        raise
+    return first, second
+
 
+def _reserve_private_outputs() -> tuple[Path, ExclusiveOutput, ExclusiveOutput]:
+    for _attempt in range(8):
+        private_directory, private_identities = _private_output_directory()
+        try:
+            first, second = _reserve_outputs(private_directory, private_identities)
+        except (FileExistsError, OSError, RuntimeError, UnsafePathError):
+            continue
+        return private_directory, first, second
+    raise UnsafePathError("PRIVATE_OUTPUT_DIRECTORY_UNAVAILABLE")
+
+
+def _prepare_outputs(output_directory: str | Path) -> tuple[Path, Path, ExclusiveOutput, ExclusiveOutput, bool, str]:
     raw_output = Path(output_directory).expanduser()
     output_path_syntax_safe = bool(str(raw_output)) and not any(part in {".", ".."} for part in raw_output.parts)
     requested_output = Path(os.path.abspath(raw_output)) if output_path_syntax_safe else Path(os.path.abspath("artifacts/titmas"))
@@ -2054,57 +2071,52 @@ def verify_pull_request(
     if not output_preflight_safe:
         output_preflight_reason = "OUTPUT_PATH_SYMLINK_NOT_ALLOWED"
 
-    def abort_reserved(reserved: ExclusiveOutput | None) -> None:
-        if reserved is not None:
-            reserved.__exit__(RuntimeError, RuntimeError("OUTPUT_RESERVATION_ABORTED"), None)
-
-    def reserve(
-        directory: Path,
-        expected_ancestor_identities: tuple[tuple[str, int, int], ...] | None = None,
-    ) -> tuple[ExclusiveOutput, ExclusiveOutput]:
-        first = ExclusiveOutput(
-            directory / "receipt.json",
-            expected_ancestor_identities=expected_ancestor_identities,
-        )
-        try:
-            second = ExclusiveOutput(
-                directory / "summary.md",
-                expected_ancestor_identities=expected_ancestor_identities,
-            )
-        except Exception:
-            abort_reserved(first)
-            raise
-        return first, second
-
-    def reserve_private() -> tuple[Path, ExclusiveOutput, ExclusiveOutput]:
-        for _attempt in range(8):
-            private_directory, private_identities = _private_output_directory()
-            try:
-                first, second = reserve(private_directory, private_identities)
-            except (FileExistsError, OSError, RuntimeError, UnsafePathError):
-                continue
-            return private_directory, first, second
-        raise UnsafePathError("PRIVATE_OUTPUT_DIRECTORY_UNAVAILABLE")
-
     if output_preflight_safe:
         try:
-            receipt_output, summary_output = reserve(output)
+            receipt_output, summary_output = _reserve_outputs(output)
         except UnsafePathError:
             output_preflight_safe = False
             output_preflight_reason = "OUTPUT_PATH_SYMLINK_NOT_ALLOWED"
         except (FileExistsError, OSError, RuntimeError):
             output_preflight_safe = False
     if not output_preflight_safe:
-        output, receipt_output, summary_output = reserve_private()
+        output, receipt_output, summary_output = _reserve_private_outputs()
 
     assert receipt_output is not None and summary_output is not None
+    return requested_output, output, receipt_output, summary_output, output_preflight_safe, output_preflight_reason
+
+
+def _output_integrity_state(receipt_output: ExclusiveOutput, summary_output: ExclusiveOutput) -> str:
+    if not receipt_output._path_is_reserved_inode() or not summary_output._path_is_reserved_inode():
+        return "IDENTITY_CHANGED"
+    if not receipt_output.pristine() or not summary_output.pristine():
+        return "CONTENT_MUTATED"
+    return "PRISTINE"
+
+
+def verify_pull_request(
+    *,
+    task_path: str | Path,
+    evidence_path: str | Path,
+    policy_path: str | Path,
+    test_command: str,
+    approval_path: str | Path | None = None,
+    output_directory: str | Path = DEFAULT_OUTPUT_DIRECTORY,
+    repository: str | None = None,
+    pull_request: int | None = None,
+    head_sha: str | None = None,
+    execution_identity: str | None = None,
+    environment: Mapping[str, str] | None = None,
+    workspace: str | Path | None = None,
+    action_configuration_path: str | Path | None = None,
+    action_configuration_root: str | Path | None = None,
+) -> dict[str, Any]:
+    """Run one bounded PR verification and create its receipt and summary once."""
+
+    requested_output, output, receipt_output, summary_output, output_preflight_safe, output_preflight_reason = _prepare_outputs(output_directory)
 
     def output_integrity_state() -> str:
-        if not receipt_output._path_is_reserved_inode() or not summary_output._path_is_reserved_inode():
-            return "IDENTITY_CHANGED"
-        if not receipt_output.pristine() or not summary_output.pristine():
-            return "CONTENT_MUTATED"
-        return "PRISTINE"
+        return _output_integrity_state(receipt_output, summary_output)
 
     observed_output_state = "PRISTINE"
 
@@ -2136,9 +2148,9 @@ def output_integrity_check() -> str:
     output_relocated = output_state != "PRISTINE"
     if output_relocated:
         reason = "RESERVED_OUTPUT_IDENTITY_CHANGED" if output_state == "IDENTITY_CHANGED" else "RESERVED_OUTPUT_CONTENT_MUTATED"
-        abort_reserved(receipt_output)
-        abort_reserved(summary_output)
-        output, receipt_output, summary_output = reserve_private()
+        _abort_reserved(receipt_output)
+        _abort_reserved(summary_output)
+        output, receipt_output, summary_output = _reserve_private_outputs()
     receipt["output_integrity"] = {
         "requested_reference": requested_output.name,
         "preflight_safe": output_preflight_safe,

src/titmas_action_gate/service.py

@@ -5,6 +5,7 @@
 import hashlib
 import hmac
 import json
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -20,7 +21,6 @@
 from .provider import GitHubProvider
 from .signing import HmacRecordSigner
 from .store import AppendOnlyStore
-from dataclasses import dataclass
 
 
 @dataclass(frozen=True)

tests/test_security_argument_injection.py

@@ -1,9 +1,8 @@
 import unittest
-import subprocess
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
 from titmas_action_gate.provider import GhCliProvider
-from titmas_action_gate.errors import ActionGateError
+
 
 class ProviderSecurityTests(unittest.TestCase):
     @patch("subprocess.run")

tests/test_workflow.py

@@ -1,8 +1,8 @@
+import json
 import tempfile
 import unittest
 from pathlib import Path
 
-import json
 from titmas_action_gate.workflow import validate_agentteams_template, write_demo_report
 
 
@@ -83,7 +83,7 @@ def test_write_demo_report(self):
         self.assertEqual(result_path, output_file)
         self.assertTrue(output_file.exists())
 
-        with open(output_file, "r", encoding="utf-8") as f:
+        with open(output_file, encoding="utf-8") as f:
             content = json.load(f)
 
         self.assertEqual(content, report_data)
@@ -95,7 +95,7 @@ def test_write_demo_report_creates_directories(self):
         write_demo_report(report_data, output_file)
 
         self.assertTrue(output_file.exists())
-        with open(output_file, "r", encoding="utf-8") as f:
+        with open(output_file, encoding="utf-8") as f:
             content = json.load(f)
 
         self.assertEqual(content, report_data)