CIAgent
88a1dab810
---ci--- phase: 7 milestone: v0.2 status: complete requirements: covered: [REQ-2-001, REQ-2-002, REQ-2-003, REQ-2-004, REQ-2-005, REQ-2-006, REQ-2-007, REQ-2-008, REQ-2-009, REQ-2-010, REQ-2-011, REQ-2-012] partial: [] ---/ci--- Milestone v0.2 (ai-tutor-architecture) merged to main. Escalation record (audit remediation, durable): P1 executor delegation failed twice (empty subagent results, zero files created); auto-resolved at full autonomy to inline execution with identical plan fidelity (commit 3271373, reflog-only after phase branch squash-delete).
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""Lab feedback endpoint tests — SSE envelope with agent=lab (REQ-2-007)."""
|
|
|
|
import json
|
|
|
|
|
|
def stream_events(client, payload) -> list[dict]:
|
|
with client.stream("POST", "/v1/lab/feedback", json=payload) as response:
|
|
assert response.status_code == 200
|
|
events = []
|
|
for line in response.iter_lines():
|
|
if line.startswith("data:"):
|
|
d = line.removeprefix("data:").strip()
|
|
if d == "[DONE]":
|
|
events.append({"type": "[DONE]"})
|
|
else:
|
|
events.append(json.loads(d))
|
|
return events
|
|
|
|
|
|
def test_lab_feedback_streams_full_envelope(client):
|
|
events = stream_events(client, {"scenario_id": "lab-scenario-strong"})
|
|
assert events[0]["type"] == "meta"
|
|
assert events[0]["agent"] == "lab"
|
|
assert events[0]["scenario_id"] == "lab-scenario-strong"
|
|
deltas = [e for e in events if e["type"] == "delta"]
|
|
assert len(deltas) >= 1
|
|
assert any(e["type"] == "done" for e in events)
|
|
assert events[-1]["type"] == "[DONE]"
|
|
|
|
|
|
def test_unknown_scenario_404(client):
|
|
response = client.post("/v1/lab/feedback", json={"scenario_id": "nope"})
|
|
assert response.status_code == 404
|
|
assert "nope" in response.json()["detail"]
|
|
|
|
|
|
def test_distinct_scenarios_distinct_replies(client):
|
|
strong = stream_events(client, {"scenario_id": "lab-scenario-strong"})
|
|
struggling = stream_events(client, {"scenario_id": "lab-scenario-struggling"})
|
|
strong_text = "".join(e["content"] for e in strong if e["type"] == "delta")
|
|
struggling_text = "".join(e["content"] for e in struggling if e["type"] == "delta")
|
|
assert strong_text != struggling_text
|
|
|
|
|
|
def test_missing_scenario_id_422(client):
|
|
response = client.post("/v1/lab/feedback", json={})
|
|
assert response.status_code == 422
|