diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 73bb19e..1792ad7 100644 --- a/.pre-commit-config.yaml +++ b/.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: diff --git a/airos/airos8.py b/airos/airos8.py index e5309b0..229192b 100644 --- a/airos/airos8.py +++ b/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") diff --git a/pyproject.toml b/pyproject.toml index a988a1b..f816fcd 100644 --- a/pyproject.toml +++ b/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" diff --git a/requirements-test.txt b/requirements-test.txt index 74d3b66..e4eb679 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1,3 +1,4 @@ pytest pytest-asyncio aiohttp +aioresponses diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..4e5cb37 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the Ubiquity AirOS python module.""" diff --git a/tests/bandit.yaml b/tests/bandit.yaml new file mode 100644 index 0000000..46566cc --- /dev/null +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d757c4c --- /dev/null +++ b/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() diff --git a/tests/test_stations.py b/tests/test_stations.py new file mode 100644 index 0000000..bf1b9ab --- /dev/null +++ b/tests/test_stations.py @@ -0,0 +1,58 @@ +"""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.""" + fixture_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + path = os.path.join(fixture_dir, f"{fixture}.json") + try: + async with aiofiles.open(path, encoding="utf-8") as f: + return json.loads(await f.read()) + except FileNotFoundError: + pytest.fail(f"Fixture file not found: {path}") + except json.JSONDecodeError as e: + pytest.fail(f"Invalid JSON in fixture file {path}: {e}") + + +@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 + mock_login_response.headers = {"X-CSRF-ID": "test-csrf-token"} + # --- Prepare fake GET /api/status response --- + fixture_data = await _read_fixture(mode) + mock_status_payload = fixture_data + mock_status_response = MagicMock() + mock_status_response.__aenter__.return_value = mock_status_response + mock_status_response.text = json.dumps(fixture_data) + 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