"""Offline dry-run tests for Worker C (no network, no LLM, no token).
Covers: 11 tools importable + plausibly shaped offline, system-prompt token
budget, golden-question routing, depth budget profiles, score_week per-turn
budget guard, llm factory defaults, narrator marker/think stripping.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts"))
import eval_agent # noqa: E402
from wavereader import agent as agent_mod # noqa: E402
from wavereader import llm as llm_mod # noqa: E402
from wavereader import narrate as narrate_mod # noqa: E402
from wavereader import tools as tools_mod # noqa: E402
EXPECTED_TOOLS = [
"get_spot_knowledge",
"get_climate_profile",
"get_seafloor_profile",
"get_session_brief",
"score_week",
"rank_region_week",
"explain_score",
"find_best_windows",
"find_spots",
"find_similar_spots",
"list_regions",
]
def test_eleven_tools_registered():
assert tools_mod.TOOL_NAMES == EXPECTED_TOOLS
assert len(tools_mod.TOOLS) == 11
for t in tools_mod.TOOLS:
assert t.description and t.inputs is not None and t.output_type in ("object", "array", "string")
def test_list_regions_offline_shape():
out = tools_mod.list_regions()
assert isinstance(out, dict) and "states" in out and "regions_by_state" in out
def test_find_spots_offline_shape():
out = tools_mod.find_spots(query="Queensland", limit=3)
assert isinstance(out, list) and len(out) <= 3
def test_spot_tools_offline_plausible_shapes():
# Unknown spot -> typed error dicts (no network touched).
assert "error" in tools_mod.get_spot_knowledge(spot_name="No Such Break XYZ")
assert "error" in tools_mod.score_week(spot_name="No Such Break XYZ")
assert "error" in tools_mod.explain_score(spot_name="No Such Break XYZ")
assert "error" in tools_mod.get_climate_profile(spot_name="No Such Break XYZ")
assert "error" in tools_mod.get_seafloor_profile(spot_name="No Such Break XYZ")
assert "error" in tools_mod.get_session_brief(spot_name="No Such Break XYZ")
assert "error" in tools_mod.rank_region_week(region="")
assert "error" in tools_mod.find_best_windows(spot_name="No Such Break XYZ")
miss = tools_mod.find_similar_spots(spot_name="No Such Break XYZ")
assert isinstance(miss, list) and miss and "error" in miss[0]
def test_known_spot_offline_shapes():
if not tools_mod.list_regions().get("states"):
return # no dataset present; nothing to resolve against
first_state = tools_mod.list_regions()["states"][0]
spots = tools_mod.find_spots(query=first_state, limit=1)
assert spots, "dataset should yield at least one spot"
name = spots[0]["name"]
region = spots[0].get("region") or first_state
knowledge = tools_mod.get_spot_knowledge(spot_name=name, region=region)
assert knowledge.get("name") == name
scored = tools_mod.score_week(spot_name=name, region=region, skill="intermediate")
assert isinstance(scored.get("scored"), list) and scored["scored"]
assert scored.get("best") and scored.get("daily")
explained = tools_mod.explain_score(spot_name=name, region=region)
assert explained.get("score") is not None and "components" in explained
windows = tools_mod.find_best_windows(spot_name=name, region=region, daypart="morning")
assert isinstance(windows.get("windows"), list)
climate = tools_mod.get_climate_profile(spot_name=name, region=region)
assert "climate" in climate and "findings" in climate
seafloor = tools_mod.get_seafloor_profile(spot_name=name, region=region)
assert seafloor.get("shelf_class") or "error" in seafloor
brief = tools_mod.get_session_brief(spot_name=name, region=region)
assert brief.get("wetsuit") or "error" in brief
ranked = tools_mod.rank_region_week(region=first_state, limit=3)
if "error" in ranked:
pytest.skip(f"forecast backend offline: {ranked['error']}")
assert isinstance(ranked.get("spots"), list)
similar = tools_mod.find_similar_spots(spot_name=name, region=region, limit=2)
assert isinstance(similar, list)
def test_system_prompt_token_budget():
est = agent_mod.prompt_token_estimate(agent_mod.SYSTEM_PROMPT)
assert est <= 1500, f"system prompt ~{est} tokens, budget is 1500"
def test_golden_routing_10_of_10():
missed = []
for question, expected in eval_agent.GOLDEN:
predicted = eval_agent.predict_tools(question)
if not all(t in predicted for t in expected):
missed.append((question, expected, predicted))
assert len(eval_agent.GOLDEN) == 10
assert not missed, f"routing misses: {missed}"
def test_budget_profiles():
quick = agent_mod.resolve_budget("quick")[1]
deep = agent_mod.resolve_budget("deep")[1]
standard = agent_mod.resolve_budget("standard")[1]
bogus_key, standard_fb = agent_mod.resolve_budget("bogus")
assert bogus_key == "standard" and standard_fb is standard
assert quick["max_steps"] < standard["max_steps"] < deep["max_steps"]
assert quick["max_tokens"] < standard["max_tokens"] < deep["max_tokens"]
assert quick["score_week_calls"] < standard["score_week_calls"] < deep["score_week_calls"]
# standard preserves the original legacy constants
assert standard == {
"max_steps": agent_mod.MAX_STEPS,
"max_tokens": agent_mod.MAX_TOKENS,
"score_week_calls": agent_mod.MAX_SCORE_WEEK_CALLS,
}
def test_score_week_budget_guard():
counter: dict = {}
guarded = agent_mod._guarded_score_week(counter)
kwargs = {"spot_name": "No Such Break XYZ", "region": None, "skill": None}
guarded(**kwargs)
guarded(**kwargs)
third = guarded(**kwargs)
assert counter["score_week"] == 2
assert "error" in third and "Budget exceeded" in third["error"]
def test_score_week_budget_guard_profile_cap():
counter: dict = {}
guarded = agent_mod._guarded_score_week(counter, cap=1) # quick profile
kwargs = {"spot_name": "No Such Break XYZ", "region": None, "skill": None}
guarded(**kwargs)
second = guarded(**kwargs)
assert counter["score_week"] == 1
assert "error" in second
assert "at most 1" in second["error"] and "Budget exceeded" in second["error"]
def test_llm_factory_defaults_no_token():
env_backup = {k: os.environ.get(k) for k in ("WR_MODEL", "WR_PROVIDER", "WR_BASE_URL")}
try:
for k in ("WR_MODEL", "WR_PROVIDER", "WR_BASE_URL"):
os.environ.pop(k, None)
assert llm_mod.get_model_id() == llm_mod.DEFAULT_MODEL
assert "Nemotron-3-Ultra" in llm_mod.get_model_id()
assert llm_mod.get_provider() == "deepinfra"
assert llm_mod.get_base_url() is None
assert "Lightning" not in llm_mod.get_model_id() and "Qwen" not in llm_mod.get_model_id()
os.environ["WR_MODEL"] = "custom/model"
os.environ["WR_PROVIDER"] = "custom-provider"
os.environ["WR_BASE_URL"] = "http://localhost:11434/v1"
assert llm_mod.get_model_id() == "custom/model"
assert llm_mod.get_provider() == "custom-provider"
assert llm_mod.get_base_url() == "http://localhost:11434/v1"
client = llm_mod.get_client(hf_token="dummy")
assert client is not None
finally:
for k, v in env_backup.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
def test_narrator_stripping():
dirty = "planning here@@REPORT@@\n- Sat: fun.\n\n**Recommendation: Sat -- light winds.**\n@@END@@"
clean = narrate_mod._clean_output(dirty)
assert "" not in clean and "@@REPORT@@" not in clean
assert "Recommendation" in clean
assert narrate_mod._clean_output("We need to produce bullets. Real text.") != ""
prompt = narrate_mod.build_report_prompt({"name": "Bells"}, "intermediate", [], None, 1)
assert "Bells" in prompt
assert narrate_mod.narrate_day({"date": "2026-09-07", "score": 7}) .startswith("2026-09-07")
def test_agent_factory_builds_with_injected_model():
from smolagents import InferenceClientModel
calls: list[dict] = []
class FakeModel(InferenceClientModel):
def __init__(self):
pass
created = agent_mod.create_agent(model=FakeModel(), max_steps=6)
assert created.max_steps == 6
tool_names = sorted(t.name for t in created.tools.values()) if isinstance(created.tools, dict) else sorted(t.name for t in created.tools)
# ToolCallingAgent adds its built-in final_answer tool; ours must all be present.
assert all(t in tool_names for t in EXPECTED_TOOLS), tool_names