Initial test implementation and update timeout api usage

Author: lp55

pyproject.toml

@@ -43,6 +43,7 @@ pycodestyle = "^2.6.0"
 pylint = "^2.10.2"
 pytest = "^6.2.1"
 pytest-cov = "^2.12.1"
+pytest-asyncio = "^0.25.3"
 
 [tool.poetry.scripts]
 surepy = 'surepy.surecli:cli'

surepy/client.py

@@ -67,9 +67,7 @@ def token_seems_valid(token: str) -> bool:
     Returns:
         bool: True if ``token`` seems valid
     """
-    return (
-        (token is not None) and token.isascii() and token.isprintable() and (320 < len(token))
-    )
+    return (token is not None) and token.isascii() and token.isprintable() and (320 < len(token))
 
 
 def find_token() -> str | None:
@@ -228,76 +226,79 @@ async def call(
 
         response_data = None
 
-        session = self._session if self._session else aiohttp.ClientSession()
-
         try:
-            with async_timeout.timeout(self._api_timeout):
-                headers = self._generate_headers()
+            session = (
+                self._session
+                if self._session
+                else aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self._api_timeout))
+            )
 
-                # use etag if available
-                if resource in self._etags:
-                    headers[ETAG] = str(self._etags.get(resource))
-                    # logger.debug("🐾 \x1b[38;2;255;26;102m·\x1b[0m etag: %s", headers[ETAG])
+            headers = self._generate_headers()
 
-                await session.options(resource, headers=headers)
-                response: aiohttp.ClientResponse = await session.request(
-                    method, resource, headers=headers, json=data
-                )
+            # use etag if available
+            if resource in self._etags:
+                headers[ETAG] = str(self._etags.get(resource))
+                # logger.debug("🐾 \x1b[38;2;255;26;102m·\x1b[0m etag: %s", headers[ETAG])
+
+            await session.options(resource, headers=headers)
+            response: aiohttp.ClientResponse = await session.request(
+                method, resource, headers=headers, json=data, timeout=self._api_timeout
+            )
 
-                if response.status == HTTPStatus.OK or response.status == HTTPStatus.CREATED:
-                    self.resources[resource] = response_data = await response.json()
-
-                    if ETAG in response.headers:
-                        self._etags[resource] = response.headers[ETAG].strip('"')
-
-                elif response.status == HTTPStatus.NOT_MODIFIED:
-                    # Etag header matched, no new data available
-                    logger.debug(
-                        "🐾 \x1b[38;2;0;255;0m·\x1b[0m %d: etag matched - no new data available",
-                        response.status,
-                    )
-
-                elif response.status == HTTPStatus.UNAUTHORIZED:
-                    logger.error(
-                        "🐾 \x1b[38;2;255;26;102m·\x1b[0m %s %s: %d | %s",
-                        method,
-                        resource.replace("https://", ""),
-                        response.status,
-                        response,
-                    )
-                    self._auth_token = None
-                    if not second_try:
-                        token_refreshed = await self.get_token()
-                        if token_refreshed:
-                            await self.call(method="GET", resource=resource, second_try=True)
-
-                    raise SurePetcareAuthenticationError()
-
-                else:
-                    logger.info(
-                        "🐾 \x1b[38;2;255;0;255m·\x1b[0m %s %s: %d | %s",
-                        method,
-                        resource.replace("https://", ""),
-                        response.status,
-                        response,
-                    )
-
-                if response_data:
-                    responselen = len(response_data.get("data", []))
-                else:
-                    responselen = 0
+            if response.status == HTTPStatus.OK or response.status == HTTPStatus.CREATED:
+                self.resources[resource] = response_data = await response.json()
+
+                if ETAG in response.headers:
+                    self._etags[resource] = response.headers[ETAG].strip('"')
+
+            elif response.status == HTTPStatus.NOT_MODIFIED:
+                # Etag header matched, no new data available
                 logger.debug(
-                    "🐾 \x1b[38;2;0;255;0m·\x1b[0m %s %s | %d",
+                    "🐾 \x1b[38;2;0;255;0m·\x1b[0m %d: etag matched - no new data available",
+                    response.status,
+                )
+
+            elif response.status == HTTPStatus.UNAUTHORIZED:
+                logger.error(
+                    "🐾 \x1b[38;2;255;26;102m·\x1b[0m %s %s: %d | %s",
                     method,
                     resource.replace("https://", ""),
-                    responselen,
+                    response.status,
+                    response,
                 )
+                self._auth_token = None
+                if not second_try:
+                    token_refreshed = await self.get_token()
+                    if token_refreshed:
+                        await self.call(method="GET", resource=resource, second_try=True)
+
+                raise SurePetcareAuthenticationError()
+
+            else:
+                logger.info(
+                    "🐾 \x1b[38;2;255;0;255m·\x1b[0m %s %s: %d | %s",
+                    method,
+                    resource.replace("https://", ""),
+                    response.status,
+                    response,
+                )
+
+            if response_data:
+                responselen = len(response_data.get("data", []))
+            else:
+                responselen = 0
+            logger.debug(
+                "🐾 \x1b[38;2;0;255;0m·\x1b[0m %s %s | %d",
+                method,
+                resource.replace("https://", ""),
+                responselen,
+            )
 
-                if method == "DELETE" and response.status == HTTPStatus.NO_CONTENT:
-                    # TODO: this does not return any data, is there a better way?
-                    return "DELETE 204 No Content"
+            if method == "DELETE" and response.status == HTTPStatus.NO_CONTENT:
+                # TODO: this does not return any data, is there a better way?
+                return "DELETE 204 No Content"
 
-                return response_data
+            return response_data
 
         except (asyncio.TimeoutError, aiohttp.ClientError) as error:
             logger.error("Can not load data from %s", resource)

tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for SurePy."""

