Spaces:
Sleeping
Sleeping
Commit ·
ab640da
1
Parent(s): fad0d42
test: add SingleCallOrchestrator tests with fake backend
Browse filesCo-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
tests/test_single_call_orchestrator.py
ADDED
|
@@ -0,0 +1,78 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Tests for SingleCallOrchestrator using a fake LLM backend."""
|
| 2 |
+
|
| 3 |
+
import json
|
| 4 |
+
from agents.orchestrator import SingleCallOrchestrator
|
| 5 |
+
from tests.conftest import FakeLLMBackend
|
| 6 |
+
|
| 7 |
+
|
| 8 |
+
class TestSingleCallOrchestrator:
|
| 9 |
+
def _make_orchestrator(self, response_json: str, tmp_output_dir):
|
| 10 |
+
backend = FakeLLMBackend(response_json)
|
| 11 |
+
return SingleCallOrchestrator(backend=backend, output_dir=tmp_output_dir), backend
|
| 12 |
+
|
| 13 |
+
def test_response_shape(self, tmp_output_dir):
|
| 14 |
+
resp = json.dumps({"agents": [
|
| 15 |
+
{"id": "design", "message": "An L-bracket would work."},
|
| 16 |
+
]})
|
| 17 |
+
orch, _ = self._make_orchestrator(resp, tmp_output_dir)
|
| 18 |
+
result = orch.chat_turn("I need a bracket", history=[])
|
| 19 |
+
assert "responses" in result
|
| 20 |
+
assert "preview" in result
|
| 21 |
+
assert "design_state" in result
|
| 22 |
+
|
| 23 |
+
def test_passes_message_to_backend(self, tmp_output_dir):
|
| 24 |
+
resp = json.dumps({"agents": [{"id": "design", "message": "OK"}]})
|
| 25 |
+
orch, backend = self._make_orchestrator(resp, tmp_output_dir)
|
| 26 |
+
orch.chat_turn("Test message", history=[])
|
| 27 |
+
assert len(backend.calls) == 1
|
| 28 |
+
last_user_msg = backend.calls[0][-1]["content"]
|
| 29 |
+
assert "Test message" in last_user_msg
|
| 30 |
+
|
| 31 |
+
def test_mentions_restrict_agents(self, tmp_output_dir):
|
| 32 |
+
resp = json.dumps({"agents": [{"id": "cnc", "message": "3-axis OK"}]})
|
| 33 |
+
orch, _ = self._make_orchestrator(resp, tmp_output_dir)
|
| 34 |
+
result = orch.chat_turn("check this", history=[], mentions=["cnc"])
|
| 35 |
+
agent_ids = [r["agent_id"] for r in result["responses"]]
|
| 36 |
+
assert "cnc" in agent_ids
|
| 37 |
+
|
| 38 |
+
def test_invalid_json_fallback(self, tmp_output_dir):
|
| 39 |
+
orch, _ = self._make_orchestrator("not json at all", tmp_output_dir)
|
| 40 |
+
result = orch.chat_turn("help", history=[])
|
| 41 |
+
assert len(result["responses"]) > 0
|
| 42 |
+
assert result["responses"][0]["agent_id"] == "design"
|
| 43 |
+
|
| 44 |
+
def test_llm_exception_fallback(self, tmp_output_dir):
|
| 45 |
+
class RaisingBackend:
|
| 46 |
+
def generate(self, messages):
|
| 47 |
+
raise RuntimeError("API error")
|
| 48 |
+
|
| 49 |
+
orch = SingleCallOrchestrator(backend=RaisingBackend(), output_dir=tmp_output_dir)
|
| 50 |
+
result = orch.chat_turn("Design a part", history=[])
|
| 51 |
+
assert len(result["responses"]) > 0
|
| 52 |
+
|
| 53 |
+
def test_unknown_agent_id_filtered(self, tmp_output_dir):
|
| 54 |
+
resp = json.dumps({"agents": [
|
| 55 |
+
{"id": "nonexistent", "message": "I don't exist"},
|
| 56 |
+
{"id": "design", "message": "Real agent"},
|
| 57 |
+
]})
|
| 58 |
+
orch, _ = self._make_orchestrator(resp, tmp_output_dir)
|
| 59 |
+
result = orch.chat_turn("test", history=[])
|
| 60 |
+
agent_ids = [r["agent_id"] for r in result["responses"]]
|
| 61 |
+
assert "nonexistent" not in agent_ids
|
| 62 |
+
assert "design" in agent_ids
|
| 63 |
+
|
| 64 |
+
def test_history_forwarded_to_backend(self, tmp_output_dir, sample_history):
|
| 65 |
+
resp = json.dumps({"agents": [{"id": "design", "message": "OK"}]})
|
| 66 |
+
orch, backend = self._make_orchestrator(resp, tmp_output_dir)
|
| 67 |
+
orch.chat_turn("continue", history=sample_history)
|
| 68 |
+
user_content = backend.calls[0][-1]["content"]
|
| 69 |
+
assert "servo bracket" in user_content.lower() or "MG996R" in user_content
|
| 70 |
+
|
| 71 |
+
def test_design_state_returned(self, tmp_output_dir):
|
| 72 |
+
resp = json.dumps({"agents": [
|
| 73 |
+
{"id": "engineering", "message": "Use aluminum 6061 with 3mm walls."},
|
| 74 |
+
]})
|
| 75 |
+
orch, _ = self._make_orchestrator(resp, tmp_output_dir)
|
| 76 |
+
result = orch.chat_turn("material?", history=[])
|
| 77 |
+
assert "design_state" in result
|
| 78 |
+
assert isinstance(result["design_state"], dict)
|