from __future__ import annotations import json import pathlib import shutil import subprocess import sys import tempfile import textwrap import unittest SKILL_ROOT = pathlib.Path(__file__).resolve().parents[1] REPO_ROOT = pathlib.Path(__file__).resolve().parents[3] SCRIPT = SKILL_ROOT / "scripts" / "regression_validation_gate_runner.py" TMP_ROOT = REPO_ROOT / "tmp" / "regression-validation-gate-runner-tests" def quote(value: str) -> str: return json.dumps(value) def py_command(source: str) -> list[str]: return [sys.executable, "-c", source] def run_cli(*args: str) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, str(SCRIPT), *args], text=True, capture_output=True, check=False, ) def write_text(path: pathlib.Path, text: str) -> pathlib.Path: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return path def write_json(path: pathlib.Path, data: object) -> pathlib.Path: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def load_report(output_dir: pathlib.Path) -> dict[str, object]: return json.loads((output_dir / "gate-run-report.json").read_text(encoding="utf-8")) class RegressionValidationGateRunnerTest(unittest.TestCase): def setUp(self) -> None: TMP_ROOT.mkdir(parents=True, exist_ok=True) self.tmp_dir = tempfile.TemporaryDirectory(dir=TMP_ROOT) self.root = pathlib.Path(self.tmp_dir.name) / "project" self.output_dir = pathlib.Path(self.tmp_dir.name) / "outputs" self.manifest = pathlib.Path(self.tmp_dir.name) / "gates.json" self.root.mkdir() def tearDown(self) -> None: self.tmp_dir.cleanup() @classmethod def tearDownClass(cls) -> None: if TMP_ROOT.exists(): shutil.rmtree(TMP_ROOT) def run_gate_runner(self, mode: str = "run") -> tuple[subprocess.CompletedProcess[str], dict[str, object]]: result = run_cli( "--project-root", str(self.root), "--gate-manifest", str(self.manifest), "--output-dir", str(self.output_dir), "--mode", mode, ) return result, load_report(self.output_dir) def test_dry_run_parses_json_manifest_without_running_commands(self) -> None: marker = self.root / "should-not-exist.txt" write_json( self.manifest, { "gates": [ { "gate_id": "unit", "command": py_command(f"open({quote(str(marker))}, 'w').write('ran')"), "working_directory": ".", "expected_exit_code": 0, "timeout_seconds": 5, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner(mode="dry_run") self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(marker.exists()) self.assertEqual(report["commands_declared"], ["unit"]) self.assertEqual(report["commands_run"], []) self.assertEqual(report["commands_skipped"][0]["reason"], "dry_run") self.assertEqual(report["machine_readable_summary"]["status"], "DRY_RUN") def test_successful_command_capture_writes_log_and_passes(self) -> None: write_json( self.manifest, { "gates": [ { "gate_id": "unit", "command": py_command("print('hello gate')"), "working_directory": ".", "expected_exit_code": 0, "log_file": "logs/unit.log", "timeout_seconds": 5, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(report["pass_fail_status"], "PASS") self.assertEqual(report["exit_codes"]["unit"], 0) log_path = self.output_dir / "logs" / "unit.log" self.assertTrue(log_path.exists()) self.assertIn("hello gate", log_path.read_text(encoding="utf-8")) self.assertEqual(report["commands_run"], ["unit"]) def test_failing_command_capture_fails_required_gate(self) -> None: write_json( self.manifest, { "gates": [ { "gate_id": "regression", "command": py_command("import sys; print('bad'); sys.exit(3)"), "expected_exit_code": 0, "timeout_seconds": 5, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual(report["pass_fail_status"], "FAIL") self.assertEqual(report["exit_codes"]["regression"], 3) gate = report["gate_results"][0] self.assertEqual(gate["status"], "FAIL") self.assertIn("bad", (self.output_dir / "logs" / "regression.log").read_text(encoding="utf-8")) def test_optional_gate_is_skipped_by_default(self) -> None: marker = self.root / "optional-ran.txt" write_json( self.manifest, { "gates": [ { "gate_id": "optional-packaging", "command": py_command(f"open({quote(str(marker))}, 'w').write('ran')"), "expected_exit_code": 0, "timeout_seconds": 5, "required_before_review": False, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(marker.exists()) self.assertEqual(report["commands_run"], []) self.assertEqual(report["commands_skipped"][0]["gate_id"], "optional-packaging") self.assertEqual(report["commands_skipped"][0]["reason"], "optional") def test_missing_command_field_is_manifest_error(self) -> None: write_json( self.manifest, {"gates": [{"gate_id": "broken", "expected_exit_code": 0, "required_before_review": True}]}, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 2) self.assertEqual(report["pass_fail_status"], "ERROR") self.assertEqual(report["machine_readable_summary"]["manifest_error_count"], 1) self.assertIn("command", report["manifest_errors"][0]) def test_timeout_handling_marks_required_gate_failed(self) -> None: write_json( self.manifest, { "gates": [ { "gate_id": "slow", "command": py_command("import time; time.sleep(3)"), "expected_exit_code": 0, "timeout_seconds": 1, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 1, result.stderr) gate = report["gate_results"][0] self.assertEqual(gate["status"], "TIMEOUT") self.assertEqual(report["pass_fail_status"], "FAIL") self.assertIn("timed out", (self.output_dir / "logs" / "slow.log").read_text(encoding="utf-8")) def test_markdown_manifest_with_yaml_fence_is_supported(self) -> None: self.manifest = pathlib.Path(self.tmp_dir.name) / "gates.md" write_text( self.manifest, textwrap.dedent( f"""\ # Gates ```yaml gates: - gate_id: markdown-gate command: - {sys.executable} - -c - "print('from markdown')" expected_exit_code: 0 timeout_seconds: 5 required_before_review: true ``` """ ), ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(report["commands_declared"], ["markdown-gate"]) self.assertEqual(report["commands_run"], ["markdown-gate"]) def test_utf8_paths_and_chinese_filenames_are_logged(self) -> None: project = pathlib.Path(self.tmp_dir.name) / "项目" project.mkdir() self.root = project write_json( self.manifest, { "gates": [ { "gate_id": "中文-gate", "command": py_command("print('中文输出')"), "working_directory": ".", "expected_exit_code": 0, "timeout_seconds": 5, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 0, result.stderr) log_path = self.output_dir / "logs" / "中文-gate.log" self.assertTrue(log_path.exists()) self.assertIn("中文输出", log_path.read_text(encoding="utf-8")) self.assertIn("中文-gate", report["log_paths"]) def test_project_file_changes_are_reported(self) -> None: changed = self.root / "generated.txt" write_json( self.manifest, { "gates": [ { "gate_id": "side-effect", "command": py_command(f"open({quote(str(changed))}, 'w', encoding='utf-8').write('new')"), "expected_exit_code": 0, "timeout_seconds": 5, "required_before_review": True, } ] }, ) result, report = self.run_gate_runner() self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(report["changed_files"], ["generated.txt"]) self.assertEqual(report["gate_results"][0]["changed_files"], ["generated.txt"]) if __name__ == "__main__": unittest.main()