test: improve coverage

Author: gmasse

pyproject.toml

@@ -41,7 +41,8 @@ dev = [
     "mkdocs == 1.6.1",
     "mkdocs-material == 9.5.50",
     "mkdocstrings == 0.28.1",
-    "mkdocstrings-python == 1.16.1"
+    "mkdocstrings-python == 1.16.1",
+    "python-dateutil == 2.9.0.post0"
 ]
 
 [tool.pytest.ini_options]

src/ecactus/async_client.py

@@ -66,6 +66,9 @@ async def login(
         except ApiResponseError as err:
             if err.code == 20414:
                 raise AuthenticationError from err
+            if err.code == 20000:
+                raise AuthenticationError("Missing Account or Password") from err
+            raise
         self.access_token = data["accessToken"]
         self.refresh_token = data["refreshToken"]
 

src/ecactus/client.py

@@ -66,6 +66,9 @@ def login(
         except ApiResponseError as err:
             if err.code == 20414:
                 raise AuthenticationError from err
+            if err.code == 20000:
+                raise AuthenticationError("Missing Account or Password") from err
+            raise
         self.access_token = data["accessToken"]
         self.refresh_token = data["refreshToken"]
 

src/ecactus/exceptions.py

@@ -92,9 +92,9 @@ class ParameterVerificationFailedError(EcosApiError):
     def __init__(self, message: str | None = None) -> None:
         """Initialize the exception with a default error message."""
         if message is None:
-            super().__init__("Parameter Verification Failed")
+            super().__init__("Parameter verification failed")
         else:
-            super().__init__(f"Parameter Verification Failed: {message}")
+            super().__init__(f"Parameter verification failed: {message}")
 
 
 class InvalidJsonError(EcosApiError):

tests/mock_server.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 from aiohttp import web
+from dateutil.relativedelta import relativedelta
 from multidict import CIMultiDictProxy
 
 # Configure logging
@@ -108,7 +109,7 @@ def _not_implemented_response() -> web.Response:
     async def handle_login(self, request: web.Request) -> web.Response:
         """Mock login endpoint."""
         output: JSON = {}
-        data: dict = await request.json()  # Parse the JSON payload
+        data = await request.json()  # Parse the JSON payload
         if (
             not data.get("clientVersion")
             or data.get("clientType") != "BROWSER"
@@ -423,45 +424,84 @@ async def handle_get_insight(self, request: web.Request) -> web.Response:
             return EcosMockServer._ok_response(
                 code=20404, message="Parameter verification failed", success=False
             )
-        if request_payload.get("periodType") != 0:
-            return EcosMockServer._not_implemented_response()
-        start_date = datetime.fromtimestamp(
-            int(request_payload.get("timestamp") / 1000)
-        ).replace(
-            hour=0, minute=0, second=0, microsecond=0
-        )  # convert timestamp (in milliseconds) to datetime
-        end_date = start_date + timedelta(days=1)
-        fake_data: dict[str, float] = await self._generate_metrics_data(
-            start_date, end_date
-        )
-        latest_timestamp = max(fake_data.keys())
-        fake_data[str(int(latest_timestamp) - 1)] = fake_data.pop(
-            latest_timestamp
-        )  # rename the latest timestamp by (timestamp - 1 sec)
-
-        return EcosMockServer._success_response(
-            data={
-                "selfPowered": 31.0,
-                "deviceRealtimeDto": {
-                    "solarPowerDps": fake_data,
-                    "batteryPowerDps": fake_data,
-                    "gridPowerDps": fake_data,
-                    "meterPowerDps": fake_data,
-                    "homePowerDps": fake_data,
-                    "epsPowerDps": fake_data,
-                },
-                "deviceStatisticsDto": {
-                    "consumptionEnergy": 42.5,
-                    "fromBattery": 0.0,
-                    "toBattery": 0.0,
-                    "fromGrid": 29.2,
-                    "toGrid": 6.3,
-                    "fromSolar": 19.6,
-                    "eps": 0.0,
-                },
-                "insightConsumptionDataDto": None,
-            }
-        )
+        if request_payload.get("periodType") == 0: # 5-minute power measurement for the provided day
+            start_date = datetime.fromtimestamp(
+                int(request_payload.get("timestamp", int(datetime.now().timestamp()*1000)) / 1000)
+            ).replace(
+                hour=0, minute=0, second=0, microsecond=0
+            )  # convert timestamp (in milliseconds) to datetime
+            end_date = start_date + timedelta(days=1)
+            fake_data: dict[str, float] = await self._generate_metrics_data(
+                start_date, end_date
+            )
+            latest_timestamp = max(fake_data.keys())
+            fake_data[str(int(latest_timestamp) - 1)] = fake_data.pop(
+                latest_timestamp
+            )  # rename the latest timestamp by (timestamp - 1 sec)
+            return EcosMockServer._success_response(
+                data={
+                    "selfPowered": 31.0,
+                    "deviceRealtimeDto": {
+                        "solarPowerDps": fake_data,
+                        "batteryPowerDps": fake_data,
+                        "gridPowerDps": fake_data,
+                        "meterPowerDps": fake_data,
+                        "homePowerDps": fake_data,
+                        "epsPowerDps": fake_data,
+                    },
+                    "deviceStatisticsDto": {
+                        "consumptionEnergy": 42.5,
+                        "fromBattery": 0.0,
+                        "toBattery": 0.0,
+                        "fromGrid": 29.2,
+                        "toGrid": 6.3,
+                        "fromSolar": 19.6,
+                        "eps": 0.0,
+                    },
+                    "insightConsumptionDataDto": None,
+                }
+            )
+        if request_payload.get("periodType") == 2: # daily energy for the provided month
+            start_date = datetime.fromtimestamp(
+                int(request_payload.get("timestamp", int(datetime.now().timestamp()*1000)) / 1000)
+            ).replace(
+                day=1, hour=0, minute=0, second=0, microsecond=0
+            )  # convert timestamp (in milliseconds) to datetime
+            end_date = start_date + relativedelta(months=1)
+            fake_data = await self._generate_metrics_data(
+                start_date, end_date, interval=timedelta(days=1)
+            )
+            latest_timestamp = max(fake_data.keys())
+            fake_data[str(int(latest_timestamp) - 1)] = fake_data.pop(
+                latest_timestamp
+            )  # rename the latest timestamp by (timestamp - 1 sec)
+            return EcosMockServer._success_response(
+                data={
+                    "selfPowered": 31.0,
+                    "deviceRealtimeDto": None,
+                    "deviceStatisticsDto": {
+                        "consumptionEnergy": 42.5,
+                        "fromBattery": 0.0,
+                        "toBattery": 0.0,
+                        "fromGrid": 29.2,
+                        "toGrid": 6.3,
+                        "fromSolar": 19.6,
+                        "eps": 0.0,
+                    },
+                    "insightConsumptionDataDto": {
+                        "fromBatteryDps": fake_data,
+                        "toBatteryDps": fake_data,
+                        "fromGridDps": fake_data,
+                        "toGridDps": fake_data,
+                        "fromSolarDps": fake_data,
+                        "homeEnergyDps": fake_data,
+                        "epsDps": fake_data,
+                        "selfPoweredDps": fake_data,
+                    },
+                }
+            )
+        # if request_payload.get("periodType") not in (0, 2):
+        return EcosMockServer._not_implemented_response()
 
     async def catch_all(self, request: web.Request) -> web.Response:
         """Catch all endpoint."""

tests/test_async_client.py

@@ -1,6 +1,6 @@
 """Unit tests for asynchronous Ecos class."""
 
-from datetime import datetime
+from datetime import datetime, timedelta
 import logging
 
 import pytest
@@ -9,7 +9,9 @@
 from ecactus.exceptions import (
     AuthenticationError,
     HomeDoesNotExistError,
+    HttpError,
     InitializationError,
+    InvalidJsonError,
     ParameterVerificationFailedError,
     UnauthorizedDeviceError,
     UnauthorizedError,
@@ -34,6 +36,28 @@ def bad_client(mock_server):
     return ecactus.AsyncEcos(url=mock_server.url, access_token="wrong_token")
 
 
+def test_exceptions():
+    """Test exceptions."""
+    exc = InitializationError()
+    assert str(exc) == "Initialization error"
+    exc = AuthenticationError()
+    assert str(exc) == "Account or password or country error"
+    exc = UnauthorizedError()
+    assert str(exc) == "Unauthorized"
+    exc = HomeDoesNotExistError()
+    assert str(exc) == "Home does not exist"
+    exc = HomeDoesNotExistError("home_id")
+    assert str(exc) == "Home does not exist: home_id"
+    exc = UnauthorizedDeviceError()
+    assert str(exc) == "Device is not authorized"
+    exc = ParameterVerificationFailedError()
+    assert str(exc) == "Parameter verification failed"
+    exc = InvalidJsonError()
+    assert str(exc) == "Invalid JSON"
+    exc = HttpError(404, "Not Found")
+    assert str(exc) == "HTTP error: 404 Not Found"
+
+
 def test_client():
     """Test ECOS client."""
     with pytest.raises(InitializationError):
@@ -44,6 +68,17 @@ def test_client():
     assert "weiheng-tech.com" in client.url
 
 
+async def test_ensure_login(mock_server):
+    """Test autologin."""
+    temp_client = ecactus.AsyncEcos(url=mock_server.url)
+    with pytest.raises(AuthenticationError) as excinfo:
+        user = await temp_client.get_user()
+    assert str(excinfo.value) == "Missing Account or Password"
+    temp_client = ecactus.AsyncEcos(email=LOGIN, password=PASSWORD, url=mock_server.url)
+    user = await temp_client.get_user()
+    assert user.username == LOGIN
+
+
 async def test_login(mock_server, client):
     """Test login."""
     with pytest.raises(AuthenticationError):
@@ -95,7 +130,32 @@ async def test_get_today_device_data(client, bad_client):
         await client.get_today_device_data(device_id=0)
     power_ts = await client.get_today_device_data(device_id=1234567890123456789)
     assert len(power_ts.metrics) > 0
-
+    # get the first  timestamp
+    first_timestamp = power_ts.metrics[0].timestamp
+    assert power_ts.find_by_timestamp(first_timestamp, exact=False).solar is not None
+    assert power_ts.find_by_timestamp(first_timestamp, exact=True).solar is not None
+    # get a timestamp that does not exist
+    before_timestamp = first_timestamp - timedelta(seconds=1)
+    assert power_ts.find_by_timestamp(before_timestamp, exact=False).solar == power_ts.metrics[0].solar # the nearst metric is returned
+    assert power_ts.find_by_timestamp(before_timestamp, exact=True) is None # exact lookup returns None
+    # get a timestamp after the last one
+    last_timestamp = power_ts.metrics[-1].timestamp
+    after_timestamp =  last_timestamp + timedelta(seconds=1)
+    assert power_ts.find_by_timestamp(after_timestamp, exact=False).solar == power_ts.metrics[-1].solar # value is 1/10th of the position
+    assert power_ts.find_by_timestamp(after_timestamp, exact=True) is None
+    # get a timestamp within 2 existings
+    if len(power_ts.metrics) > 2:
+        timestamp1 = power_ts.metrics[1].timestamp
+        timestamp2 = power_ts.metrics[2].timestamp
+        delta = timestamp2 - timestamp1
+        if delta >= timedelta(seconds=2):
+            timestamp = timestamp1 + delta/2
+            assert power_ts.find_by_timestamp(timestamp, exact=False).solar == power_ts.metrics[1].solar # returns the value in 2nd position
+            timestamp = timestamp1 + delta/2 + timedelta(seconds=1)
+            assert power_ts.find_by_timestamp(timestamp, exact=False).solar == power_ts.metrics[2].solar # returns the value in 3rd position
+    # return series between 2 dates
+    between_ts = power_ts.find_between(first_timestamp, last_timestamp)
+    assert len(between_ts.metrics) == len(power_ts.metrics)
 
 async def test_get_realtime_device_data(client, bad_client):
     """Test get realtime device data."""
@@ -147,10 +207,19 @@ async def test_get_insight(client, bad_client):
         await client.get_insight(
             device_id=1234567890123456789, start_date=now, period_type=1
         )
+    with pytest.raises(ParameterVerificationFailedError):
+        await client.get_insight(
+            device_id=1234567890123456789, period_type=1
+        )
     insight = await client.get_insight(
         device_id=1234567890123456789, start_date=now, period_type=0
     )
     assert len(insight.power_timeseries.metrics) > 1
+    insight = await client.get_insight(
+        device_id=1234567890123456789, start_date=now, period_type=2
+    )
+    assert len(insight.energy_timeseries.metrics) > 1
+
 
 
 # TODO test 404