Add (very basic) test

Author: CoMPaTech

.pre-commit-config.yaml

@@ -40,16 +40,16 @@ repos:
           - --quiet-level=2
         exclude_types: [csv, json]
         exclude: ^userdata/|^fixtures/
-#  - repo: https://github.com/PyCQA/bandit
-#    rev: 1.8.5
-#    hooks:
-#      - id: bandit
-#        name: "Bandit checking"
-#        args:
-#          - --quiet
-#          - --format=custom
-#          - --configfile=tests/bandit.yaml
-#        files: ^(airos|tests)/.+\.py$
+  - repo: https://github.com/PyCQA/bandit
+    rev: 1.8.5
+    hooks:
+      - id: bandit
+        name: "Bandit checking"
+        args:
+          - --quiet
+          - --format=custom
+          - --configfile=tests/bandit.yaml
+        files: ^(airos|tests)/.+\.py$
   - repo: https://github.com/adrienverge/yamllint.git
     rev: v1.37.1
     hooks:

airos/airos8.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+from urllib.parse import urlparse
 
 import aiohttp
 
@@ -12,7 +13,7 @@
 logger = logging.getLogger(__name__)
 
 
-class AirOS8:
+class AirOS:
     """Set up connection to AirOS."""
 
     def __init__(
@@ -21,12 +22,22 @@ def __init__(
         username: str,
         password: str,
         session: aiohttp.ClientSession,
+        use_ssl: bool = True,
         verify_ssl: bool = True,
     ):
         """Initialize AirOS8 class."""
         self.username = username
         self.password = password
-        self.base_url = f"https://{host}"
+
+        parsed_host = urlparse(host)
+        scheme = (
+            parsed_host.scheme
+            if parsed_host.scheme
+            else ("https" if use_ssl else "http")
+        )
+        hostname = parsed_host.hostname if parsed_host.hostname else host
+
+        self.base_url = f"{scheme}://{hostname}"
 
         self.session = session
         self.verify_ssl = verify_ssl
@@ -51,6 +62,8 @@ def __init__(
             "X-Requested-With": "XMLHttpRequest",
         }
 
+        self.connected = False
+
     async def login(self) -> bool:
         """Log in to the device assuring cookies and tokens set correctly."""
         # --- Step 0: Pre-inject the 'ok=1' cookie before login POST (mimics curl) ---
@@ -146,11 +159,10 @@ async def login(self) -> bool:
                 if not airos_cookie_found and not ok_cookie_found:
                     raise DataMissingError from None
 
-                response_text = await response.text()
-
                 if response.status == 200:
                     try:
-                        json.loads(response_text)
+                        json.loads(response.text)
+                        self.connected = True
                         return True
                     except json.JSONDecodeError as err:
                         logger.exception("JSON Decode Error")
@@ -166,6 +178,10 @@ async def login(self) -> bool:
 
     async def status(self) -> dict:
         """Retrieve status from the device."""
+        if not self.connected:
+            logger.error("Not connected, login first")
+            raise ConnectionFailedError from None
+
         # --- Step 2: Verify authenticated access by fetching status.cgi ---
         authenticated_get_headers = {**self._common_headers}
         if self.current_csrf_token:
@@ -177,18 +193,16 @@ async def status(self) -> dict:
                 headers=authenticated_get_headers,
                 ssl=self.verify_ssl,
             ) as response:
-                status_response_text = await response.text()
-
                 if response.status == 200:
                     try:
-                        return json.loads(status_response_text)
+                        return json.loads(response.text)
                     except json.JSONDecodeError:
                         logger.exception(
                             "JSON Decode Error in authenticated status response"
                         )
                         raise DataMissingError from None
                 else:
-                    log = f"Authenticated status.cgi failed: {response.status}. Response: {status_response_text}"
+                    log = f"Authenticated status.cgi failed: {response.status}. Response: {response.text}"
                     logger.error(log)
         except aiohttp.ClientError as err:
             logger.exception("Error during authenticated status.cgi call")

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "airos"
-version         = "0.0.4"
+version         = "0.0.5"
 license         = "MIT"
 description     = "Ubiquity airOS module(s) for Python 3."
 readme          = "README.md"
@@ -534,4 +534,4 @@ testpaths = [
     "tests",
 ]
 asyncio_default_fixture_loop_scope = "session"
-asyncio_mode = "strict"
+asyncio_mode = "auto"

requirements-test.txt

@@ -1,3 +1,4 @@
 pytest
 pytest-asyncio
 aiohttp
+aioresponses

tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Ubiquity AirOS python module."""

tests/bandit.yaml

@@ -0,0 +1,20 @@
+# https://bandit.readthedocs.io/en/latest/config.html
+
+tests:
+  - B103
+  - B108
+  - B306
+  - B307
+  - B313
+  - B314
+  - B315
+  - B316
+  - B317
+  - B318
+  - B319
+  - B320
+  - B601
+  - B602
+  - B604
+  - B608
+  - B609

tests/conftest.py

@@ -0,0 +1,21 @@
+"""Ubiquity AirOS test fixtures."""
+
+from airos.airos8 import AirOS
+import pytest
+
+import aiohttp
+
+
+@pytest.fixture
+def base_url():
+    """Return a testing url."""
+    return "http://device.local"
+
+
+@pytest.fixture
+async def airos_device(base_url):
+    """AirOS device fixture."""
+    session = aiohttp.ClientSession(cookie_jar=aiohttp.CookieJar())
+    instance = AirOS(base_url, "username", "password", session, use_ssl=False)
+    yield instance
+    await session.close()

tests/test_stations.py

@@ -0,0 +1,51 @@
+"""Ubiquity AirOS tests."""
+
+from http.cookies import SimpleCookie
+import json
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+import aiofiles
+
+
+async def _read_fixture(fixture: str = "ap-ptp"):
+    """Read fixture file per device type."""
+    path = os.path.join(os.path.dirname(__file__), f"../fixtures/{fixture}.json")
+    async with aiofiles.open(path, encoding="utf-8") as f:
+        return json.loads(await f.read())
+
+
+@pytest.mark.parametrize("mode", ["ap-ptp", "sta-ptp"])
+@pytest.mark.asyncio
+async def test_ap(airos_device, base_url, mode):
+    """Test device operation."""
+    cookie = SimpleCookie()
+    cookie["session_id"] = "test-cookie"
+    cookie["AIROS_TOKEN"] = "abc123"
+
+    # --- Prepare fake POST /api/auth response with cookies ---
+    mock_login_response = MagicMock()
+    mock_login_response.__aenter__.return_value = mock_login_response
+    mock_login_response.text = "{}"
+    mock_login_response.status = 200
+    mock_login_response.cookies = cookie
+
+    # --- Prepare fake GET /api/status response ---
+    mock_status_payload = {"mode": await _read_fixture(fixture=mode)}
+    mock_status_response = MagicMock()
+    mock_status_response.__aenter__.return_value = mock_status_response
+    mock_status_response.text = json.dumps(await _read_fixture(mode))
+    mock_status_response.status = 200
+    mock_status_response.json = AsyncMock(return_value=mock_status_payload)
+
+    with (
+        patch.object(airos_device.session, "post", return_value=mock_login_response),
+        patch.object(airos_device.session, "get", return_value=mock_status_response),
+    ):
+        assert await airos_device.login()
+        status = await airos_device.status()
+
+        # Verify the fixture returns the correct mode
+        assert status.get("wireless", {}).get("mode") == mode