Spaces:
Sleeping
Sleeping
| """Agent-panel guarantee: every answered turn lands a visualisation. | |
| Offline: the agent stream is faked (typed v2 events, no LLM/token/network) | |
| so each test drives ``ui.panels.agent.chat_fn`` through a turn whose tools | |
| produce no scored hours, then asserts the panel still ends with a chart in | |
| the ``agent_strip`` slot (``AGENT_KEYS`` index 6) and a 📊 line in the chat. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| import plotly.graph_objects as go | |
| ROOT = Path(__file__).resolve().parent.parent | |
| sys.path.insert(0, str(ROOT)) | |
| from ui import _compat as C # noqa: E402 | |
| from ui import contracts # noqa: E402 | |
| from ui.panels import agent as agent_panel # noqa: E402 | |
| from wavereader import tools as _tools # noqa: E402 | |
| AGENT_STRIP = contracts.AGENT_KEYS.index("agent_strip") | |
| CHATBOT = contracts.AGENT_KEYS.index("chatbot") | |
| _BELLS_ARGS = {"spot_name": "Bells Beach", "region": "Victoria"} | |
| def _tool_events(name: str, args: dict, obs) -> list[tuple]: | |
| return [("tool_call", {"name": name, "arguments": args}), | |
| ("tool_result", {"name": name, "arguments": args, | |
| "observation": obs, "summary": "ok", | |
| "ms": 1.0, "preview": ""})] | |
| def _run(monkeypatch, events: list[tuple], final: str, query: str): | |
| def stream(message, **kwargs): | |
| yield ("step", {"n": 1}) | |
| yield from events | |
| yield ("final", final) | |
| yield ("usage", {"steps": 1, "profile": "standard"}) | |
| monkeypatch.setattr(C, "agent_run_stream", stream) | |
| records = C.list_breaks() | |
| selected = C.find_break(records, "Bells Beach") | |
| assert selected is not None, "Bells Beach must resolve from the catalogue" | |
| return list(agent_panel.chat_fn( | |
| query, [], "intermediate", "", selected, | |
| agent_panel.fresh_store(), "standard", C.ALL, C.ALL, records)) | |
| def _last_emit(monkeypatch, obs, tool_name: str, args: dict, | |
| final: str = "**Best: Bells Beach @ Sat am — 6/10.**", | |
| query: str = "test question"): | |
| outs = _run(monkeypatch, _tool_events(tool_name, args, obs), final, query) | |
| return outs, outs[-1] | |
| def test_climate_turn_charts_swell_rose(monkeypatch): | |
| obs = _tools.get_climate_profile(**_BELLS_ARGS) | |
| assert "error" not in obs | |
| _outs, last = _last_emit(monkeypatch, obs, "get_climate_profile", _BELLS_ARGS) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(t.type == "barpolar" for t in fig.data) | |
| assert "📊" in last[CHATBOT][-1]["content"] | |
| def test_seafloor_turn_charts_depth_map(monkeypatch): | |
| obs = _tools.get_seafloor_profile(**_BELLS_ARGS) | |
| assert "error" not in obs | |
| grid = {"lats": [-38.31, -38.30, -38.29], "lngs": [144.29, 144.30, 144.31], | |
| "elev": [-20.0, -12.0, -6.0, -14.0, -8.0, -4.0, | |
| -10.0, -5.0, -1.0], | |
| "n": 3, "center": {"lat": -38.30, "lng": 144.30}} | |
| monkeypatch.setattr(C, "get_seafloor", | |
| lambda br, radius_km=1.2: {"grid": grid, "analysis": "", | |
| "stats": {}}) | |
| _outs, last = _last_emit(monkeypatch, obs, "get_seafloor_profile", _BELLS_ARGS) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(t.type == "heatmap" for t in fig.data) | |
| assert "📊" in last[CHATBOT][-1]["content"] | |
| def test_rank_turn_charts_leaderboard(monkeypatch): | |
| obs = {"region": "Victoria", "skill": "intermediate", "count": 2, | |
| "spots": [ | |
| {"name": "Bells Beach", "region": "Surf Coast", | |
| "best_score": 7.5, "best_time": "2026-09-12T06:00"}, | |
| {"name": "Anglesea", "region": "Surf Coast", | |
| "best_score": 5.0, "best_time": "2026-09-12T07:00"}, | |
| ]} | |
| _outs, last = _last_emit(monkeypatch, obs, "rank_region_week", {"region": "Victoria"}) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(getattr(t, "orientation", None) == "h" for t in fig.data) | |
| assert "📊" in last[CHATBOT][-1]["content"] | |
| def test_similarity_turn_charts_comparison(monkeypatch): | |
| obs = [ | |
| {"name": "Kirra", "region": "Gold Coast", "breakType": "Point break", | |
| "skillLevel": "intermediate", "peakType": "Right", "similarity": 8.5}, | |
| {"name": "Lennox Head", "region": "Northern Rivers", | |
| "breakType": "Point break", "skillLevel": "intermediate", | |
| "peakType": "Right", "similarity": 7.5}, | |
| ] | |
| _outs, last = _last_emit(monkeypatch, obs, "find_similar_spots", | |
| {"spot_name": "Snapper Rocks"}) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(getattr(t, "orientation", None) == "h" for t in fig.data) | |
| assert "📊" in last[CHATBOT][-1]["content"] | |
| def test_explain_turn_charts_component_split(monkeypatch): | |
| obs = _tools.explain_score(**_BELLS_ARGS) | |
| assert "error" not in obs and obs.get("components") | |
| _outs, last = _last_emit(monkeypatch, obs, "explain_score", _BELLS_ARGS) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(t.type == "bar" and getattr(t, "orientation", None) is None | |
| for t in fig.data) # vertical bars leave orientation=None | |
| assert "📊" in last[CHATBOT][-1]["content"] | |
| def test_knowledge_only_turn_charts_viewed_spot(monkeypatch): | |
| # Nothing chartable in the tool payloads → the viewed spot's own | |
| # scored week (engine fetch stubbed) closes the guarantee. | |
| hours = [{"time": f"2026-09-09T{h:02d}:00", "score": 4.0 + h % 4, | |
| "wave_height_m": 1.2, "wave_period_s": 9.0, | |
| "wind_speed_kt": 7.0, "wind_direction_deg": 220, | |
| "daylight": True} for h in range(6, 18)] | |
| monkeypatch.setattr( | |
| C, "get_scored_week", | |
| lambda br, skill="intermediate", days=7: {"scored": hours, | |
| "skill": skill}) | |
| obs = _tools.get_spot_knowledge(**_BELLS_ARGS) | |
| assert "error" not in obs | |
| _outs, last = _last_emit(monkeypatch, obs, "get_spot_knowledge", _BELLS_ARGS) | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) and fig.data | |
| assert "Charted" in last[CHATBOT][-1]["content"] | |
| def test_scored_turn_still_charts_strip_midway(monkeypatch): | |
| # When score_week runs, the week strip is charted mid-turn and the | |
| # fallback must not replace it with a second end-of-turn chart (so | |
| # the stubbed engine fetch below must never fire, and later emits | |
| # only gr.skip() the plot slot). | |
| hours = [{"time": f"2026-09-09T{h:02d}:00", "score": 5.0, | |
| "wave_height_m": 1.0, "wave_period_s": 8.0, | |
| "wind_speed_kt": 6.0, "wind_direction_deg": 200, | |
| "daylight": True} for h in range(6, 18)] | |
| obs = {"spot": {"name": "Bells Beach", "region": "Surf Coast"}, | |
| "skill": "intermediate", "scored": hours, | |
| "daily": [], "best": hours[0], "stub": True} | |
| monkeypatch.setattr(C, "get_scored_week", lambda *a, **k: (_ for _ in ()).throw( | |
| AssertionError("fallback fetch must not run when hours were charted"))) | |
| outs, last = _last_emit(monkeypatch, obs, "score_week", _BELLS_ARGS) | |
| strip = contracts.AGENT_KEYS.index("agent_strip") | |
| charted = [o[strip] for o in outs if isinstance(o[strip], go.Figure)] | |
| assert charted, "mid-turn week strip never rendered" | |
| assert not isinstance(last[strip], go.Figure) or charted[-1] is last[strip] | |
| def _scored_obs(spot_name: str, region: str, score: float) -> dict: | |
| hours = [{"time": f"2026-09-09T{h:02d}:00", "score": score - (h % 3) * 0.5, | |
| "wave_height_m": 1.2, "wave_period_s": 9.0, | |
| "wind_speed_kt": 7.0, "wind_direction_deg": 220, | |
| "daylight": True} for h in range(6, 18)] | |
| return {"spot": {"name": spot_name, "region": region}, | |
| "skill": "intermediate", "scored": hours, | |
| "daily": [], "best": hours[0], "stub": True} | |
| def test_where_query_sweeps_region_when_agent_didnt_rank(monkeypatch): | |
| # A recommendation question the agent answered without a rank sweep | |
| # still closes on the recommended-scores leaderboard: the panel runs a | |
| # deterministic engine sweep of the region named in the question | |
| # (NSW -> New South Wales), never the viewed spot's unrelated week. | |
| sweep = {"region": "New South Wales", "skill": "intermediate", "count": 2, | |
| "spots": [ | |
| {"name": "Bondi Beach", "region": "Sydney", | |
| "best_score": 6.5, "best_time": "2026-09-12T07:00"}, | |
| {"name": "Byron Bay", "region": "Northern Rivers", | |
| "best_score": 5.5, "best_time": "2026-09-12T08:00"}, | |
| ]} | |
| monkeypatch.setattr(C, "api_rank_region_week", | |
| lambda region, skill="intermediate", limit=10: sweep) | |
| monkeypatch.setattr(C, "get_scored_week", lambda *a, **k: (_ for _ in ()).throw( | |
| AssertionError("viewed-spot fallback must not run when the sweep hits"))) | |
| obs = _tools.get_spot_knowledge(**_BELLS_ARGS) | |
| assert "error" not in obs | |
| _outs, last = _last_emit( | |
| monkeypatch, obs, "get_spot_knowledge", _BELLS_ARGS, | |
| query="Where should I surf in NSW this weekend as a beginner?") | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(getattr(t, "orientation", None) == "h" for t in fig.data) | |
| assert "recommended breaks — New South Wales" in last[CHATBOT][-1]["content"] | |
| def test_seafloor_query_prefers_depth_map_over_leaderboard(monkeypatch): | |
| # When the question asks about the seafloor, the depth map wins even | |
| # with a rank sweep payload on the table — the chart matches the | |
| # question, not the tool order. | |
| rank_obs = {"region": "Victoria", "skill": "intermediate", "count": 1, | |
| "spots": [{"name": "Bells Beach", "region": "Surf Coast", | |
| "best_score": 7.5, "best_time": "2026-09-12T06:00"}]} | |
| seafloor_obs = _tools.get_seafloor_profile(**_BELLS_ARGS) | |
| assert "error" not in seafloor_obs | |
| grid = {"lats": [-38.31, -38.30, -38.29], "lngs": [144.29, 144.30, 144.31], | |
| "elev": [-20.0, -12.0, -6.0, -14.0, -8.0, -4.0, | |
| -10.0, -5.0, -1.0], | |
| "n": 3, "center": {"lat": -38.30, "lng": 144.30}} | |
| monkeypatch.setattr(C, "get_seafloor", | |
| lambda br, radius_km=1.2: {"grid": grid, "analysis": "", | |
| "stats": {}}) | |
| events = _tool_events("rank_region_week", {"region": "Victoria"}, rank_obs) | |
| events += _tool_events("get_seafloor_profile", _BELLS_ARGS, seafloor_obs) | |
| outs = _run(monkeypatch, events, | |
| "**Best: Bells Beach @ Sat am — 6/10.**", | |
| "What's the seafloor like at Bells Beach?") | |
| fig = outs[-1][AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(t.type == "heatmap" for t in fig.data) | |
| assert not any(getattr(t, "orientation", None) == "h" for t in fig.data) | |
| def test_compare_query_charts_multi_spot_scores(monkeypatch): | |
| # Two score_week calls in one turn -> the turn closes on the | |
| # side-by-side best-score compare, not just the last spot's strip. | |
| events = [] | |
| for name, region, score in (("Bells Beach", "Victoria", 7.0), | |
| ("Snapper Rocks", "Queensland", 5.0)): | |
| args = {"spot_name": name, "region": region} | |
| events += _tool_events("score_week", args, | |
| _scored_obs(name, region, score)) | |
| outs = _run(monkeypatch, events, | |
| "**Best: Bells Beach @ Sun am — 7/10.**", | |
| "Compare Bells Beach and Snapper Rocks this weekend.") | |
| last = outs[-1] | |
| fig = last[AGENT_STRIP] | |
| assert isinstance(fig, go.Figure) | |
| assert any(getattr(t, "orientation", None) == "h" for t in fig.data) | |
| assert "best-score compare (2 spots)" in last[CHATBOT][-1]["content"] | |