Adding Tests

.github/workflows/code_coverage.yml

@@ -0,0 +1,42 @@
+---
+name: Code Coverage
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  test:
+    name: Python ${{ matrix.python }} on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}-latest
+    strategy:
+      matrix:
+        os: [ubuntu]
+        python: [3.7, 3.8, 3.9]
+    steps:
+      - name: Checking out code from GitHub
+        uses: actions/[email protected]
+      - name: Set up Python ${{ matrix.python }}
+        uses: actions/[email protected]
+        with:
+          python-version: ${{ matrix.python }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip setuptools wheel
+          pip install -r requirements.test.txt
+          pip install -r requirements.txt
+          pip list
+      - name: Pytest with coverage reporting
+        run: pytest --cov=sharkiq --cov-report=xml
+      - name: Upload coverage to Codecov
+        if: matrix.python == 3.9 && matrix.os == 'ubuntu'
+        uses: codecov/[email protected]
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          file: ./coverage.xml
+          flags: unittests
+          name: codecov-umbrella

README.md

@@ -1,3 +1,4 @@
+[![codecov](https://codecov.io/gh/JeffResc/sharkiq/branch/main/graph/badge.svg?token=DO96BWVXA7)](https://codecov.io/gh/JeffResc/sharkiq)
 # sharkiq
 Unofficial SDK for Shark IQ robot vacuums, designed primarily to support an integration for [Home Assistant](https://www.home-assistant.io/).
 

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode=strict

requirements.test.txt

@@ -0,0 +1,3 @@
+pytest-asyncio==0.18.3
+pytest==7.1.1
+pytest-cov==3.0.0

sharkiq/ayla_api.py

@@ -48,7 +48,7 @@ def __init__(
         self._app_secret = app_secret
         self.websession = websession
 
-    def ensure_session(self) -> aiohttp.ClientSession:
+    async def ensure_session(self) -> aiohttp.ClientSession:
         """Ensure that we have an aiohttp ClientSession"""
         if self.websession is None:
             self.websession = aiohttp.ClientSession()
@@ -75,7 +75,7 @@ def _set_credentials(self, status_code: int, login_result: Dict):
         self._access_token = login_result["access_token"]
         self._refresh_token = login_result["refresh_token"]
         self._auth_expiration = datetime.now() + timedelta(seconds=login_result["expires_in"])
-        self._is_authed = True
+        self._is_authed = True  # TODO: Any non 200 status code should cause this to be false
 
     def sign_in(self):
         """Authenticate to Ayla API synchronously."""
@@ -91,14 +91,14 @@ def refresh_auth(self):
 
     async def async_sign_in(self):
         """Authenticate to Ayla API synchronously."""
-        session = self.ensure_session()
+        session = await self.ensure_session()
         login_data = self._login_data
         async with session.post(f"{LOGIN_URL:s}/users/sign_in.json", json=login_data) as resp:
             self._set_credentials(resp.status, await resp.json())
 
     async def async_refresh_auth(self):
         """Refresh the authentication synchronously."""
-        session = self.ensure_session()
+        session = await self.ensure_session()
         refresh_data = {"user": {"refresh_token": self._refresh_token}}
         async with session.post(f"{LOGIN_URL:s}/users/refresh_token.json", json=refresh_data) as resp:
             self._set_credentials(resp.status, await resp.json())
@@ -122,7 +122,7 @@ def sign_out(self):
 
     async def async_sign_out(self):
         """Sign out and invalidate the access token"""
-        session = self.ensure_session()
+        session = await self.ensure_session()
         async with session.post(f"{LOGIN_URL:s}/users/sign_out.json", json=self.sign_out_data) as _:
             pass
         self._clear_auth()
@@ -183,7 +183,7 @@ def request(self, method: str, url: str, **kwargs) -> requests.Response:
         return requests.request(method, url, headers=headers, **kwargs)
 
     async def async_request(self, http_method: str, url: str, **kwargs):
-        session = self.ensure_session()
+        session = await self.ensure_session()
         headers = self._get_headers(kwargs)
         return session.request(http_method, url, headers=headers, **kwargs)
 

tests/conftest.py

@@ -0,0 +1,37 @@
+import pytest
+import os
+from sharkiq.ayla_api import get_ayla_api
+from datetime import datetime, timedelta
+
+@pytest.fixture
+def dummy_api():
+    """AylaApi object with invalid auth creds and attributes populated."""
+    username = "[email protected]"
+    password = "mypassword"
+
+    dummy_api = get_ayla_api(username=username, password=password)
+    dummy_api._access_token = "token123"
+    dummy_api._refresh_token = "token321"
+    dummy_api._is_authed = True
+    dummy_api._auth_expiration = datetime.now() + timedelta(seconds=700)
+    return dummy_api
+
+
+@pytest.fixture
+def sample_api():
+    """AylaApi object using user-supplied auth creds via SHARKIQ_USERNAME and 
+    SHARKIQ_PASSWORD environement variables."""
+    username = os.getenv("SHARKIQ_USERNAME")
+    password = os.getenv("SHARKIQ_PASSWORD")
+
+    assert username is not None, "SHARKIQ_USERNAME environment variable unset"
+    assert password is not None, "SHARKIQ_PASSWORD environment variable unset"
+
+    return get_ayla_api(username=username, password=password)
+
+
+@pytest.fixture
+def sample_api_logged_in(sample_api):
+    """Sample API object with user-supplied creds after performing auth flow."""
+    sample_api.sign_in()
+    return sample_api

tests/test_ayla_api.py

@@ -0,0 +1,223 @@
+import aiohttp
+import pytest
+from sharkiq.ayla_api import get_ayla_api, AylaApi
+from sharkiq.const import SHARK_APP_ID, SHARK_APP_SECRET
+from sharkiq.exc import (
+    SharkIqAuthError,
+    SharkIqAuthExpiringError,
+    SharkIqNotAuthedError,
+    AUTH_EXPIRED_MESSAGE,
+    NOT_AUTHED_MESSAGE,
+)
+from datetime import datetime, timedelta
+
+
+def test_get_ayla_api():
+    api = get_ayla_api("[email protected]", "mypassword")
+
+    assert api._email == "[email protected]"
+    assert api._password == "mypassword"
+    assert api._access_token is None
+    assert api._refresh_token is None
+    assert api._auth_expiration is None
+    assert api._is_authed == False
+    assert api._app_id == SHARK_APP_ID
+    assert api._app_secret == SHARK_APP_SECRET
+    assert api.websession is None
+
+
+class TestAylaApi:
+    def test_init__required_vals(self):
+        api = AylaApi(
+            "[email protected]", "mypassword", "app_id_123", "appsecret_123"
+        )
+
+        assert api._email == "[email protected]"
+        assert api._password == "mypassword"
+        assert api._access_token is None
+        assert api._refresh_token is None
+        assert api._auth_expiration is None
+        assert api._is_authed == False
+        assert api._app_id == "app_id_123"
+        assert api._app_secret == "appsecret_123"
+        assert api.websession is None
+
+    @pytest.mark.asyncio
+    async def test_ensure_session(self, dummy_api):
+        # Initially created with no websession
+        assert dummy_api.websession is None
+
+        session = await dummy_api.ensure_session()
+
+        # Check that session was created and returned
+        assert isinstance(session, aiohttp.ClientSession)
+        assert dummy_api.websession is session
+
+    def test_property__login_data(self, dummy_api):
+        assert dummy_api._login_data == {
+            "user": {
+                "email": "[email protected]",
+                "password": "mypassword",
+                "application": {
+                    "app_id": SHARK_APP_ID,
+                    "app_secret": SHARK_APP_SECRET,
+                },
+            }
+        }
+
+    def test_set_credentials__404_response(self, dummy_api):
+        with pytest.raises(SharkIqAuthError) as e:
+            dummy_api._set_credentials(404, {"error": {"message": "Not found"}})
+        assert (
+            e.value.args[0] == "Not found (Confirm app_id and app_secret are correct)"
+        )
+
+    def test_set_credentials__401_response(self, dummy_api):
+        with pytest.raises(SharkIqAuthError) as e:
+            dummy_api._set_credentials(401, {"error": {"message": "Unauthorized"}})
+        assert e.value.args[0] == "Unauthorized"
+
+    def test_set_credentials__valid_response(self, dummy_api):
+        assert dummy_api._access_token is "token123"
+        assert dummy_api._refresh_token is "token321"
+        assert dummy_api._auth_expiration.timestamp() == pytest.approx(
+            (datetime.now() + timedelta(seconds=700)).timestamp()
+        )
+        assert dummy_api._is_authed == True
+
+        t1 = datetime.now() + timedelta(seconds=3600)
+        dummy_api._set_credentials(
+            200,
+            {
+                "access_token": "token123",
+                "refresh_token": "token321",
+                "expires_in": 3600,
+            },
+        )
+
+        assert dummy_api._access_token == "token123"
+        assert dummy_api._refresh_token == "token321"
+        assert dummy_api._auth_expiration.timestamp() == pytest.approx(t1.timestamp())
+        assert dummy_api._is_authed == True
+
+    def test_property__sign_out_data(self, dummy_api):
+        assert dummy_api.sign_out_data == {
+            "user": {"access_token": dummy_api._access_token}
+        }
+
+    def test_clear_auth(self, dummy_api):
+        assert dummy_api._is_authed == True
+
+        dummy_api._clear_auth()
+
+        assert dummy_api._access_token is None
+        assert dummy_api._refresh_token is None
+        assert dummy_api._auth_expiration is None
+        assert dummy_api._is_authed == False
+
+    def test_property__auth_expiration__not_authed(self, dummy_api):
+        dummy_api._is_authed = False
+        dummy_api._auth_expiration = None
+
+        assert dummy_api.auth_expiration is None
+
+    def test_property__auth_expiration__no_expiration(self, dummy_api):
+        # mock the invalid state
+        dummy_api._is_authed = True
+        dummy_api._auth_expiration = None
+
+        # Check that the correct exception is raised when accessing property
+        with pytest.raises(SharkIqNotAuthedError) as e:
+            _ = dummy_api.auth_expiration
+        assert e.value.args[0] == "Invalid state.  Please reauthorize."
+
+    def test_property__auth_expiration__not_authed(self, dummy_api):
+        dummy_api._is_authed = True
+        t = datetime.now() + timedelta(seconds=3600)
+        dummy_api._auth_expiration = t
+
+        assert dummy_api.auth_expiration == t
+
+    def test_property__token_expired__false(self, dummy_api):
+        dummy_api._is_authed = True
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=3600)
+        assert dummy_api.token_expired == False
+
+    def test_property__token_expired__true(self, dummy_api):
+        dummy_api._is_authed = True
+        dummy_api._auth_expiration = datetime.now() - timedelta(seconds=3600)
+        assert dummy_api.token_expired == True
+
+    def test_property__token_expired__not_authed(self, dummy_api):
+        dummy_api._is_authed = False
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=3600)
+        assert dummy_api.token_expired == True
+
+    def test_property__token_expiring_soon__false(self, dummy_api):
+        dummy_api._is_authed = True
+        # "soon" is considered to be within 600 seconds from the current time
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=605)
+        assert dummy_api.token_expiring_soon == False
+
+    def test_property__token_expiring_soon__true(self, dummy_api):
+        dummy_api._is_authed = True
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=595)
+        assert dummy_api.token_expiring_soon == True
+
+    def test_property__token_expiring_soon__not_authed(self, dummy_api):
+        dummy_api._is_authed = False
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=3600)
+        assert dummy_api.token_expiring_soon == True
+
+    @pytest.mark.parametrize(
+        "access_token,auth_state,auth_timedelta",
+        [
+            ("token123", True, timedelta(seconds=-100)),  # auth expiry passed
+            (None, True, timedelta(seconds=700)),  # invalid token
+            ("token123", False, timedelta(seconds=-100)),  # not authed
+        ],
+    )
+    def test_check_auth__not_authed(
+        self, dummy_api, access_token, auth_state, auth_timedelta
+    ):
+        dummy_api._access_token = access_token
+        dummy_api._is_authed = auth_state
+        dummy_api._auth_expiration = datetime.now() + auth_timedelta
+
+        with pytest.raises(SharkIqNotAuthedError) as e:
+            dummy_api.check_auth()
+
+        assert e.value.args[0] == NOT_AUTHED_MESSAGE
+        assert dummy_api._is_authed == False
+
+    def test_check_auth__expiring_soon_exception(self, dummy_api):
+        dummy_api._auth_expiration = datetime.now() + timedelta(seconds=400)
+
+        with pytest.raises(SharkIqAuthExpiringError) as e:
+            dummy_api.check_auth(raise_expiring_soon=True)
+
+        assert e.value.args[0] == AUTH_EXPIRED_MESSAGE
+
+        # No exception raised when set to False
+        dummy_api.check_auth(raise_expiring_soon=False)
+
+    def test_check_auth__valid(self, dummy_api):
+        assert dummy_api.check_auth() is None
+
+    def test_auth_header(self, dummy_api):
+        dummy_api._access_token = "myfaketoken"
+        assert dummy_api.auth_header == {
+            "Authorization": "auth_token myfaketoken"
+        }
+
+    def test_get_headers__no_kwargs(self, dummy_api):
+        headers = dummy_api._get_headers({})
+        assert headers == dummy_api.auth_header
+
+    def test_get_headers__kwargs_(self, dummy_api):
+        headers = dummy_api._get_headers({"headers": {"X-Test": "val"}})
+
+        assert headers == {
+            "X-Test": "val",
+            "Authorization": f"auth_token {dummy_api._access_token}"
+        }