63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from pydantic import ValidationError
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
RUNNER_ROOT = REPO_ROOT / "apps" / "runner"
|
|
sys.path.insert(0, str(RUNNER_ROOT))
|
|
|
|
from src.application.job_output_validation import ( # noqa: E402
|
|
_backend_contracts_module,
|
|
validate_agent_job_output,
|
|
)
|
|
|
|
|
|
class RunnerJobOutputValidationTest(unittest.TestCase):
|
|
def test_valid_runner_job_output_uses_backend_contracts(self) -> None:
|
|
output = validate_agent_job_output(
|
|
{
|
|
"status": "SUCCEEDED",
|
|
"output_files": [{"path": "outputs/plan.json"}],
|
|
"payload": {"artifact": "plan"},
|
|
}
|
|
)
|
|
|
|
self.assertEqual("SUCCEEDED", output.status.value)
|
|
|
|
def test_invalid_runner_job_output_status_is_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
validate_agent_job_output({"status": "NOT_A_STATUS"})
|
|
|
|
def test_runner_image_layout_can_use_copied_backend_contracts(self) -> None:
|
|
source_contracts = REPO_ROOT / "apps" / "backend" / "src" / "domain" / "contracts"
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
copied_contracts = Path(temp_dir) / "backend_contracts" / "contracts"
|
|
shutil.copytree(source_contracts, copied_contracts)
|
|
|
|
previous = os.environ.get("PIPELINE_BACKEND_CONTRACTS_DIR")
|
|
os.environ["PIPELINE_BACKEND_CONTRACTS_DIR"] = str(copied_contracts)
|
|
_backend_contracts_module.cache_clear()
|
|
try:
|
|
output = validate_agent_job_output({"status": "SUCCEEDED"})
|
|
finally:
|
|
_backend_contracts_module.cache_clear()
|
|
if previous is None:
|
|
os.environ.pop("PIPELINE_BACKEND_CONTRACTS_DIR", None)
|
|
else:
|
|
os.environ["PIPELINE_BACKEND_CONTRACTS_DIR"] = previous
|
|
|
|
self.assertEqual("SUCCEEDED", output.status.value)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|