152 lines
8.0 KiB
Python
152 lines
8.0 KiB
Python
"""Weather checks use mocked JSON and an isolated cache, without network access."""
|
|
from dataclasses import replace
|
|
from datetime import datetime, timezone
|
|
from email.message import Message
|
|
from email.utils import format_datetime
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
from urllib.error import HTTPError
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
from talking_clock.audio import Audio
|
|
from talking_clock.config import Settings
|
|
from talking_clock.weather import WeatherClient
|
|
|
|
CURRENT = {"current": {"temperature_2m": 25, "weather_code": 3},
|
|
"current_units": {"temperature_2m": "°C"}}
|
|
PLACE = {"city": "Raleigh", "region": "North Carolina", "country": "US", "loc": "35.78,-78.64"}
|
|
|
|
|
|
class WeatherTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.client = WeatherClient(self.temporary.name)
|
|
self.client._request = Mock(return_value=CURRENT)
|
|
self.settings = Settings(latitude="35.78", longitude="-78.64")
|
|
|
|
def test_units_share_cache_and_conditions_are_plain_text(self):
|
|
with patch("talking_clock.weather.time.time", return_value=1000):
|
|
self.assertEqual(self.client.report(self.settings), "77 degrees Fahrenheit and cloudy")
|
|
self.assertEqual(self.client.report(replace(self.settings, temperature_unit="C")),
|
|
"25 degrees Celsius and cloudy")
|
|
self.client._request.assert_called_once()
|
|
self.assertIn("current=temperature_2m%2Cweather_code", self.client._request.call_args.args[0])
|
|
|
|
def test_negative_temperature_and_clear_condition(self):
|
|
self.client._request.return_value = {"current": {"temperature_2m": -10, "weather_code": 0},
|
|
"current_units": {"temperature_2m": "°C"}}
|
|
self.assertEqual(self.client.report(self.settings), "14 degrees Fahrenheit and clear")
|
|
|
|
def test_every_second_checks_and_restart_only_fetch_at_next_quarter(self):
|
|
for now in range(1000, 1800):
|
|
with patch("talking_clock.weather.time.time", return_value=now):
|
|
self.assertIsNotNone(self.client.report(self.settings))
|
|
self.assertEqual(self.client._request.call_count, 1)
|
|
restarted = WeatherClient(self.temporary.name)
|
|
restarted._request = self.client._request
|
|
with patch("talking_clock.weather.time.time", return_value=1799):
|
|
self.assertIsNotNone(restarted.report(self.settings))
|
|
self.assertEqual(self.client._request.call_count, 1)
|
|
with patch("talking_clock.weather.time.time", return_value=1800):
|
|
self.assertIsNotNone(restarted.report(self.settings))
|
|
self.assertEqual(self.client._request.call_count, 2)
|
|
|
|
def test_changed_coordinates_do_not_bypass_request_budget(self):
|
|
with patch("talking_clock.weather.time.time", return_value=1000):
|
|
self.assertIsNotNone(self.client.report(self.settings))
|
|
self.assertIsNone(self.client.report(replace(self.settings, latitude="40")))
|
|
self.client._request.assert_called_once()
|
|
|
|
def test_parallel_requests_share_disk_lock(self):
|
|
other = WeatherClient(self.temporary.name)
|
|
other._request = self.client._request
|
|
with patch("talking_clock.weather.time.time", return_value=1000):
|
|
threads = [threading.Thread(target=client.report, args=(self.settings,))
|
|
for client in (self.client, other, self.client, other)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(5)
|
|
self.assertFalse(thread.is_alive())
|
|
self.client._request.assert_called_once()
|
|
|
|
def test_ip_location_is_cached_for_a_day_and_used_without_geocoding(self):
|
|
self.client._request.side_effect = [PLACE, CURRENT, PLACE]
|
|
with patch("talking_clock.weather.time.time", return_value=1000):
|
|
place = self.client.detect_location()
|
|
self.assertEqual(place["location"], "Raleigh, North Carolina, US")
|
|
self.assertIsNotNone(self.client.report(Settings()))
|
|
self.assertEqual(self.client._request.call_count, 2)
|
|
with patch("talking_clock.weather.time.time", return_value=87399):
|
|
self.assertEqual(self.client.detect_location(), place)
|
|
self.assertEqual(self.client._request.call_count, 2)
|
|
with patch("talking_clock.weather.time.time", return_value=87400):
|
|
self.assertEqual(self.client.detect_location(), place)
|
|
self.assertEqual(self.client._request.call_count, 3)
|
|
|
|
def test_manual_city_uses_geocoding(self):
|
|
self.client._request.side_effect = [
|
|
{"results": [{"name": "Berlin", "country": "Germany", "latitude": 52.52, "longitude": 13.41}]},
|
|
CURRENT]
|
|
self.assertIsNotNone(self.client.report(Settings(location="Berlin, Germany")))
|
|
self.assertIn("geocoding-api.open-meteo.com", self.client._request.call_args_list[0].args[0])
|
|
self.assertIn("latitude=52.52", self.client._request.call_args_list[1].args[0])
|
|
|
|
def test_invalid_responses_are_not_spoken_or_retried_each_second(self):
|
|
for invalid in ({}, {"current": {"temperature_2m": None, "weather_code": 3}},
|
|
{"current": {"temperature_2m": 25, "weather_code": None}},
|
|
{"results": []}):
|
|
with self.subTest(invalid=invalid), tempfile.TemporaryDirectory() as directory:
|
|
client = WeatherClient(directory)
|
|
client._request = Mock(return_value=invalid)
|
|
with patch("talking_clock.weather.time.time", return_value=1000), self.assertLogs(level="WARNING"):
|
|
self.assertIsNone(client.report(self.settings))
|
|
with patch("talking_clock.weather.time.time", return_value=1001):
|
|
self.assertIsNone(client.report(self.settings))
|
|
client._request.assert_called_once()
|
|
|
|
def test_http_429_honors_retry_after_and_persists_failure_backoff(self):
|
|
for retry in ("7200", format_datetime(datetime.fromtimestamp(8200, timezone.utc))):
|
|
with self.subTest(retry=retry), tempfile.TemporaryDirectory() as directory:
|
|
client = WeatherClient(directory)
|
|
headers = Message()
|
|
headers["Retry-After"] = retry
|
|
error = HTTPError("https://example.test", 429, "limit", headers, None)
|
|
self.addCleanup(error.close)
|
|
client._request = Mock(side_effect=error)
|
|
with patch("talking_clock.weather.time.time", return_value=1000), self.assertLogs(level="WARNING"):
|
|
self.assertIsNone(client.report(self.settings))
|
|
restarted = WeatherClient(directory)
|
|
restarted._request = client._request
|
|
with patch("talking_clock.weather.time.time", return_value=8199):
|
|
self.assertIsNone(restarted.report(self.settings))
|
|
client._request.assert_called_once()
|
|
with patch("talking_clock.weather.time.time", return_value=8200), self.assertLogs(level="WARNING"):
|
|
self.assertIsNone(restarted.report(self.settings))
|
|
self.assertEqual(client._request.call_count, 2)
|
|
|
|
def test_cache_failure_does_not_contact_server(self):
|
|
blocker = Path(self.temporary.name) / "not-a-directory"
|
|
blocker.touch()
|
|
client = WeatherClient(blocker)
|
|
client._request = Mock()
|
|
with self.assertLogs(level="WARNING"):
|
|
self.assertIsNone(client.report(self.settings))
|
|
client._request.assert_not_called()
|
|
|
|
def test_weather_failure_still_speaks_the_time(self):
|
|
audio = Audio()
|
|
audio.weather.report = Mock(return_value=None)
|
|
audio.speak = Mock()
|
|
audio._announce(datetime(2026, 10, 23, 12), Settings(weather=True, chime=False))
|
|
self.assertEqual(audio.speak.call_args.args[0], "12 o'clock P M")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|