Ryan Malloy
c7320e599b
4 new tools (tidal_phase, deployment_briefing, catch_tidal_context, water_level_anomaly) and 2 prompts (smartpot_deployment, crab_pot_analysis) for autonomous crab pot deployment planning and catch correlation. Pure tidal phase classification in tidal.py with no MCP dependencies. 65 tests passing, lint clean.
146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
"""Tests for station discovery and data retrieval tools."""
|
|
|
|
import json
|
|
|
|
from fastmcp import Client
|
|
|
|
|
|
async def test_tool_registration(mcp_client: Client):
|
|
"""All 13 tools should be registered (9 original + 4 SmartPot)."""
|
|
tools = await mcp_client.list_tools()
|
|
tool_names = {t.name for t in tools}
|
|
expected = {
|
|
"search_stations",
|
|
"find_nearest_stations",
|
|
"get_station_info",
|
|
"get_tide_predictions",
|
|
"get_observed_water_levels",
|
|
"get_meteorological_data",
|
|
"marine_conditions_snapshot",
|
|
"visualize_tides",
|
|
"visualize_conditions",
|
|
"tidal_phase",
|
|
"deployment_briefing",
|
|
"catch_tidal_context",
|
|
"water_level_anomaly",
|
|
}
|
|
assert expected == tool_names
|
|
|
|
|
|
async def test_search_stations_by_state(mcp_client: Client):
|
|
result = await mcp_client.call_tool("search_stations", {"state": "RI"})
|
|
stations = json.loads(result.content[0].text)
|
|
assert len(stations) == 2
|
|
assert all(s["state"] == "RI" for s in stations)
|
|
|
|
|
|
async def test_search_stations_by_name(mcp_client: Client):
|
|
result = await mcp_client.call_tool("search_stations", {"query": "providence"})
|
|
stations = json.loads(result.content[0].text)
|
|
assert len(stations) == 1
|
|
assert stations[0]["id"] == "8454000"
|
|
|
|
|
|
async def test_search_stations_no_match(mcp_client: Client):
|
|
result = await mcp_client.call_tool("search_stations", {"query": "nonexistent"})
|
|
# FastMCP may return empty content list for empty results
|
|
if result.content:
|
|
stations = json.loads(result.content[0].text)
|
|
assert len(stations) == 0
|
|
else:
|
|
# Empty content means no matches — that's correct
|
|
pass
|
|
|
|
|
|
async def test_find_nearest_stations(mcp_client: Client):
|
|
# Search near Providence coordinates
|
|
result = await mcp_client.call_tool(
|
|
"find_nearest_stations",
|
|
{"latitude": 41.8, "longitude": -71.4},
|
|
)
|
|
stations = json.loads(result.content[0].text)
|
|
assert len(stations) >= 1
|
|
# Closest should be Providence
|
|
assert stations[0]["id"] == "8454000"
|
|
assert "distance_nm" in stations[0]
|
|
# Should be sorted by distance
|
|
distances = [s["distance_nm"] for s in stations]
|
|
assert distances == sorted(distances)
|
|
|
|
|
|
async def test_get_station_info(mcp_client: Client):
|
|
result = await mcp_client.call_tool("get_station_info", {"station_id": "8454000"})
|
|
info = json.loads(result.content[0].text)
|
|
assert info["id"] == "8454000"
|
|
assert info["name"] == "Providence"
|
|
assert "sensors" in info
|
|
|
|
|
|
async def test_get_tide_predictions(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"get_tide_predictions", {"station_id": "8454000"}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert "predictions" in data
|
|
preds = data["predictions"]
|
|
assert len(preds) == 4
|
|
assert preds[0]["type"] == "H"
|
|
assert preds[1]["type"] == "L"
|
|
|
|
|
|
async def test_get_observed_water_levels(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"get_observed_water_levels", {"station_id": "8454000"}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert "data" in data
|
|
assert len(data["data"]) == 2
|
|
|
|
|
|
async def test_get_meteorological_data_wind(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"get_meteorological_data",
|
|
{"station_id": "8454000", "product": "wind"},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert "data" in data
|
|
assert data["data"][0]["dr"] == "SW"
|
|
|
|
|
|
async def test_marine_conditions_snapshot(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"marine_conditions_snapshot", {"station_id": "8454000"}
|
|
)
|
|
snapshot = json.loads(result.content[0].text)
|
|
assert snapshot["station_id"] == "8454000"
|
|
assert "fetched_utc" in snapshot
|
|
assert "predictions" in snapshot
|
|
assert "water_level" in snapshot
|
|
assert "wind" in snapshot
|
|
|
|
|
|
async def test_resource_registration(mcp_client: Client):
|
|
"""Resources should be registered."""
|
|
resources = await mcp_client.list_resources()
|
|
uris = {str(r.uri) for r in resources}
|
|
assert "noaa://stations" in uris
|
|
|
|
|
|
async def test_prompt_registration(mcp_client: Client):
|
|
"""Prompts should be registered."""
|
|
prompts = await mcp_client.list_prompts()
|
|
prompt_names = {p.name for p in prompts}
|
|
assert "plan_fishing_trip" in prompt_names
|
|
assert "marine_safety_check" in prompt_names
|
|
|
|
|
|
async def test_prompt_plan_fishing_trip(mcp_client: Client):
|
|
result = await mcp_client.get_prompt(
|
|
"plan_fishing_trip", {"location": "Narragansett Bay"}
|
|
)
|
|
assert len(result.messages) >= 1
|
|
text = result.messages[0].content
|
|
if hasattr(text, "text"):
|
|
text = text.text
|
|
assert "Narragansett Bay" in str(text)
|