90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
RUNNER_ROOT = REPO_ROOT / "apps" / "runner"
|
|
sys.path.insert(0, str(RUNNER_ROOT))
|
|
|
|
from src.application.fake_runner import run_fake_runner_once
|
|
|
|
|
|
class FakeBackendJobClient:
|
|
def __init__(self, job: dict[str, Any] | None) -> None:
|
|
self.job = job
|
|
self.completed: dict[str, Any] | None = None
|
|
|
|
def claim_job(self) -> dict[str, Any] | None:
|
|
job = self.job
|
|
self.job = None
|
|
return job
|
|
|
|
def complete_job(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
self.completed = {"id": job_id, **payload}
|
|
return self.completed
|
|
|
|
|
|
class FakeRunnerTest(unittest.TestCase):
|
|
def test_fake_runner_claims_job_writes_workspace_and_completes(self) -> None:
|
|
backend = FakeBackendJobClient(
|
|
{
|
|
"id": "11111111-1111-1111-1111-111111111111",
|
|
"agent_profile": "fake-codex",
|
|
}
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
result = run_fake_runner_once(
|
|
backend,
|
|
workspace_root=Path(temp_dir),
|
|
)
|
|
self.assertIsNotNone(result)
|
|
assert result is not None
|
|
workspace_path = Path(result["workspace_path"])
|
|
self.assertTrue((workspace_path / "inputs" / "job.json").is_file())
|
|
self.assertTrue((workspace_path / "logs" / "stdout.log").is_file())
|
|
self.assertTrue((workspace_path / "logs" / "stderr.log").is_file())
|
|
self.assertTrue((workspace_path / "outputs" / "result.json").is_file())
|
|
self.assertEqual(0, result["exit_code"])
|
|
self.assertEqual("SUCCEEDED", result["output"]["status"])
|
|
self.assertEqual([{"path": "outputs/result.json"}], result["output"]["output_files"])
|
|
|
|
def test_fake_runner_can_emit_invalid_schema_fixture(self) -> None:
|
|
backend = FakeBackendJobClient(
|
|
{
|
|
"id": "22222222-2222-2222-2222-222222222222",
|
|
"agent_profile": "fake-invalid-schema",
|
|
}
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
result = run_fake_runner_once(
|
|
backend,
|
|
workspace_root=Path(temp_dir),
|
|
)
|
|
|
|
self.assertIsNotNone(result)
|
|
assert result is not None
|
|
self.assertEqual({"status": "NOT_A_STATUS"}, result["output"])
|
|
|
|
def test_fake_runner_noops_without_queued_job(self) -> None:
|
|
backend = FakeBackendJobClient(None)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
result = run_fake_runner_once(
|
|
backend,
|
|
workspace_root=Path(temp_dir),
|
|
)
|
|
|
|
self.assertIsNone(result)
|
|
self.assertIsNone(backend.completed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|