tests/test_client.py

@@ -0,0 +1,76 @@
+from pathlib import Path
+import asyncio
+import pytest
+import async_timeout
+import pprint
+from aiohttp import ClientSession, TCPConnector
+from shutil import copyfile
+from datetime import datetime, timedelta
+import configparser
+
+from surepy import Surepy
+from surepy.client import SureAPIClient
+from surepy.entities import SurepyEntity
+from surepy.entities.devices import Feeder, Felaqua, Flap, Hub, SurepyDevice
+from surepy.entities.pet import Pet
+from surepy.enums import EntityType
+
+token_file = Path("~/.surepy.token").expanduser()
+old_token_file = token_file.with_suffix(".old_token")
+auth_file = Path("~/.surepy.auth").expanduser()
+config = configparser.ConfigParser()
+config.read(str(auth_file))
+
+
+def file_older_then(file: Path, delta: timedelta) -> bool:
+    return datetime.fromtimestamp(file.stat().st_mtime) < (datetime.now() - delta)
+
+
+async def get_surepy() -> Surepy | None:
+    if not token_file.exists() or file_older_then(token_file, timedelta(minutes=5)):
+
+        spy = Surepy(
+            email=config.get("Login", "email"),
+            password=config.get("Login", "password"),
+        )
+
+        if surepy_token := await spy.sac.get_token():
+
+            if token_file.exists() and surepy_token != token_file.read_text(encoding="utf-8"):
+                copyfile(token_file, old_token_file)
+
+            token_file.write_text(surepy_token, encoding="utf-8")
+
+        return spy
+    else:
+        return Surepy(auth_token=token_file.read_text(encoding="utf-8"))
+
+
+@pytest.mark.asyncio
+async def test_get_entities() -> None:
+    spy = await get_surepy()
+
+    response = await spy.get_entities(refresh=True)
+
+    with open("response.txt", "w") as file:
+        file.write(pprint.pformat(response))
+
+    assert len(response) > 0
+
+
+@pytest.mark.asyncio
+async def test_get_actions() -> None:
+    spy = await get_surepy()
+
+    response = await spy.get_actions(household_id=config.get("IDs", "household"))
+
+    assert len(response) > 0
+
+
+@pytest.mark.asyncio
+async def test_get_latest_anonymous_drinks() -> None:
+    spy = await get_surepy()
+
+    response = await spy.get_latest_anonymous_drinks(household_id=config.get("IDs", "household"))
+
+    assert len(response) > 0