jdejaegh/irm-kmi-api
1
0
Fork 0
mirror of https://github.com/jdejaegh/irm-kmi-api.git synced 2025-06-26 20:05:40 +02:00
Code Issues Projects Releases Packages Wiki Activity

Improve tests

Browse source
This commit is contained in:
Jules 2025-05-03 15:46:47 +02:00
parent f1d94d6590
commit 7166c0ccc8
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
8 changed files with 365 additions and 129 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

108
irm_kmi_api/api.py
View file

@ -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

39
irm_kmi_api/rain_graph.py
View file

@ -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

6
tests/fixtures/forecast.json vendored
View file

@ -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
},

188
tests/test_api.py Normal file
View file

@ -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"""<svg>HEY</svg"""
api._api_wrapper = AsyncMock(return_value=mocked_response)
url = "my://best-url"
result = await api.get_svg(url=url)
assert result == mocked_response.decode()
api._api_wrapper.assert_called_once_with(params={}, base_url=url)
api._api_wrapper.assert_awaited_once_with(params={}, base_url=url)
async def test_get_image_api_called() -> 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()

20
tests/test_current_weather.py
View file

@ -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)

10
tests/test_daily_forecast.py
View file

@ -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'

12
tests/test_hourly_forecast.py
View file

@ -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)

111
tests/test_rain_graph.py
View file

@ -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}')