Add tests

src/open_irceline/api.py

@@ -47,7 +47,6 @@ async def _api_wrapper(self, url: str, querystring: dict = None, headers: dict =
             headers = dict()
         if 'User-Agent' not in headers:
             headers |= {'User-Agent': _user_agent}
-
         try:
             async with async_timeout.timeout(60):
                 response = await self._session.request(

src/open_irceline/forecast.py

@@ -5,7 +5,7 @@
 
 from aiohttp import ClientResponse, ClientResponseError
 
-from .api import IrcelineBaseClient, _rio_wfs_base_url, _forecast_wms_base_url, IrcelineApiError
+from .api import IrcelineBaseClient, _forecast_wms_base_url, IrcelineApiError
 from .data import ForecastFeature, FeatureValue
 
 
@@ -40,14 +40,13 @@ async def get_data(self,
         for feature, d in product(features, range(4)):
             querystring = base_querystring | {"layers": f"{feature}_d{d}",
                                               "query_layers": f"{feature}_d{d}"}
-            print(querystring)
             try:
-                r: ClientResponse = await self._api_wrapper(_rio_wfs_base_url, querystring)
+                r: ClientResponse = await self._api_wrapper(_forecast_wms_base_url, querystring)
                 r: dict = await r.json()
                 result[(feature, timestamp + timedelta(days=d))] = FeatureValue(
                     value=r.get('features', [{}])[0].get('properties', {}).get('GRAY_INDEX'),
                     timestamp=datetime.fromisoformat(r.get('timeStamp')) if 'timeStamp' in r else None)
-            except (IrcelineApiError, ClientResponseError):
+            except (IrcelineApiError, ClientResponseError, IndexError):
                 result[(feature, timestamp + timedelta(days=d))] = FeatureValue(value=None, timestamp=None)
 
         return result
@@ -72,5 +71,5 @@ def _parse_capabilities(xml_string: str) -> Set[str]:
             return set()
 
         path = './/Capability/Layer/Layer/Name'
-        feature_type_names = {t.text for t in root.findall(path) if t.text.endswith('_d0')}
+        feature_type_names = {t.text for t in root.findall(path)}
         return feature_type_names

tests/conftest.py

@@ -27,28 +27,3 @@ def get_mock_session(json_file=None, text_file=None):
     mock_session.request = AsyncMock(return_value=mock_response)
     return mock_session
 
-
-def create_mock_response(*args, **kwargs):
-    etag = 'my-etag-here'
-    mock_response = Mock()
-    if '20240619' not in kwargs.get('url', ''):
-        mock_response.status = 404
-        mock_response.raise_for_status = Mock(side_effect=aiohttp.ClientResponseError(Mock(), Mock()))
-    elif etag in kwargs.get('headers', {}).get('If-None-Match', ''):
-        mock_response.text = AsyncMock(return_value='')
-        mock_response.status = 304
-    else:
-        mock_response.text = AsyncMock(return_value=get_api_data('forecast.csv', plain=True))
-        mock_response.status = 200
-
-    if '20240619' in kwargs.get('url', ''):
-        mock_response.headers = {'ETag': etag}
-    else:
-        mock_response.headers = dict()
-    return mock_response
-
-
-def get_mock_session_many_csv():
-    mock_session = Mock(aiohttp.ClientSession)
-    mock_session.request = AsyncMock(side_effect=create_mock_response)
-    return mock_session

tests/fixtures/forecast_wms_feature_info.json

@@ -0,0 +1,17 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "id": "",
+      "geometry": null,
+      "properties": {
+        "GRAY_INDEX": 10.853286743164062
+      }
+    }
+  ],
+  "totalFeatures": "unknown",
+  "numberReturned": 1,
+  "timeStamp": "2024-06-30T13:00:21.520Z",
+  "crs": null
+}

tests/fixtures/forecast_wms_feature_info_invalid.json

@@ -0,0 +1,17 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "id": "",
+      "geometry": null,
+      "properties": {
+        "WRONG": 10.853286743164062
+      }
+    }
+  ],
+  "totalFeatures": "unknown",
+  "numberReturned": 1,
+  "timeStamp": "2024-06-30T13:00:21.520Z",
+  "crs": null
+}

