File size: 5,786 Bytes
ed3aeeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | 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()
|