"""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()