tests/fixtures/forecast_wms_feature_info_no_field.json

@@ -0,0 +1,8 @@
+{
+  "type": "FeatureCollection",
+  "features": [],
+  "totalFeatures": "unknown",
+  "numberReturned": 1,
+  "timeStamp": "2024-06-30T13:00:21.520Z",
+  "crs": null
+}

tests/test_api_forecast.py

@@ -0,0 +1,125 @@
+from datetime import datetime
+from itertools import product
+from unittest.mock import call
+
+from freezegun import freeze_time
+
+from src.open_irceline import IrcelineForecastClient, ForecastFeature, FeatureValue
+from src.open_irceline.api import _user_agent, _forecast_wms_base_url
+from tests.conftest import get_api_data, get_mock_session
+
+
+def test_parse_capabilities():
+    data = get_api_data('forecast_wms_capabilities.xml', plain=True)
+    result = IrcelineForecastClient._parse_capabilities(data)
+
+    expected = {'o3_maxhmean_wl_d3', 'pm25_dmean_wl_d0', 'o3_max8hmean_chimv2022_d2', 'no2_maxhmean_tf_d2',
+                'belaqi_forecast_chimv2022_d2', 'pm25_dmean_chimv2022_d3', 'pm10_dmean_chimv2022_d0',
+                'no2_maxhmean_wl_d0', 'no2_maxhmean_d2', 'no2_dmean_chimv2022_d2', 'o3_maxhmean_chimv2022_d3',
+                'pm25_dmean_wl_d3', 'o3_maxhmean_chimv2022_d0', 'pm25_dmean', 'pm25_dmean_tf_d0', 'no2_dmean_wl_d2',
+                'o3_max8hmean_chimv2022_d3', 'pm25_dmean_d2', 'o3_max8hmean_chimv2022_d0', 'o3_maxhmean_wl_d2',
+                'no2_maxhmean_wl_d1', 'pm10_dmean_tf_d2', 'pm25_dmean_d1', 'o3_maxhmean_chimv2022_d2',
+                'pm10_dmean_chimv2022_d2', 'o3_maxhmean_vl', 'belaqi_wl_d2', 'pm10_dmean_wl', 'pm10_dmean_d2',
+                'no2_dmean_wl_d0', 'no2_dmean_d1', 'o3_maxhmean_d2', 'o3_maxhmean_wl', 'pm25_dmean_wl_d2',
+                'o3_maxhmean_d3', 'o3_max8hmean_wl_d3', 'belaqi_d0', 'no2_maxhmean_wl_d2', 'no2_maxhmean_wl',
+                'pm10_dmean_wl_d1', 'no2_dmean_chimv2022_d3', 'o3_maxhmean_tf_d1', 'pm25_dmean_vl', 'pm10_dmean_d0',
+                'o3_max8hmean_d0', 'o3_max8hmean_d2', 'no2_maxhmean_vl', 'o3_max8hmean_chimv2022_d1', 'pm10_dmean',
+                'pm10_dmean_wl_d2', 'euaqi_d3', 'belaqi_d1', 'o3_max8hmean_d1', 'o3_maxhmean_chimv2022_d1', 'belaqi_vl',
+                'belaqi_wl_d0', 'no2_dmean_chimv2022_d0', 'pm25_dmean_wl_d1', 'pm25_dmean_tf_d2', 'no2_dmean_d2',
+                'o3_maxhmean', 'belaqi_wl', 'no2_maxhmean_d0', 'no2_maxhmean_d3', 'o3_max8hmean_d3', 'euaqi_forecast',
+                'o3_max8hmean_wl_d1', 'pm10_dmean_chimv2022_d3', 'no2_maxhmean_wl_d3', 'o3_maxhmean_d1',
+                'no2_dmean_wl_d1', 'o3_maxhmean_wl_d1', 'no2_dmean_d3', 'belaqi_d3', 'belaqi', 'pm25_dmean_d3',
+                'belaqi_forecast', 'no2_dmean_d0', 'pm25_dmean_chimv2022_d1', 'belaqi_wl_d1', 'pm10_dmean_d3',
+                'no2_dmean_wl_d3', 'pm25_dmean_tf_d1', 'euaqi_d0', 'o3_maxhmean_wl_d0', 'belaqi_forecast_chimv2022_d3',
+                'no2_dmean_chimv2022_d1', 'o3_max8hmean_wl_d0', 'o3_max8hmean_wl_d2', 'pm10_dmean_chimv2022_d1',
+                'pm10_dmean_wl_d3', 'pm25_dmean_wl', 'belaqi_forecast_chimv2022_d1', 'euaqi_d2', 'pm10_dmean_d1',
+                'belaqi_wl_d3', 'belaqi_forecast_chimv2022_d0', 'o3_maxhmean_tf_d0', 'euaqi_d1', 'no2_maxhmean',
+                'pm25_dmean_chimv2022_d2', 'belaqi_d2', 'pm25_dmean_d0', 'no2_maxhmean_tf_d0', 'pm10_dmean_tf_d0',
+                'pm25_dmean_chimv2022_d0', 'o3_maxhmean_d0', 'pm10_dmean_tf_d1', 'pm10_dmean_vl', 'no2_maxhmean_tf_d1',
+                'o3_maxhmean_tf_d2', 'pm10_dmean_wl_d0', 'no2_maxhmean_d1'}
+
+    assert result == expected
+
+    for f, d in product(ForecastFeature, range(4)):
+        assert f"{f.split(':')[1]}_d{d}" in result
+
+
+async def test_aget_capabilities():
+    session = get_mock_session(text_file='forecast_wms_capabilities.xml')
+
+    client = IrcelineForecastClient(session)
+    _ = await client.get_capabilities()
+
+    session.request.assert_called_once_with(
+        method='GET',
+        url=_forecast_wms_base_url,
+        params={"service": "WMS",
+                "version": "1.1.1",
+                "request": "GetCapabilities"},
+        headers={'User-Agent': _user_agent}
+    )
+
+
+@freeze_time(datetime.fromisoformat("2024-06-30T13:00:21.520Z"))
+async def test_api_forecast_error():
+    pos = (50.4657, 4.8647)
+    session = get_mock_session('forecast_wms_feature_info_invalid.json')
+
+    client = IrcelineForecastClient(session)
+
+    features = [ForecastFeature.NO2_DMEAN, ForecastFeature.O3_MAXHMEAN]
+    result = await client.get_data(features, pos)
+
+    for k, v in result.items():
+        assert v == FeatureValue(timestamp=datetime.fromisoformat("2024-06-30T13:00:21.520Z"), value=None)
+
+
+async def test_api_forecast_no_field():
+    pos = (50.4657, 4.8647)
+    session = get_mock_session('forecast_wms_feature_info_no_field.json')
+
+    client = IrcelineForecastClient(session)
+
+    features = [ForecastFeature.NO2_DMEAN, ForecastFeature.O3_MAXHMEAN]
+    result = await client.get_data(features, pos)
+
+    for k, v in result.items():
+        assert v == FeatureValue(timestamp=None, value=None)
+
+
+async def test_api_forecast():
+    pos = (50.4657, 4.8647)
+    lat, lon = pos
+    session = get_mock_session('forecast_wms_feature_info.json')
+
+    client = IrcelineForecastClient(session)
+
+    features = [ForecastFeature.NO2_DMEAN, ForecastFeature.O3_MAXHMEAN]
+    _ = await client.get_data(features, pos)
+
+    base = {"service": "WMS",
+            "version": "1.1.1",
+            "request": "GetFeatureInfo",
+            "info_format": "application/json",
+            "width": "1",
+            "height": "1",
+            "srs": "EPSG:4326",
+            "bbox": f"{lon},{lat},{lon + 0.00001},{lat + 0.00001}",
+            "X": "1",
+            "Y": "1"}
+
+    calls = [call(
+        method='GET',
+        url=_forecast_wms_base_url,
+        params=base | {"layers": f"{feature}_d{d}",
+                       "query_layers": f"{feature}_d{d}"},
+        headers={'User-Agent': _user_agent},
+    )
+        for feature, d in product(features, range(4))]
+
+    session.request.assert_has_calls(calls, any_order=True)
+
+
+def test_parse_capabilities_with_error():
+    result = IrcelineForecastClient._parse_capabilities("wow there no valid XML")
+    assert result == set()