diff --git a/.gitignore b/.gitignore index ae69a1f..24c743c 100644 --- a/.gitignore +++ b/.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/ diff --git a/src/titmas_action_gate/pr_gate.py b/src/titmas_action_gate/pr_gate.py index 36d3c80..a452032 100644 --- a/src/titmas_action_gate/pr_gate.py +++ b/src/titmas_action_gate/pr_gate.py @@ -1353,6 +1353,111 @@ def _observe_stream(stream: BinaryIO, exceeded: threading.Event, budget: _Shared return _StreamObservation(digest.hexdigest(), observed_bytes, limit_exceeded) +def _get_unexecuted_test_result(command: list[str]) -> dict[str, Any]: + return { + "executed": False, + "exit_code": None, + "command_sha256": sha256_json(command), + "stdout_sha256": None, + "stderr_sha256": None, + "duration_ms": 0, + "stdout_bytes": 0, + "stderr_bytes": 0, + "combined_output_bytes": 0, + "max_total_output_bytes": MAX_TOTAL_TEST_OUTPUT_BYTES, + "output_limit_exceeded": False, + "timed_out": False, + "process_group_cleanup": "NOT_STARTED", + "environment": { + "policy_version": TEST_ENVIRONMENT_POLICY_VERSION, + "allowed_parent_names": list(_TEST_ENVIRONMENT_ALLOWLIST), + "removed_names": [], + "removed_count": 0, + "fresh_home": False, + "fresh_tmpdir": False, + }, + } + + +def _execute_test_process( + command: list[str], + child_environment: Mapping[str, str], + workspace: Path | None, + output_limit_event: threading.Event, + started: float, + observe: Callable[[str, BinaryIO], None], +) -> tuple[int, bool, str]: + timed_out = False + cleanup = "NOT_STARTED" + exit_code = 127 + threads: list[threading.Thread] = [] + process: subprocess.Popen[bytes] | None = None + try: + process = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=str(workspace) if workspace is not None else None, + env=child_environment, + start_new_session=True, + close_fds=True, + bufsize=0, + ) + if process.stdout is None or process.stderr is None: + raise OSError("TEST_OUTPUT_PIPE_UNAVAILABLE") + threads = [ + threading.Thread(target=observe, args=("stdout", process.stdout), daemon=True), + threading.Thread(target=observe, args=("stderr", process.stderr), daemon=True), + ] + for thread in threads: + thread.start() + while process.poll() is None: + if output_limit_event.is_set(): + _terminate_process_group(process) + break + if time.monotonic() - started >= TEST_TIMEOUT_SECONDS: + timed_out = True + _terminate_process_group(process) + break + time.sleep(0.01) + try: + exit_code = process.wait(timeout=3) + except subprocess.TimeoutExpired: + _terminate_process_group(process) + try: + exit_code = process.wait(timeout=3) + except subprocess.TimeoutExpired: + exit_code = 124 + cleaned = process.poll() is not None and _terminate_process_group(process) + cleanup = "COMPLETE" if cleaned else "FAILED" + except OSError: + if process is None: + cleanup = "NOT_STARTED" + exit_code = 127 + else: + cleanup = "FAILED" + finally: + for thread in threads: + thread.join(timeout=2) + if any(thread.is_alive() for thread in threads): + cleanup = "FAILED" + if process is not None: + if process.poll() is None: + group_cleaned = _terminate_process_group(process) + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=3) + cleanup = "COMPLETE" if group_cleaned and process.poll() is not None else "FAILED" + if process.stdout is not None: + process.stdout.close() + if process.stderr is not None: + process.stderr.close() + return exit_code, timed_out, cleanup + + def _run_test( command: list[str], *, @@ -1361,37 +1466,11 @@ def _run_test( workspace: Path | None, ) -> dict[str, Any]: if not execute: - return { - "executed": False, - "exit_code": None, - "command_sha256": sha256_json(command), - "stdout_sha256": None, - "stderr_sha256": None, - "duration_ms": 0, - "stdout_bytes": 0, - "stderr_bytes": 0, - "combined_output_bytes": 0, - "max_total_output_bytes": MAX_TOTAL_TEST_OUTPUT_BYTES, - "output_limit_exceeded": False, - "timed_out": False, - "process_group_cleanup": "NOT_STARTED", - "environment": { - "policy_version": TEST_ENVIRONMENT_POLICY_VERSION, - "allowed_parent_names": list(_TEST_ENVIRONMENT_ALLOWLIST), - "removed_names": [], - "removed_count": 0, - "fresh_home": False, - "fresh_tmpdir": False, - }, - } + return _get_unexecuted_test_result(command) started = time.monotonic() - timed_out = False output_limit_event = threading.Event() output_budget = _SharedOutputBudget(MAX_TOTAL_TEST_OUTPUT_BYTES) - cleanup = "NOT_STARTED" - exit_code = 127 observations: dict[str, _StreamObservation] = {} - threads: list[threading.Thread] = [] def observe(name: str, stream: BinaryIO) -> None: observations[name] = _observe_stream(stream, output_limit_event, output_budget) @@ -1403,70 +1482,14 @@ def observe(name: str, stream: BinaryIO) -> None: home.mkdir(mode=0o700) temporary.mkdir(mode=0o700) child_environment, environment_metadata = _test_environment(environment, home=home, temporary_directory=temporary) - process: subprocess.Popen[bytes] | None = None - try: - process = subprocess.Popen( - command, - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - cwd=str(workspace) if workspace is not None else None, - env=child_environment, - start_new_session=True, - close_fds=True, - bufsize=0, - ) - if process.stdout is None or process.stderr is None: - raise OSError("TEST_OUTPUT_PIPE_UNAVAILABLE") - threads = [ - threading.Thread(target=observe, args=("stdout", process.stdout), daemon=True), - threading.Thread(target=observe, args=("stderr", process.stderr), daemon=True), - ] - for thread in threads: - thread.start() - while process.poll() is None: - if output_limit_event.is_set(): - _terminate_process_group(process) - break - if time.monotonic() - started >= TEST_TIMEOUT_SECONDS: - timed_out = True - _terminate_process_group(process) - break - time.sleep(0.01) - try: - exit_code = process.wait(timeout=3) - except subprocess.TimeoutExpired: - _terminate_process_group(process) - try: - exit_code = process.wait(timeout=3) - except subprocess.TimeoutExpired: - exit_code = 124 - cleaned = process.poll() is not None and _terminate_process_group(process) - cleanup = "COMPLETE" if cleaned else "FAILED" - except OSError: - if process is None: - cleanup = "NOT_STARTED" - exit_code = 127 - else: - cleanup = "FAILED" - finally: - for thread in threads: - thread.join(timeout=2) - if any(thread.is_alive() for thread in threads): - cleanup = "FAILED" - if process is not None: - if process.poll() is None: - group_cleaned = _terminate_process_group(process) - try: - process.wait(timeout=3) - except subprocess.TimeoutExpired: - process.kill() - process.wait(timeout=3) - cleanup = "COMPLETE" if group_cleaned and process.poll() is not None else "FAILED" - if process.stdout is not None: - process.stdout.close() - if process.stderr is not None: - process.stderr.close() + exit_code, timed_out, cleanup = _execute_test_process( + command, + child_environment, + workspace, + output_limit_event, + started, + observe, + ) empty_digest = hashlib.sha256(b"").hexdigest() stdout_observation = observations.get("stdout", _StreamObservation(empty_digest, 0, False)) stderr_observation = observations.get("stderr", _StreamObservation(empty_digest, 0, False)) diff --git a/src/titmas_action_gate/service.py b/src/titmas_action_gate/service.py index 90d7164..df6bc08 100644 --- a/src/titmas_action_gate/service.py +++ b/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) diff --git a/tests/test_security_argument_injection.py b/tests/test_security_argument_injection.py index 3990256..4d8bb42 100644 --- a/tests/test_security_argument_injection.py +++ b/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") diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 61c356d..461a3bf 100644 --- a/tests/test_workflow.py +++ b/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)