127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from sssf_opencode.adapter import OpenCodeResult
|
|
from sssf_opencode.config import load_factory_config
|
|
from sssf_opencode.runner import FactoryRunError, FactoryRunner
|
|
|
|
|
|
REPOSITORY = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class RecordingAdapter:
|
|
def __init__(self, text: str = '{"status": "success", "summary": "phase complete"}') -> None:
|
|
self.requests = []
|
|
self.text = text
|
|
|
|
def run(self, request):
|
|
self.requests.append(request)
|
|
return OpenCodeResult(
|
|
session_id=f"ses_{len(self.requests)}",
|
|
text=self.text,
|
|
input_tokens=1,
|
|
output_tokens=2,
|
|
cache_read_tokens=0,
|
|
cache_write_tokens=0,
|
|
reasoning_tokens=0,
|
|
cost=0.01,
|
|
)
|
|
|
|
|
|
class FactoryRunnerTests(unittest.TestCase):
|
|
def test_persists_the_session_and_reuses_it_for_phase_corrections(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
worktree = Path(temp)
|
|
adapter = RecordingAdapter()
|
|
runner = FactoryRunner(
|
|
adapter=adapter,
|
|
config=load_factory_config(REPOSITORY / "factory.json"),
|
|
worktree=worktree,
|
|
run_id="run_001",
|
|
)
|
|
|
|
first = runner.run_phase("builder", "implement the accepted plan")
|
|
second = runner.run_phase("builder", "repair the malformed report")
|
|
|
|
self.assertEqual(first.envelope["summary"], "phase complete")
|
|
self.assertEqual(second.session_id, "ses_2")
|
|
self.assertIsNone(adapter.requests[0].session_id)
|
|
self.assertEqual(adapter.requests[1].session_id, "ses_1")
|
|
self.assertEqual(adapter.requests[0].model, "openai/gpt-5.3-codex")
|
|
self.assertEqual(adapter.requests[0].agent, "factory-builder")
|
|
self.assertEqual(
|
|
json.loads((worktree / ".sssf/runs/run_001/sessions.json").read_text()),
|
|
{"builder": "ses_2"},
|
|
)
|
|
self.assertTrue((worktree / ".sssf/runs/run_001/builder.jsonl").exists())
|
|
self.assertEqual(
|
|
json.loads(
|
|
(worktree / ".sssf/runs/run_001/builder-envelope.json").read_text()
|
|
),
|
|
second.envelope,
|
|
)
|
|
|
|
def test_accepts_a_single_json_code_fence_and_persists_its_session(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
worktree = Path(temp)
|
|
adapter = RecordingAdapter(
|
|
"```json\n{\"status\": \"success\", \"summary\": \"fenced report\"}\n```"
|
|
)
|
|
runner = FactoryRunner(
|
|
adapter=adapter,
|
|
config=load_factory_config(REPOSITORY / "factory.json"),
|
|
worktree=worktree,
|
|
run_id="run_002",
|
|
)
|
|
|
|
result = runner.run_phase("planner", "produce a fenced report")
|
|
|
|
self.assertEqual(result.envelope["summary"], "fenced report")
|
|
self.assertEqual(
|
|
json.loads((worktree / ".sssf/runs/run_002/sessions.json").read_text()),
|
|
{"planner": "ses_1"},
|
|
)
|
|
|
|
def test_persists_session_when_the_response_envelope_is_invalid(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
worktree = Path(temp)
|
|
adapter = RecordingAdapter("not JSON")
|
|
runner = FactoryRunner(
|
|
adapter=adapter,
|
|
config=load_factory_config(REPOSITORY / "factory.json"),
|
|
worktree=worktree,
|
|
run_id="run_003",
|
|
)
|
|
|
|
with self.assertRaisesRegex(FactoryRunError, "returned invalid JSON"):
|
|
runner.run_phase("planner", "produce an invalid report")
|
|
|
|
self.assertEqual(
|
|
json.loads((worktree / ".sssf/runs/run_003/sessions.json").read_text()),
|
|
{"planner": "ses_1"},
|
|
)
|
|
|
|
def test_accepts_a_final_json_object_after_a_model_prelude(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
worktree = Path(temp)
|
|
adapter = RecordingAdapter(
|
|
"I inspected the worktree. Here is the requested envelope:\n\n"
|
|
'{"status": "success", "summary": "prefaced report"}'
|
|
)
|
|
runner = FactoryRunner(
|
|
adapter=adapter,
|
|
config=load_factory_config(REPOSITORY / "factory.json"),
|
|
worktree=worktree,
|
|
run_id="run_004",
|
|
)
|
|
|
|
result = runner.run_phase("planner", "produce a prefaced report")
|
|
|
|
self.assertEqual(result.envelope["summary"], "prefaced report")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|