Ryan Malloy
66ec2ba9e7
New visualize_tides and visualize_conditions MCP tools that generate PNG (inline via MCP ImageContent) or interactive HTML (Plotly) charts. Optional dependency group [viz] keeps the base install lightweight. - charts/ package: rendering logic separated from MCP tool wiring - Shared marine color palette (ocean blue, teal, slate, sand, coral) - 14 new tests (parsing, PNG/HTML rendering, tool registration) - Example chart images in README with realistic synthetic tidal data
142 lines
4.6 KiB
Python
142 lines
4.6 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 9 tools should be registered."""
|
|
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",
|
|
}
|
|
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)
|