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.
254 lines
8.1 KiB
Python
254 lines
8.1 KiB
Python
"""Integration tests for SmartPot tidal intelligence tools and prompts."""
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
# Fixed "now" in the middle of the ebb phase:
|
|
# Between H at 04:30 and L at 10:42 on 2026-02-21
|
|
MOCK_NOW = datetime(2026, 2, 21, 7, 30, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def freeze_time():
|
|
"""Patch datetime.now() in the smartpot module to return a fixed time."""
|
|
with patch("mcnoaa_tides.tools.smartpot.datetime") as mock_dt:
|
|
mock_dt.now.return_value = MOCK_NOW
|
|
mock_dt.strptime = datetime.strptime
|
|
mock_dt.fromisoformat = datetime.fromisoformat
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
yield mock_dt
|
|
|
|
|
|
# --- Tool registration ---
|
|
|
|
|
|
async def test_smartpot_tools_registered(mcp_client: Client):
|
|
"""All 4 SmartPot tools should appear in the tool list."""
|
|
tools = await mcp_client.list_tools()
|
|
names = {t.name for t in tools}
|
|
assert "tidal_phase" in names
|
|
assert "deployment_briefing" in names
|
|
assert "catch_tidal_context" in names
|
|
assert "water_level_anomaly" in names
|
|
|
|
|
|
async def test_total_tool_count(mcp_client: Client):
|
|
"""Should have 13 tools total (9 original + 4 SmartPot)."""
|
|
tools = await mcp_client.list_tools()
|
|
assert len(tools) == 13
|
|
|
|
|
|
# --- tidal_phase ---
|
|
|
|
|
|
async def test_tidal_phase_by_station(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"tidal_phase", {"station_id": "8454000"}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert data["station"]["id"] == "8454000"
|
|
assert data["phase"] in ("ebb", "flood", "slack_high", "slack_low")
|
|
assert data["previous"] is not None
|
|
assert data["next"] is not None
|
|
assert "timestamp_utc" in data
|
|
|
|
|
|
async def test_tidal_phase_by_gps(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"tidal_phase", {"latitude": 41.8, "longitude": -71.4}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
# Should resolve to Providence (nearest to 41.8, -71.4)
|
|
assert data["station"]["id"] == "8454000"
|
|
assert "station_distance_nm" in data
|
|
assert data["phase"] in ("ebb", "flood", "slack_high", "slack_low")
|
|
|
|
|
|
async def test_tidal_phase_missing_params(mcp_client: Client):
|
|
"""Should error when neither station_id nor lat/lon provided."""
|
|
with pytest.raises(ToolError):
|
|
await mcp_client.call_tool("tidal_phase", {})
|
|
|
|
|
|
# --- deployment_briefing ---
|
|
|
|
|
|
async def test_deployment_briefing(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"deployment_briefing",
|
|
{"latitude": 41.8, "longitude": -71.4, "soak_hours": 48},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert data["station"]["id"] == "8454000"
|
|
assert data["station_distance_nm"] is not None
|
|
assert data["assessment"] in ("GO", "CAUTION", "NO-GO")
|
|
assert "soak_window" in data
|
|
assert data["soak_window"]["hours"] == 48
|
|
assert "tide_schedule" in data
|
|
assert isinstance(data["advisories"], list)
|
|
|
|
|
|
async def test_deployment_briefing_cold_water_advisory(mcp_client: Client):
|
|
"""Mock water temp is 38.7°F — should trigger cold-water advisory."""
|
|
result = await mcp_client.call_tool(
|
|
"deployment_briefing",
|
|
{"latitude": 41.8, "longitude": -71.4},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
cold_advisories = [a for a in data["advisories"] if "cold-water" in a.lower()]
|
|
assert len(cold_advisories) > 0, "Expected cold-water advisory for 38.7°F"
|
|
|
|
|
|
async def test_deployment_briefing_conditions(mcp_client: Client):
|
|
"""Should include conditions from meteorological data."""
|
|
result = await mcp_client.call_tool(
|
|
"deployment_briefing",
|
|
{"latitude": 41.8, "longitude": -71.4},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
conditions = data["conditions"]
|
|
# Wind data available in mock
|
|
if "wind" in conditions:
|
|
assert "speed_kn" in conditions["wind"]
|
|
# Water temp available in mock
|
|
if "water_temperature_f" in conditions:
|
|
assert conditions["water_temperature_f"] == 38.7
|
|
|
|
|
|
# --- catch_tidal_context ---
|
|
|
|
|
|
async def test_catch_tidal_context_enrichment(mcp_client: Client):
|
|
events = [
|
|
{
|
|
"timestamp": "2026-02-21 07:30",
|
|
"latitude": 41.8,
|
|
"longitude": -71.4,
|
|
"catch_count": 12,
|
|
"species": "blue_crab",
|
|
},
|
|
{
|
|
"timestamp": "2026-02-21 13:00",
|
|
"latitude": 41.8,
|
|
"longitude": -71.4,
|
|
"catch_count": 5,
|
|
},
|
|
]
|
|
result = await mcp_client.call_tool(
|
|
"catch_tidal_context", {"events": events}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert len(data) == 2
|
|
|
|
# First event should be enriched with tidal info
|
|
assert "tidal" in data[0]
|
|
assert "phase" in data[0]["tidal"]
|
|
|
|
# Passthrough fields preserved
|
|
assert data[0]["catch_count"] == 12
|
|
assert data[0]["species"] == "blue_crab"
|
|
assert data[1]["catch_count"] == 5
|
|
|
|
|
|
async def test_catch_tidal_context_iso_timestamp(mcp_client: Client):
|
|
"""ISO-8601 timestamps with 'T' separator should also work."""
|
|
events = [
|
|
{
|
|
"timestamp": "2026-02-21T07:30:00Z",
|
|
"latitude": 41.8,
|
|
"longitude": -71.4,
|
|
},
|
|
]
|
|
result = await mcp_client.call_tool(
|
|
"catch_tidal_context", {"events": events}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert "tidal" in data[0]
|
|
assert "phase" in data[0]["tidal"]
|
|
|
|
|
|
async def test_catch_tidal_context_missing_location(mcp_client: Client):
|
|
"""Events without lat/lon should get an error tidal entry."""
|
|
events = [{"timestamp": "2026-02-21 07:30"}]
|
|
result = await mcp_client.call_tool(
|
|
"catch_tidal_context", {"events": events}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert "error" in data[0]["tidal"]
|
|
|
|
|
|
async def test_catch_tidal_context_limit(mcp_client: Client):
|
|
"""Should reject >100 events."""
|
|
events = [
|
|
{"timestamp": "2026-02-21 07:30", "latitude": 41.8, "longitude": -71.4}
|
|
for _ in range(101)
|
|
]
|
|
with pytest.raises(ToolError):
|
|
await mcp_client.call_tool("catch_tidal_context", {"events": events})
|
|
|
|
|
|
# --- water_level_anomaly ---
|
|
|
|
|
|
async def test_water_level_anomaly(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"water_level_anomaly", {"station_id": "8454000"}
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert data["station_id"] == "8454000"
|
|
assert data["risk_level"] in ("normal", "elevated", "high")
|
|
assert "max_deviation_ft" in data
|
|
assert "mean_deviation_ft" in data
|
|
assert "direction" in data
|
|
assert "explanation" in data
|
|
|
|
|
|
async def test_water_level_anomaly_custom_threshold(mcp_client: Client):
|
|
result = await mcp_client.call_tool(
|
|
"water_level_anomaly",
|
|
{"station_id": "8454000", "threshold_ft": 0.1},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
assert data["threshold_ft"] == 0.1
|
|
|
|
|
|
# --- Prompt registration ---
|
|
|
|
|
|
async def test_smartpot_prompts_registered(mcp_client: Client):
|
|
"""Both SmartPot prompts should appear in the prompt list."""
|
|
prompts = await mcp_client.list_prompts()
|
|
names = {p.name for p in prompts}
|
|
assert "smartpot_deployment" in names
|
|
assert "crab_pot_analysis" in names
|
|
|
|
|
|
async def test_total_prompt_count(mcp_client: Client):
|
|
"""Should have 4 prompts total (2 original + 2 SmartPot)."""
|
|
prompts = await mcp_client.list_prompts()
|
|
assert len(prompts) == 4
|
|
|
|
|
|
async def test_smartpot_deployment_prompt(mcp_client: Client):
|
|
result = await mcp_client.get_prompt(
|
|
"smartpot_deployment",
|
|
{"latitude": "41.8", "longitude": "-71.4", "soak_hours": "48"},
|
|
)
|
|
assert len(result.messages) >= 1
|
|
text = str(result.messages[0].content)
|
|
assert "41.8" in text
|
|
assert "-71.4" in text
|
|
assert "deployment_briefing" in text
|
|
|
|
|
|
async def test_crab_pot_analysis_prompt(mcp_client: Client):
|
|
result = await mcp_client.get_prompt("crab_pot_analysis", {})
|
|
assert len(result.messages) >= 1
|
|
text = str(result.messages[0].content)
|
|
assert "catch_tidal_context" in text
|