From 7166c0ccc867cd3a978f144e51bf44eb8ca3166f Mon Sep 17 00:00:00 2001 From: Jules Dejaeghere Date: Sat, 3 May 2025 15:46:47 +0200 Subject: [PATCH] Improve tests --- irm_kmi_api/api.py | 108 ++++++++----------- irm_kmi_api/rain_graph.py | 39 ++++--- tests/fixtures/forecast.json | 6 +- tests/test_api.py | 188 ++++++++++++++++++++++++++++++++++ tests/test_current_weather.py | 20 ++-- tests/test_daily_forecast.py | 10 +- tests/test_hourly_forecast.py | 12 +-- tests/test_rain_graph.py | 111 ++++++++++++++++---- 8 files changed, 365 insertions(+), 129 deletions(-) create mode 100644 tests/test_api.py diff --git a/irm_kmi_api/api.py b/irm_kmi_api/api.py index 1a082e6..66d99d4 100644 --- a/irm_kmi_api/api.py +++ b/irm_kmi_api/api.py @@ -15,7 +15,7 @@ import aiohttp import async_timeout from .const import MAP_WARNING_ID_TO_SLUG as SLUG_MAP -from .const import OPTION_STYLE_SATELLITE, STYLE_TO_PARAM_MAP, WEEKDAYS +from .const import STYLE_TO_PARAM_MAP, WEEKDAYS from .data import (AnimationFrameData, CurrentWeatherData, Forecast, IrmKmiForecast, IrmKmiRadarForecast, RadarAnimationData, WarningData) @@ -154,11 +154,11 @@ class IrmKmiApiClientHa(IrmKmiApiClient): def get_country(self) -> str | None: return self._api_data.get('country', None) - async def get_current_weather(self, tz: ZoneInfo) -> CurrentWeatherData: + def get_current_weather(self, tz: ZoneInfo) -> CurrentWeatherData: """Parse the API data to build a CurrentWeatherData.""" - now_hourly = await self._get_now_hourly(tz) - uv_index = await self._get_uv_index() + now_hourly = self._get_now_hourly(tz) + uv_index = self._get_uv_index() try: pressure = float(now_hourly.get('pressure', None)) if now_hourly is not None else None @@ -222,7 +222,7 @@ class IrmKmiApiClientHa(IrmKmiApiClient): return current_weather - async def _get_uv_index(self) -> float | None: + def _get_uv_index(self) -> float | None: uv_index = None module_data = self._api_data.get('module', None) if not (module_data is None or not isinstance(module_data, list)): @@ -231,7 +231,7 @@ class IrmKmiApiClientHa(IrmKmiApiClient): uv_index = module.get('data', {}).get('levelValue') return uv_index - async def _get_now_hourly(self, tz: ZoneInfo) -> dict | None: + def _get_now_hourly(self, tz: ZoneInfo) -> dict | None: now_hourly = None hourly_forecast_data = self._api_data.get('for', {}).get('hourly') now = datetime.now(tz) @@ -245,11 +245,11 @@ class IrmKmiApiClientHa(IrmKmiApiClient): break return now_hourly - async def get_daily_forecast(self, tz: ZoneInfo, lang: str) -> List[IrmKmiForecast] | None: + def get_daily_forecast(self, tz: ZoneInfo, lang: str) -> List[IrmKmiForecast]: """Parse data from the API to create a list of daily forecasts""" data = self._api_data.get('for', {}).get('daily') if data is None or not isinstance(data, list) or len(data) == 0: - return None + return [] forecasts = list() forecast_day = datetime.now(tz) @@ -337,12 +337,12 @@ class IrmKmiApiClientHa(IrmKmiApiClient): return forecasts - async def get_hourly_forecast(self, tz: ZoneInfo) -> List[Forecast] | None: + def get_hourly_forecast(self, tz: ZoneInfo) -> List[Forecast]: """Parse data from the API to create a list of hourly forecasts""" data = self._api_data.get('for', {}).get('hourly') if data is None or not isinstance(data, list) or len(data) == 0: - return None + return [] forecasts = list() day = datetime.now(tz).replace(hour=0, minute=0, second=0, microsecond=0) @@ -389,12 +389,12 @@ class IrmKmiApiClientHa(IrmKmiApiClient): return forecasts - def get_radar_forecast(self) -> List[IrmKmiRadarForecast] | None: + def get_radar_forecast(self) -> List[IrmKmiRadarForecast]: """Create a list of short term forecasts for rain based on the data provided by the rain radar""" data = self._api_data.get('animation', {}) - if data is None: - return None + if not isinstance(data, dict): + return [] sequence = data.get("sequence", []) unit = data.get("unit", {}).get("en", None) ratios = [f['value'] / f['position'] for f in sequence if f['position'] > 0] @@ -418,8 +418,12 @@ class IrmKmiApiClientHa(IrmKmiApiClient): ) return forecast - async def get_animation_data(self, tz: ZoneInfo, lang: str, style: str, dark_mode: bool) -> (RadarAnimationData, - str, Tuple[int, int]): + def get_animation_data(self, + tz: ZoneInfo, + lang: str, + style: str, + dark_mode: bool + ) -> RadarAnimationData: """From the API data passed in, call the API to get all the images and create the radar animation data object. Frames from the API are merged with the background map and the location marker to create each frame.""" animation_data = self._api_data.get('animation', {}).get('sequence') @@ -441,16 +445,29 @@ class IrmKmiApiClientHa(IrmKmiApiClient): location=localisation ) - r = self._get_rain_graph_data( - radar_animation, - animation_data, - country, - images_from_api, - tz, - style, - dark_mode) + sequence: List[AnimationFrameData] = list() - return r + current_time = datetime.now(tz) + most_recent_frame = None + + for idx, item in enumerate(animation_data): + frame = AnimationFrameData( + image=images_from_api[idx], + time=datetime.fromisoformat(item.get('time')) if item.get('time', None) is not None else None, + value=item.get('value', 0), + position=item.get('position', 0), + position_lower=item.get('positionLower', 0), + position_higher=item.get('positionHigher', 0) + ) + sequence.append(frame) + + if most_recent_frame is None and current_time < frame['time']: + most_recent_frame = idx - 1 if idx > 0 else idx + + radar_animation['sequence'] = sequence + radar_animation['most_recent_image_idx'] = most_recent_frame + + return radar_animation def get_warnings(self, lang: str) -> List[WarningData]: """Create a list of warning data instances based on the api data""" @@ -518,46 +535,3 @@ class IrmKmiApiClientHa(IrmKmiApiClient): new_url = parsed_url._replace(query=new_query) return str(urllib.parse.urlunparse(new_url)) - @staticmethod - def _get_rain_graph_data(radar_animation: RadarAnimationData, - api_animation_data: List[dict], - country: str | None, - images_from_api: list[str], - tz: ZoneInfo, - style: str, - dark_mode: bool - ) -> (RadarAnimationData, str, Tuple[int, int]): - """Create a RainGraph object that is ready to output animated and still SVG images""" - sequence: List[AnimationFrameData] = list() - - current_time = datetime.now(tz) - most_recent_frame = None - - for idx, item in enumerate(api_animation_data): - frame = AnimationFrameData( - image=images_from_api[idx], - time=datetime.fromisoformat(item.get('time')) if item.get('time', None) is not None else None, - value=item.get('value', 0), - position=item.get('position', 0), - position_lower=item.get('positionLower', 0), - position_higher=item.get('positionHigher', 0) - ) - sequence.append(frame) - - if most_recent_frame is None and current_time < frame['time']: - most_recent_frame = idx - 1 if idx > 0 else idx - - radar_animation['sequence'] = sequence - radar_animation['most_recent_image_idx'] = most_recent_frame - - satellite_mode = style == OPTION_STYLE_SATELLITE - - if country == 'NL': - image_path = "custom_components/irm_kmi/resources/nl.png" - bg_size = (640, 600) - else: - image_path = (f"custom_components/irm_kmi/resources/be_" - f"{'satellite' if satellite_mode else 'black' if dark_mode else 'white'}.png") - bg_size = (640, 490) - - return radar_animation, image_path, bg_size diff --git a/irm_kmi_api/rain_graph.py b/irm_kmi_api/rain_graph.py index a83bab8..7d509ba 100644 --- a/irm_kmi_api/rain_graph.py +++ b/irm_kmi_api/rain_graph.py @@ -12,6 +12,7 @@ from svgwrite.animate import Animate from svgwrite.container import FONT_TEMPLATE from .api import IrmKmiApiClient +from .const import OPTION_STYLE_SATELLITE from .data import AnimationFrameData, RadarAnimationData from .resources import be_black, be_satellite, be_white, nl, roboto @@ -21,8 +22,8 @@ _LOGGER = logging.getLogger(__name__) class RainGraph: def __init__(self, animation_data: RadarAnimationData, - background_image_path: str, - background_size: (int, int), + country: str, + style: str, dark_mode: bool = False, tz: datetime.tzinfo = None, svg_width: float = 640, @@ -36,17 +37,23 @@ class RainGraph: ): self._animation_data: RadarAnimationData = animation_data - self._background_image_path: str = background_image_path - self._background_size: (int, int) = background_size + self._country: str = country + + if self._country == 'NL': + self._background_size: (int, int) = (640, 600) + else: + self._background_size: (int, int) = (640, 490) + + self._style = style self._dark_mode: bool = dark_mode self._tz = tz self._svg_width: float = svg_width self._inset: float = inset self._graph_height: float = graph_height - self._top_text_space: float = top_text_space + background_size[1] - self._top_text_y_pos: float = top_text_y_pos + background_size[1] + self._top_text_space: float = top_text_space + self._background_size[1] + self._top_text_y_pos: float = top_text_y_pos + self._background_size[1] self._bottom_text_space: float = bottom_text_space - self._bottom_text_y_pos: float = bottom_text_y_pos + background_size[1] + self._bottom_text_y_pos: float = bottom_text_y_pos + self._background_size[1] self._api_client = api_client self._frame_count: int = len(self._animation_data['sequence']) @@ -115,6 +122,7 @@ class RainGraph: if idx is not None and type(imgs[idx]) is str: _LOGGER.info("Download single cloud image") + print("Download single cloud image") result = await self.download_images_from_api([imgs[idx]]) self._animation_data['sequence'][idx]['image'] = result[0] @@ -403,13 +411,12 @@ class RainGraph: return copy.deepcopy(self._dwg) def get_background_png_b64(self): - _LOGGER.debug(f"Get b64 for {self._background_image_path}") - if self._background_image_path.endswith('be_black.png'): - return be_black.be_black_b64 - elif self._background_image_path.endswith('be_white.png'): - return be_white.be_white_b64 - elif self._background_image_path.endswith('be_satellite.png'): - return be_satellite.be_satelitte_b64 - elif self._background_image_path.endswith('nl.png'): + _LOGGER.debug(f"Get b64 for {self._country} {self._style} {'dark' if self._dark_mode else 'light'} mode") + if self._country == 'NL': return nl.nl_b64 - return None + elif self._style == OPTION_STYLE_SATELLITE: + return be_satellite.be_satelitte_b64 + elif self._dark_mode: + return be_black.be_black_b64 + else: + return be_white.be_white_b64 diff --git a/tests/fixtures/forecast.json b/tests/fixtures/forecast.json index ba3b66a..ba4817a 100644 --- a/tests/fixtures/forecast.json +++ b/tests/fixtures/forecast.json @@ -1401,7 +1401,7 @@ "time": "2023-12-26T17:30:00+01:00", "uri": "https:\/\/app.meteo.be\/services\/appv4\/?s=getIncaImage&i=202312261640&f=2&k=4a71be18d6cb09f98c49c53f59902f8c&d=202312261720", "value": 0, - "position": 0, + "position": 8, "positionLower": 0, "positionHigher": 0 }, @@ -1409,7 +1409,7 @@ "time": "2023-12-26T17:40:00+01:00", "uri": "https:\/\/app.meteo.be\/services\/appv4\/?s=getIncaImage&i=202312261650&f=2&k=4a71be18d6cb09f98c49c53f59902f8c&d=202312261720", "value": 0.1, - "position": 0, + "position": 4, "positionLower": 0, "positionHigher": 0 }, @@ -1417,7 +1417,7 @@ "time": "2023-12-26T17:50:00+01:00", "uri": "https:\/\/app.meteo.be\/services\/appv4\/?s=getIncaImage&i=202312261700&f=2&k=4a71be18d6cb09f98c49c53f59902f8c&d=202312261720", "value": 0.01, - "position": 0, + "position": 12, "positionLower": 0, "positionHigher": 0 }, diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..d9b0a7e --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,188 @@ +import json +import time +from datetime import datetime as dt, timedelta +from unittest.mock import AsyncMock, MagicMock +from zoneinfo import ZoneInfo + +import freezegun +import pytest + +from irm_kmi_api.api import IrmKmiApiClient, IrmKmiApiClientHa +from irm_kmi_api.data import CurrentWeatherData +from irm_kmi_api.pollen import PollenParser + + +@freezegun.freeze_time(dt.fromisoformat('2025-05-03T17:30:00+00:00')) +async def test_get_forecast_coord_api_called() -> None: + api = IrmKmiApiClient(session=MagicMock(), user_agent="test-user-agent") + mocked_response = b""" + { + "cityName": "Floreffe", + "country": "BE", + "obs": { + "temp": -2, + "timestamp": "2024-01-12T10:10:00+01:00", + "ww": 27, + "dayNight": "d" + }, + "other": "things" + } + """ + + api._api_wrapper = AsyncMock(return_value=mocked_response) + + result = await api.get_forecasts_coord({'lat': 12.123456789123456, 'long': 42}) + + assert result == json.loads(mocked_response) + params = {'s': 'getForecasts', 'k': '17ab82306446289383960bb13b3dcee4', 'lat': 12.123457, 'long': 42} + api._api_wrapper.assert_called_once_with(params=params) + api._api_wrapper.assert_awaited_once_with(params=params) + + +async def test_get_svg_api_called() -> None: + api = IrmKmiApiClient(session=MagicMock(), user_agent="test-user-agent") + mocked_response = b"""HEY None: + api = IrmKmiApiClient(session=MagicMock(), user_agent="test-user-agent") + mocked_response = b"""//PNG-data-here""" + + api._api_wrapper = AsyncMock(return_value=mocked_response) + + url = "my://best-url" + result = await api.get_image(url=url) + + assert result == mocked_response + api._api_wrapper.assert_called_once_with(params={}, base_url=url) + api._api_wrapper.assert_awaited_once_with(params={}, base_url=url) + + +def test_expire_cache_clears_items() -> None: + api = IrmKmiApiClient(session=MagicMock(), user_agent="test-user-agent") + assert api.cache_max_age == 60 * 60 * 2 + + api.cache = { + 'first-url': { + 'timestamp': time.time() - timedelta(hours=3).seconds, + 'response': 'wowo', + 'etag': 'etag-1' + }, + 'second-url': { + 'timestamp': time.time() - timedelta(hours=1).seconds, + 'response': 'wowo', + 'etag': 'etag-2' + } + } + + assert len(api.cache) == 2 + + api.expire_cache() + + assert len(api.cache) == 1 + assert 'second-url' in api.cache + + +async def test_api_wrapper_puts_response_in_cache() -> None: + response = MagicMock() + response.raise_for_status = MagicMock() + response.status = 200 + response.read = AsyncMock(return_value=b"response value") + response.headers = {'ETag': 'test-etag'} + + session = MagicMock() + session.request = AsyncMock(return_value=response) + + api = IrmKmiApiClient(session=session, user_agent="test-user-agent") + + r = await api._api_wrapper(params={}, base_url='test-url') + + assert r == b"response value" + assert len(api.cache) == 1 + assert 'test-url' in api.cache + + session.request.assert_awaited_once_with( + method='get', url='test-url', headers={'User-Agent': 'test-user-agent'}, json=None, params={} + ) + + +async def test_api_wrapper_gets_response_from_cache() -> None: + response = MagicMock() + response.raise_for_status = MagicMock() + response.status = 304 + response.read = AsyncMock(side_effect=AssertionError("Should not read the response")) + response.headers = {'ETag': 'test-etag'} + + session = MagicMock() + session.request = AsyncMock(return_value=response) + + api = IrmKmiApiClient(session=session, user_agent="test-user-agent") + api.cache = { + 'test-url': { + 'timestamp': time.time(), + 'response': b"response value", + 'etag': 'test-etag' + } + } + + r = await api._api_wrapper(params={}, base_url='test-url') + + assert r == b"response value" + assert len(api.cache) == 1 + assert 'test-url' in api.cache + + session.request.assert_awaited_once_with( + method='get', + url='test-url', + headers={'User-Agent': 'test-user-agent', 'If-None-Match': 'test-etag'}, + json=None, + params={} + ) + + +async def test_default_value_when_empty_data() -> None: + api = IrmKmiApiClientHa(session=MagicMock(), user_agent='hey', cdt_map={}) + tz = ZoneInfo('Europe/Brussels') + lang = 'en' + + assert api.get_city() is None + + assert api.get_country() is None + + assert api.get_current_weather(tz) == CurrentWeatherData( + condition=None, + temperature=None, + wind_speed=None, + wind_gust_speed=None, + wind_bearing=None, + pressure=None, + uv_index=None + ) + + assert api._get_uv_index() is None + + assert api._get_now_hourly(tz) is None + + assert api.get_daily_forecast(tz, lang) == [] + + assert api.get_hourly_forecast(tz) == [] + + assert api.get_radar_forecast() == [] + + with pytest.raises(ValueError): + api.get_animation_data(tz, lang, 'style', True) + + assert api.get_warnings(lang) == [] + + assert await api.get_pollen() == PollenParser.get_default_data() + + diff --git a/tests/test_current_weather.py b/tests/test_current_weather.py index 6a9c0e5..adf6ef0 100644 --- a/tests/test_current_weather.py +++ b/tests/test_current_weather.py @@ -10,10 +10,10 @@ from tests.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_PARTLYCLOUDY @freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00')) -async def test_current_weather_be() -> None: +def test_current_weather_be() -> None: api = get_api_with_data("forecast.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_current_weather(tz) + result = api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_CLOUDY, @@ -29,10 +29,10 @@ async def test_current_weather_be() -> None: @freeze_time(datetime.fromisoformat("2023-12-28T15:30:00")) -async def test_current_weather_nl() -> None: +def test_current_weather_nl() -> None: api = get_api_with_data("forecast_nl.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_current_weather(tz) + result = api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_CLOUDY, @@ -48,11 +48,11 @@ async def test_current_weather_nl() -> None: @freeze_time("2024-06-09T13:40:00+00:00") -async def test_current_condition_forecast_nl() -> None: +def test_current_condition_forecast_nl() -> None: api = get_api_with_data("forecast_ams_no_ww.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_current_weather(tz) + result = api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_PARTLYCLOUDY, @@ -122,7 +122,7 @@ async def test_current_condition_forecast_nl() -> None: ('pressure', 1010, 'midnight-bug-31-05-2024T00-13.json'), ('pressure', 1010, 'no-midnight-bug-31-05-2024T01-55.json') ]) -async def test_current_weather_attributes( +def test_current_weather_attributes( sensor: str, expected: int | float | None, filename: str @@ -133,10 +133,10 @@ async def test_current_weather_attributes( tz = ZoneInfo("Europe/Brussels") @freeze_time(datetime.fromisoformat(time) + timedelta(seconds=45, minutes=1)) - async def run(sensor_: str, expected_: int | float | None): + def run(sensor_: str, expected_: int | float | None): - current_weather = await api.get_current_weather(tz) + current_weather = api.get_current_weather(tz) r = current_weather.get(sensor_, None) assert r == expected_ - await run(sensor, expected) + run(sensor, expected) diff --git a/tests/test_daily_forecast.py b/tests/test_daily_forecast.py index ae6716a..15f973e 100644 --- a/tests/test_daily_forecast.py +++ b/tests/test_daily_forecast.py @@ -13,7 +13,7 @@ async def test_daily_forecast() -> None: api = get_api_with_data("forecast.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_daily_forecast(tz, 'fr') + result = api.get_daily_forecast(tz, 'fr') assert isinstance(result, list) assert len(result) == 8 @@ -43,7 +43,7 @@ async def test_daily_forecast_midnight_bug() -> None: api = get_api_with_data("midnight-bug-31-05-2024T00-13.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_daily_forecast(tz, 'en') + result = api.get_daily_forecast(tz, 'en') assert result[0]['datetime'] == '2024-05-31' assert not result[0]['is_daytime'] @@ -63,7 +63,7 @@ async def test_datetime_daily_forecast_nl() -> None: api = get_api_with_data("forecast_ams_no_ww.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_daily_forecast(tz, 'en') + result = api.get_daily_forecast(tz, 'en') assert result[0]['datetime'] == '2024-06-09' assert result[0]['is_daytime'] @@ -80,7 +80,7 @@ async def test_sunrise_sunset_nl() -> None: api = get_api_with_data("forecast_ams_no_ww.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_daily_forecast(tz, 'en') + result = api.get_daily_forecast(tz, 'en') assert result[0]['sunrise'] == '2024-06-09T05:19:28+02:00' assert result[0]['sunset'] == '2024-06-09T22:01:09+02:00' @@ -97,7 +97,7 @@ async def test_sunrise_sunset_be() -> None: api = get_api_with_data("forecast.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_daily_forecast(tz, 'en') + result = api.get_daily_forecast(tz, 'en') assert result[1]['sunrise'] == '2023-12-27T08:44:00+01:00' assert result[1]['sunset'] == '2023-12-27T16:43:00+01:00' diff --git a/tests/test_hourly_forecast.py b/tests/test_hourly_forecast.py index 4f14629..ccee648 100644 --- a/tests/test_hourly_forecast.py +++ b/tests/test_hourly_forecast.py @@ -9,10 +9,10 @@ from tests.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_RAINY @freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00')) -async def test_hourly_forecast() -> None: +def test_hourly_forecast() -> None: api = get_api_with_data("forecast.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_hourly_forecast(tz) + result = api.get_hourly_forecast(tz) assert isinstance(result, list) assert len(result) == 49 @@ -35,11 +35,11 @@ async def test_hourly_forecast() -> None: @freeze_time(datetime.fromisoformat('2024-05-31T01:50:00+02:00')) -async def test_hourly_forecast_bis() -> None: +def test_hourly_forecast_bis() -> None: api = get_api_with_data("no-midnight-bug-31-05-2024T01-55.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_hourly_forecast(tz) + result = api.get_hourly_forecast(tz) assert isinstance(result, list) @@ -53,12 +53,12 @@ async def test_hourly_forecast_bis() -> None: @freeze_time(datetime.fromisoformat('2024-05-31T00:10:00+02:00')) -async def test_hourly_forecast_midnight_bug() -> None: +def test_hourly_forecast_midnight_bug() -> None: # Related to https://github.com/jdejaegh/irm-kmi-ha/issues/38 api = get_api_with_data("midnight-bug-31-05-2024T00-13.json") tz = ZoneInfo("Europe/Brussels") - result = await api.get_hourly_forecast(tz) + result = api.get_hourly_forecast(tz) assert isinstance(result, list) diff --git a/tests/test_rain_graph.py b/tests/test_rain_graph.py index 8eb8195..626faa7 100644 --- a/tests/test_rain_graph.py +++ b/tests/test_rain_graph.py @@ -1,10 +1,16 @@ import base64 import datetime +import json from datetime import datetime as dt from datetime import timedelta +from unittest.mock import MagicMock, AsyncMock +from zoneinfo import ZoneInfo +from irm_kmi_api.api import IrmKmiApiClientHa +from irm_kmi_api.const import OPTION_STYLE_SATELLITE from irm_kmi_api.data import AnimationFrameData, RadarAnimationData from irm_kmi_api.rain_graph import RainGraph +from tests.conftest import load_fixture def get_radar_animation_data() -> RadarAnimationData: @@ -38,8 +44,8 @@ async def test_svg_frame_setup(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) await rain_graph.draw_svg_frame() @@ -58,8 +64,8 @@ def test_svg_hint(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.write_hint() @@ -74,8 +80,8 @@ def test_svg_time_bars(): rain_graph = RainGraph( tz = datetime.UTC, animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_hour_bars() @@ -93,8 +99,8 @@ def test_draw_chances_path(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_chances_path() @@ -111,8 +117,8 @@ def test_draw_data_line(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_data_line() @@ -129,8 +135,8 @@ async def test_insert_background(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) await rain_graph.insert_background() @@ -152,8 +158,8 @@ def test_draw_current_frame_line_moving(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_current_fame_line() @@ -180,8 +186,8 @@ def test_draw_current_frame_line_index(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_current_fame_line(0) @@ -209,8 +215,8 @@ def test_draw_description_text(): rain_graph = RainGraph( tz=datetime.UTC, animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.draw_description_text() @@ -236,8 +242,8 @@ def test_draw_cloud_layer(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) rain_graph.insert_cloud_layer() @@ -256,8 +262,8 @@ async def test_draw_location_layer(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="irm_kmi_api/resources/be_white.png", - background_size=(640, 490), + country='BE', + style='STD', ) await rain_graph.draw_location() @@ -268,3 +274,64 @@ async def test_draw_location_layer(): png_b64 = base64.b64encode(file.read()).decode('utf-8') assert png_b64 in str_svg + + +def test_get_animation_data(): + api = IrmKmiApiClientHa(session=MagicMock(), user_agent='testing', cdt_map={}) + + tz = ZoneInfo('Europe/Brussels') + lang = 'en' + style = OPTION_STYLE_SATELLITE + dark_mode = False + + api._api_data = json.loads(load_fixture("forecast.json")) + + data = api.get_animation_data(tz, lang, style, dark_mode) + print(data) + + assert list(map(lambda x: x.get('value'), data['sequence'])) == [0, 0, 0, 0, 0.1, 0.01, 0.12, 1.2, 2, 0, 0] + assert list(map(lambda x: x.get('position'), data['sequence'])) == [0, 0, 0, 8, 4, 12, 0, 0, 0, 0, 0] + assert list(map(lambda x: x.get('position_lower'), data['sequence'])) == [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] + assert list(map(lambda x: x.get('position_higher'), data['sequence'])) == [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] + + for element in data['sequence']: + assert 'rs=4' in element['image'] + + +async def test_download_single_cloud(): + data = get_radar_animation_data() + for i, item in enumerate(data['sequence']): + item['image'] = f'image-url-{i}' + + rain_graph = RainGraph( + animation_data=data, + country='BE', + style='STD', + ) + + rain_graph._api_client = MagicMock() + rain_graph._api_client.get_image = AsyncMock() + + await rain_graph.download_clouds(2) + + rain_graph._api_client.get_image.assert_called_once_with('image-url-2') + +async def test_download_many_clouds(): + data = get_radar_animation_data() + for i, item in enumerate(data['sequence']): + item['image'] = f'image-url-{i}' + + rain_graph = RainGraph( + animation_data=data, + country='BE', + style='STD', + ) + + rain_graph._api_client = MagicMock() + rain_graph._api_client.get_image = AsyncMock() + + await rain_graph.download_clouds() + + for i in range(10): + rain_graph._api_client.get_image.assert_any_call(f'image-url-{i}') +