Oleksandr Kozachuk
121 lines
5.3 KiB
Python
121 lines
5.3 KiB
Python
"""Unit coverage for SPEC-016 weather tool (WEA-01..04)."""
|
|
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fjerkroa_bot.openai_responder import OpenAIResponder
|
|
from fjerkroa_bot.weather import GET_WEATHER_TOOL, Weather, _reduce
|
|
|
|
CONFIG = {"openai-token": "t", "model": "m", "system": "s", "history-limit": 5}
|
|
LOCATIONS = [["Sleneset", 66.58, 12.68], ["Berlin", 52.52, 13.41]]
|
|
|
|
MET_DATA = {
|
|
"properties": {
|
|
"timeseries": [
|
|
{
|
|
"time": f"2026-07-17T{10 + i if 10 + i < 24 else 10 + i - 24:02d}:00:00Z",
|
|
"data": {
|
|
"instant": {"details": {"air_temperature": 14.0 + i, "wind_speed": 5.0}},
|
|
"next_1_hours": {"summary": {"symbol_code": "lightrain"}, "details": {"precipitation_amount": 0.3}},
|
|
},
|
|
}
|
|
for i in range(30)
|
|
]
|
|
}
|
|
}
|
|
|
|
|
|
def _tool_names(responder):
|
|
return [f["name"] for f in responder._available_tools()]
|
|
|
|
|
|
class TestToolOffered(unittest.TestCase):
|
|
def test_gate_needs_flag_and_locations(self):
|
|
"""WEA-01: get_weather offered only with enable-weather AND locations."""
|
|
self.assertNotIn("get_weather", _tool_names(OpenAIResponder(CONFIG, "chat")))
|
|
flag_only = OpenAIResponder(dict(CONFIG, **{"enable-weather": True}), "chat")
|
|
self.assertNotIn("get_weather", _tool_names(flag_only))
|
|
on = OpenAIResponder(dict(CONFIG, **{"enable-weather": True, "weather-locations": LOCATIONS}), "chat")
|
|
self.assertIn("get_weather", _tool_names(on))
|
|
self.assertEqual(GET_WEATHER_TOOL["name"], "get_weather")
|
|
|
|
|
|
class TestReduce(unittest.TestCase):
|
|
def test_compact_shape(self):
|
|
"""WEA-02: now + few forecast points; temperature/wind/conditions/precip only."""
|
|
out = _reduce(MET_DATA, "Sleneset")
|
|
self.assertEqual(out["location"], "Sleneset")
|
|
self.assertEqual(out["now"]["temp_c"], 14.0)
|
|
self.assertEqual(out["now"]["wind_ms"], 5.0)
|
|
self.assertEqual(out["now"]["conditions"], "lightrain")
|
|
self.assertEqual(out["now"]["precip_mm"], 0.3)
|
|
self.assertEqual(len(out["forecast"]), 3) # +6h, +12h, +24h
|
|
self.assertEqual(out["forecast"][0]["temp_c"], 20.0)
|
|
self.assertNotIn("error", out)
|
|
|
|
def test_location_name_sanitized_and_empty_series(self):
|
|
"""WEA-02: name passes sanitizer; empty timeseries -> error dict."""
|
|
out = _reduce(MET_DATA, "@everyone town")
|
|
self.assertNotIn("@everyone", out["location"])
|
|
self.assertIn("error", _reduce({"properties": {"timeseries": []}}, "x"))
|
|
|
|
|
|
class TestLocationPick(unittest.TestCase):
|
|
def setUp(self):
|
|
self.weather = Weather(lambda: {"enable-weather": True, "weather-locations": LOCATIONS})
|
|
|
|
def test_substring_case_insensitive(self):
|
|
"""WEA-03: case-insensitive substring match."""
|
|
self.assertEqual(self.weather._pick("berlin")[0], "Berlin")
|
|
self.assertEqual(self.weather._pick("slen")[0], "Sleneset")
|
|
|
|
def test_unknown_or_absent_defaults_to_first(self):
|
|
"""WEA-03: unknown/absent location -> first configured entry."""
|
|
self.assertEqual(self.weather._pick(None)[0], "Sleneset")
|
|
self.assertEqual(self.weather._pick("Atlantis")[0], "Sleneset")
|
|
|
|
def test_bad_entries_skipped(self):
|
|
"""WEA-03: malformed location entries are ignored, not fatal."""
|
|
weather = Weather(lambda: {"enable-weather": True, "weather-locations": [["broken"], ["OK", 1.0, 2.0]]})
|
|
self.assertEqual(weather._pick(None)[0], "OK")
|
|
|
|
|
|
class TestForecast(unittest.IsolatedAsyncioTestCase):
|
|
async def test_error_returned_not_raised(self):
|
|
"""WEA-04: network failure -> {error}, never an exception."""
|
|
weather = Weather(lambda: {"enable-weather": True, "weather-locations": LOCATIONS})
|
|
with patch.object(Weather, "_fetch_json", new_callable=AsyncMock, side_effect=RuntimeError("boom")):
|
|
out = await weather.forecast("Berlin")
|
|
self.assertIn("error", out)
|
|
|
|
async def test_forecast_happy_path(self):
|
|
"""WEA-02/03: full flow with mocked API."""
|
|
weather = Weather(lambda: {"enable-weather": True, "weather-locations": LOCATIONS})
|
|
with patch.object(Weather, "_fetch_json", new_callable=AsyncMock, return_value=MET_DATA):
|
|
out = await weather.forecast("berlin")
|
|
self.assertEqual(out["location"], "Berlin")
|
|
self.assertEqual(out["now"]["temp_c"], 14.0)
|
|
|
|
|
|
class TestMetering(unittest.IsolatedAsyncioTestCase):
|
|
async def test_daily_cap(self):
|
|
"""WEA-04: per-user daily cap refuses beyond weather-daily-per-user."""
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = dict(
|
|
CONFIG,
|
|
**{
|
|
"enable-weather": True,
|
|
"weather-locations": LOCATIONS,
|
|
"weather-daily-per-user": 1,
|
|
"history-directory": tmp,
|
|
},
|
|
)
|
|
responder = OpenAIResponder(config, "chat")
|
|
with patch.object(Weather, "_fetch_json", new_callable=AsyncMock, return_value=MET_DATA):
|
|
first = await responder._dispatch_tool("get_weather", {}, "alice")
|
|
second = await responder._dispatch_tool("get_weather", {}, "alice")
|
|
self.assertNotIn("error", first)
|
|
self.assertIn("error", second)
|