| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
|
|
|
|
| REPO_ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| class PipelineIntegrationTests(unittest.TestCase): |
| def run_cli(self, *arguments: str) -> subprocess.CompletedProcess[str]: |
| return subprocess.run( |
| [sys.executable, str(REPO_ROOT / "scripts" / "run_model.py"), *arguments], |
| cwd=REPO_ROOT, |
| capture_output=True, |
| text=True, |
| check=False, |
| ) |
|
|
| def test_dry_run_records_all_resolved_commands(self) -> None: |
| with tempfile.TemporaryDirectory() as temporary: |
| result_path = Path(temporary) / "dry-run.json" |
| completed = self.run_cli( |
| "--config", "configs/examples/example_dry_run.json", "--dry-run", "--result-out", str(result_path) |
| ) |
| self.assertEqual(completed.returncode, 0, completed.stderr) |
| result = json.loads(result_path.read_text(encoding="utf-8")) |
| self.assertEqual(result["overall_status"], "QUEUED") |
| self.assertEqual(len(result["stages"]), 4) |
| self.assertTrue(all(stage["command_argv"] for stage in result["stages"])) |
|
|
| def test_success_is_reused_and_failure_does_not_hide(self) -> None: |
| model_dir = REPO_ROOT / "models" / "automation_fixture" / "T00FIXTURE" |
| if model_dir.exists(): |
| import shutil |
| shutil.rmtree(model_dir) |
| first = self.run_cli("--config", "configs/examples/example_dry_run.json") |
| self.assertEqual(first.returncode, 0, first.stderr) |
| result_path = model_dir / "run_result.json" |
| first_result = json.loads(result_path.read_text(encoding="utf-8")) |
| self.assertEqual(first_result["overall_status"], "PASS") |
| second = self.run_cli("--config", "configs/examples/example_dry_run.json", "--resume-failed") |
| self.assertEqual(second.returncode, 0, second.stderr) |
| second_result = json.loads(result_path.read_text(encoding="utf-8")) |
| self.assertTrue(all(stage["reused_from_previous_run"] for stage in second_result["stages"])) |
|
|
| def test_missing_input_is_recorded_as_failure(self) -> None: |
| config = json.loads( |
| (REPO_ROOT / "configs" / "examples" / "example_dry_run.json").read_text(encoding="utf-8") |
| ) |
| config["model"]["model_id"] = "T00FAILFIXTURE" |
| config["stages"][0]["command"][3] = "{repo_root}/tests/fixtures/does-not-exist.fixture" |
| config["stages"][0]["inputs"] = ["{repo_root}/tests/fixtures/does-not-exist.fixture"] |
| with tempfile.TemporaryDirectory() as temporary: |
| config_path = Path(temporary) / "failure.json" |
| config_path.write_text(json.dumps(config), encoding="utf-8") |
| completed = self.run_cli("--config", str(config_path)) |
| self.assertEqual(completed.returncode, 1, completed.stderr) |
| result_path = REPO_ROOT / "models" / "automation_fixture" / "T00FAILFIXTURE" / "run_result.json" |
| result = json.loads(result_path.read_text(encoding="utf-8")) |
| self.assertEqual(result["stages"][0]["status"], "FAIL") |
| self.assertEqual(result["stages"][0]["failure_code"], "FAIL_SOURCE") |
| self.assertEqual(result["stages"][1]["status"], "BLOCKED") |
|
|
| def test_adding_stage_reuses_unchanged_successful_stages(self) -> None: |
| model_dir = REPO_ROOT / "models" / "automation_fixture" / "T00EXTENDFIXTURE" |
| if model_dir.exists(): |
| import shutil |
| shutil.rmtree(model_dir) |
| config = json.loads( |
| (REPO_ROOT / "configs" / "examples" / "example_dry_run.json").read_text(encoding="utf-8") |
| ) |
| config["model"]["model_id"] = "T00EXTENDFIXTURE" |
| with tempfile.TemporaryDirectory() as temporary: |
| first_path = Path(temporary) / "first.json" |
| first_path.write_text(json.dumps(config), encoding="utf-8") |
| first = self.run_cli("--config", str(first_path)) |
| self.assertEqual(first.returncode, 0, first.stderr) |
| config["stages"].append({ |
| "id": "new_pair_check", |
| "stage": "new_pair_check", |
| "variant": "pair", |
| "artifact_id": "T00EXTENDFIXTURE-pair", |
| "command": [ |
| "{python}", "{repo_root}/scripts/stages/inspect_file.py", |
| "--input", "{model_dir}/source/quantized/public_quantized.fixture", |
| "--report", "{model_dir}/analysis/new_pair_check.json", |
| "--expected-prefix", "PUBLIC_QUANTIZED_FIXTURE_V1", |
| ], |
| "inputs": ["{model_dir}/source/quantized/public_quantized.fixture"], |
| "outputs": ["{model_dir}/analysis/new_pair_check.json"], |
| "requires": ["validate_quantized"], |
| "timeout_sec": 30, |
| "failure_code_on_error": "FAIL_ANALYSIS", |
| "validation_report": "{model_dir}/analysis/new_pair_check.json", |
| "options": {}, |
| "patch": None, |
| }) |
| second_path = Path(temporary) / "second.json" |
| second_path.write_text(json.dumps(config), encoding="utf-8") |
| second = self.run_cli("--config", str(second_path), "--resume-failed") |
| self.assertEqual(second.returncode, 0, second.stderr) |
| result = json.loads((model_dir / "run_result.json").read_text(encoding="utf-8")) |
| self.assertTrue(all(stage["reused_from_previous_run"] for stage in result["stages"][:-1])) |
| self.assertEqual(result["stages"][-1]["status"], "PASS") |
| self.assertFalse(result["stages"][-1]["reused_from_previous_run"]) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
