| """Agent orchestrator (stub/scripted path — CI never needs a model).""" |
| import base64 |
| import json |
|
|
| from calendar_out.ics import events_to_ics |
| from server.orchestrator import run_orchestrator |
| from server.schema import Event |
|
|
| THREAD = ( |
| "Room parent: picture day is Thursday June 18 at 9am, wear green!\n" |
| "Me: adding it now" |
| ) |
|
|
|
|
| def _cal_b64() -> str: |
| ics = events_to_ics([ |
| Event(title="Standup", start="2026-06-18T09:00:00", end="2026-06-18T09:30:00"), |
| ]) |
| return base64.b64encode(ics).decode("ascii") |
|
|
|
|
| def test_orchestrator_scripted_end_to_end(): |
| steps = list(run_orchestrator(THREAD)) |
| kinds = [s["kind"] for s in steps] |
| assert kinds[0] == "plan" |
| assert "tool_call" in kinds and "tool_result" in kinds |
| assert kinds[-1] == "final" |
| final = steps[-1] |
| assert final["summary"]["events"] >= 1 |
| assert final["plan"]["events"], "extracted at least one event" |
| |
| assert final["ics_base64"] |
| assert b"BEGIN:VCALENDAR" in base64.b64decode(final["ics_base64"]) |
| |
| json.dumps(steps) |
|
|
|
|
| def test_orchestrator_checks_conflicts_when_calendar_given(): |
| steps = list(run_orchestrator(THREAD, ics_b64=_cal_b64())) |
| tools = [s.get("tool") for s in steps if s["kind"] == "tool_call"] |
| assert "check_conflicts" in tools |
| assert steps[-1]["kind"] == "final" |
|
|
|
|
| def test_orchestrator_skips_conflicts_without_calendar(): |
| steps = list(run_orchestrator(THREAD)) |
| tools = [s.get("tool") for s in steps if s["kind"] == "tool_call"] |
| assert "check_conflicts" not in tools |
|
|
|
|
| def test_orchestrator_handles_eventless_thread(): |
| steps = list(run_orchestrator("ok thanks!! see you")) |
| final = steps[-1] |
| assert final["kind"] == "final" |
| assert final["summary"]["events"] == 0 |
| assert final["ics_base64"] is None |
|
|
|
|
| def test_extract_events_tool_accepts_memory(): |
| from server.mcp_tools import extract_events |
|
|
| plan = extract_events(THREAD, memory="Dana is the soccer coach") |
| assert isinstance(plan, dict) and "events" in plan |
|
|
|
|
| def test_homepage_analyze_runs_the_agentic_workflow(): |
| """The homepage's _on_analyze is orchestrator-driven: the thinking panel |
| carries the agent's step trace, results keep the legacy 5-tuple contract.""" |
| from ui.blocks import _on_analyze |
|
|
| *_, last = list(_on_analyze(THREAD, None, None)) |
| trace, rows, plan_md, reply, status = last |
| assert "ag-final" in trace and "ag-tool_call" in trace |
| assert rows and rows[0][0] |
| assert status.startswith("Found ") |
|
|
