From 5983eea5475c83135d254c3edb43c19af313a813 Mon Sep 17 00:00:00 2001 From: Tom Scholten Date: Tue, 12 Aug 2025 20:03:50 +0200 Subject: [PATCH 1/4] Adding mypy --- .github/workflows/verify.yml | 3 +- .pre-commit-config.yaml | 9 ++++ airos/airos8.py | 24 +++++----- airos/data.py | 13 +++--- airos/discovery.py | 4 +- pyproject.toml | 8 +++- requirements-test.txt | 1 + script/generate_ha_fixture.py | 7 +-- script/mashumaro-step-debug.py | 2 +- script/run-in-env.sh | 32 ++++++++++++++ tests/conftest.py | 9 ++-- tests/test_airos8.py | 39 ++++++++-------- tests/test_data.py | 2 +- tests/test_discovery.py | 77 ++++++++++++++++++++------------ tests/test_stations.py | 81 +++++++++++++++++++++------------- 15 files changed, 199 insertions(+), 112 deletions(-) create mode 100755 script/run-in-env.sh diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 879ff42..77503e9 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -106,7 +106,6 @@ jobs: include-hidden-files: true mypy: - if: false # disables the job --> "Code is not up to par for mypy, skipping" runs-on: ubuntu-latest name: Run mypy needs: @@ -135,7 +134,7 @@ jobs: needs: - ruff - pytest - # - mypy + - mypy steps: - name: Check out committed code uses: actions/checkout@v4 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 12fa535..391a39a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -80,3 +80,12 @@ repos: rev: v0.45.0 hooks: - id: markdownlint + - repo: local + hooks: + - id: mypy + name: mypy + entry: script/run-in-env.sh mypy + language: script + require_serial: true + types_or: [python, pyi] + files: ^(airos|tests|scripts)/.+\.(py|pyi)$ diff --git a/airos/airos8.py b/airos/airos8.py index 6c4762b..c8a35cc 100644 --- a/airos/airos8.py +++ b/airos/airos8.py @@ -54,7 +54,7 @@ def __init__( self._status_cgi_url = f"{self.base_url}/status.cgi" # AirOS 8 self._stakick_cgi_url = f"{self.base_url}/stakick.cgi" # AirOS 8 self._provmode_url = f"{self.base_url}/api/provmode" # AirOS 8 - self.current_csrf_token = None + self.current_csrf_token: str | None = None self._use_json_for_login_post = False @@ -87,7 +87,7 @@ async def login(self) -> bool: login_request_headers = {**self._common_headers} - post_data = None + post_data: dict[str, str] | str | None = None if self._use_json_for_login_post: login_request_headers["Content-Type"] = "application/json" post_data = json.dumps(login_payload) @@ -114,7 +114,7 @@ async def login(self) -> bool: # If the AIROS_ cookie was parsed but isn't automatically added to the jar, add it manually if ( morsel.key.startswith("AIROS_") - and morsel.key not in self.session.cookie_jar + and morsel.key not in self.session.cookie_jar # type: ignore[operator] ): # `SimpleCookie`'s Morsel objects are designed to be compatible with cookie jars. # We need to set the domain if it's missing, otherwise the cookie might not be sent. @@ -152,7 +152,7 @@ async def login(self) -> bool: if new_csrf_token: self.current_csrf_token = new_csrf_token else: - return + return False # Re-check cookies in self.session.cookie_jar AFTER potential manual injection airos_cookie_found = False @@ -186,18 +186,16 @@ async def login(self) -> bool: log = f"Login failed with status {response.status}. Full Response: {response.text}" _LOGGER.error(log) raise AirOSConnectionAuthenticationError from None - except (TimeoutError, aiohttp.client_exceptions.ClientError) as err: + except (TimeoutError, aiohttp.ClientError) as err: _LOGGER.exception("Error during login") raise AirOSDeviceConnectionError from err except asyncio.CancelledError: _LOGGER.info("Login task was cancelled") raise - def derived_data( - self, response: dict[str, Any] | None = None - ) -> dict[str, Any] | None: + def derived_data(self, response: dict[str, Any] = {}) -> dict[str, Any]: """Add derived data to the device response.""" - derived = { + derived: dict[str, Any] = { "station": False, "access_point": False, "ptp": False, @@ -302,14 +300,14 @@ async def status(self) -> AirOSData: response_text, ) raise AirOSDeviceConnectionError - except (TimeoutError, aiohttp.client_exceptions.ClientError) as err: + except (TimeoutError, aiohttp.ClientError) as err: _LOGGER.exception("Status API call failed: %s", err) raise AirOSDeviceConnectionError from err except asyncio.CancelledError: _LOGGER.info("API status retrieval task was cancelled") raise - async def stakick(self, mac_address: str = None) -> bool: + async def stakick(self, mac_address: str | None = None) -> bool: """Reconnect client station.""" if not self.connected: _LOGGER.error("Not connected, login first") @@ -340,7 +338,7 @@ async def stakick(self, mac_address: str = None) -> bool: log = f"Unable to restart connection response status {response.status} with {response_text}" _LOGGER.error(log) return False - except (TimeoutError, aiohttp.client_exceptions.ClientError) as err: + except (TimeoutError, aiohttp.ClientError) as err: _LOGGER.exception("Error during call to reconnect remote: %s", err) raise AirOSDeviceConnectionError from err except asyncio.CancelledError: @@ -379,7 +377,7 @@ async def provmode(self, active: bool = False) -> bool: log = f"Unable to change provisioning mode response status {response.status} with {response_text}" _LOGGER.error(log) return False - except (TimeoutError, aiohttp.client_exceptions.ClientError) as err: + except (TimeoutError, aiohttp.ClientError) as err: _LOGGER.exception("Error during call to change provisioning mode: %s", err) raise AirOSDeviceConnectionError from err except asyncio.CancelledError: diff --git a/airos/data.py b/airos/data.py index d3377f8..0cbb4d1 100644 --- a/airos/data.py +++ b/airos/data.py @@ -39,7 +39,7 @@ def is_ip_address(value: str) -> bool: return False -def redact_data_smart(data: dict) -> dict: +def redact_data_smart(data: dict[str, Any]) -> dict[str, Any]: """Recursively redacts sensitive keys in a dictionary.""" sensitive_keys = { "hostname", @@ -56,10 +56,7 @@ def redact_data_smart(data: dict) -> dict: "platform", } - def _redact(d: dict): - if not isinstance(d, dict): - return d - + def _redact(d: dict[str, Any]) -> dict[str, Any]: redacted_d = {} for k, v in d.items(): if k in sensitive_keys: @@ -73,15 +70,15 @@ def _redact(d: dict): isinstance(i, str) and is_ip_address(i) for i in v ): # Redact list of IPs to a dummy list - redacted_d[k] = ["127.0.0.3"] + redacted_d[k] = ["127.0.0.3"] # type: ignore[assignment] else: redacted_d[k] = "REDACTED" elif isinstance(v, dict): - redacted_d[k] = _redact(v) + redacted_d[k] = _redact(v) # type: ignore[assignment] elif isinstance(v, list): redacted_d[k] = [ _redact(item) if isinstance(item, dict) else item for item in v - ] + ] # type: ignore[assignment] else: redacted_d[k] = v return redacted_d diff --git a/airos/discovery.py b/airos/discovery.py index 7b1b6de..d473d6d 100644 --- a/airos/discovery.py +++ b/airos/discovery.py @@ -29,7 +29,7 @@ class AirOSDiscoveryProtocol(asyncio.DatagramProtocol): """ - def __init__(self, callback: Callable[[dict[str, Any]], None]) -> None: + def __init__(self, callback: Callable[[dict[str, Any]], Any]) -> None: """Initialize AirOSDiscoveryProtocol. Args: @@ -43,7 +43,7 @@ def __init__(self, callback: Callable[[dict[str, Any]], None]) -> None: def connection_made(self, transport: asyncio.BaseTransport) -> None: """Set up the UDP socket for broadcasting and reusing the address.""" self.transport = transport # type: ignore[assignment] # transport is DatagramTransport - sock: socket.socket = self.transport.get_extra_info("socket") + sock: socket.socket = transport.get_extra_info("socket") sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) log = f"AirOS discovery listener (low-level) started on UDP port {DISCOVERY_PORT}." diff --git a/pyproject.toml b/pyproject.toml index a967d6e..2e3f95a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "airos" -version = "0.2.8" +version = "0.2.9a0" license = "MIT" description = "Ubiquity airOS module(s) for Python 3." readme = "README.md" @@ -396,6 +396,12 @@ warn_return_any = true warn_unreachable = true exclude = [] +[[tool.mypy.overrides]] +module = "tests.*" +ignore_missing_imports = true # You'll likely need this for test-only dependencies +disallow_untyped_decorators = false # The fix for your current errors +check_untyped_defs = false + [tool.coverage.run] source = [ "airos" ] omit= [ diff --git a/requirements-test.txt b/requirements-test.txt index 5e63144..df67449 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -7,3 +7,4 @@ aioresponses aioresponses==0.7.8 aiofiles==24.1.0 radon==6.0.1 +types-aiofiles==24.1.0.20250809 \ No newline at end of file diff --git a/script/generate_ha_fixture.py b/script/generate_ha_fixture.py index d553c1f..d189b64 100644 --- a/script/generate_ha_fixture.py +++ b/script/generate_ha_fixture.py @@ -15,10 +15,11 @@ # NOTE: This assumes the airos module is correctly installed or available in the project path. # If not, you might need to adjust the import statement. -from airos.airos8 import AirOS, AirOSData # noqa: E402 +from airos.airos8 import AirOS # noqa: E402 +from airos.data import AirOS8Data as AirOSData # noqa: E402 -def generate_airos_fixtures(): +def generate_airos_fixtures() -> None: """Process all (intended) JSON files from the userdata directory to potential fixtures.""" # Define the paths to the directories @@ -44,7 +45,7 @@ def generate_airos_fixtures(): with open(base_fixture_path) as source: source_data = json.loads(source.read()) - derived_data = AirOS.derived_data(None, source_data) + derived_data = AirOS.derived_data(None, source_data) # type: ignore[arg-type] new_data = AirOSData.from_dict(derived_data) with open(new_fixture_path, "w") as new: diff --git a/script/mashumaro-step-debug.py b/script/mashumaro-step-debug.py index dce5f2d..628981e 100644 --- a/script/mashumaro-step-debug.py +++ b/script/mashumaro-step-debug.py @@ -18,7 +18,7 @@ _LOGGER = logging.getLogger(__name__) -def main(): +def main() -> None: """Debug data.""" if len(sys.argv) <= 1: _LOGGER.info("Use with file to check") diff --git a/script/run-in-env.sh b/script/run-in-env.sh new file mode 100755 index 0000000..14bd9a3 --- /dev/null +++ b/script/run-in-env.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env sh +set -eu + +# Used in venv activate script. +# Would be an error if undefined. +OSTYPE="${OSTYPE-}" + +# Activate pyenv and virtualenv if present, then run the specified command + +# pyenv, pyenv-virtualenv +if [ -s .python-version ]; then + PYENV_VERSION=$(head -n 1 .python-version) + export PYENV_VERSION +fi + +# shellcheck source=/dev/null +if [ -n "${VIRTUAL_ENV-}" ] && [ -f "${VIRTUAL_ENV}/bin/activate" ]; then + # shellcheck source=/dev/null + . "${VIRTUAL_ENV}/bin/activate" +else + # other common virtualenvs + my_path=$(git rev-parse --show-toplevel) + + for venv in venv .venv .; do + if [ -f "${my_path}/${venv}/bin/activate" ]; then + . "${my_path}/${venv}/bin/activate" + break + fi + done +fi + +exec "$@" diff --git a/tests/conftest.py b/tests/conftest.py index 29aab81..ae280a5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ """Ubiquity AirOS test fixtures.""" +from _collections_abc import AsyncGenerator, Generator import asyncio from unittest.mock import AsyncMock, MagicMock, patch @@ -11,13 +12,13 @@ @pytest.fixture -def base_url(): +def base_url() -> str: """Return a testing url.""" return "http://device.local" @pytest.fixture -async def airos_device(base_url): +async def airos_device(base_url: str) -> AsyncGenerator[AirOS, None]: """AirOS device fixture.""" session = aiohttp.ClientSession(cookie_jar=aiohttp.CookieJar()) instance = AirOS(base_url, "username", "password", session, use_ssl=False) @@ -26,7 +27,9 @@ async def airos_device(base_url): @pytest.fixture -def mock_datagram_endpoint(): +def mock_datagram_endpoint() -> Generator[ + tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], None, None +]: """Fixture to mock the creation of the UDP datagram endpoint.""" # Define the mock objects FIRST, so they are in scope mock_transport = MagicMock(spec=asyncio.DatagramTransport) diff --git a/tests/test_airos8.py b/tests/test_airos8.py index 5f5b01d..e88afab 100644 --- a/tests/test_airos8.py +++ b/tests/test_airos8.py @@ -4,6 +4,7 @@ import json from unittest.mock import AsyncMock, MagicMock, patch +from airos.airos8 import AirOS import airos.exceptions import pytest @@ -13,7 +14,7 @@ # --- Tests for Login and Connection Errors --- @pytest.mark.asyncio -async def test_login_no_csrf_token(airos_device): +async def test_login_no_csrf_token(airos_device: AirOS) -> None: """Test login response without a CSRF token header.""" cookie = SimpleCookie() cookie["AIROS_TOKEN"] = "abc" @@ -28,11 +29,11 @@ async def test_login_no_csrf_token(airos_device): with patch.object(airos_device.session, "post", return_value=mock_login_response): # We expect a return of None as the CSRF token is missing result = await airos_device.login() - assert result is None + assert result is False @pytest.mark.asyncio -async def test_login_connection_error(airos_device): +async def test_login_connection_error(airos_device: AirOS) -> None: """Test aiohttp ClientError during login attempt.""" with ( patch.object(airos_device.session, "post", side_effect=aiohttp.ClientError), @@ -43,7 +44,7 @@ async def test_login_connection_error(airos_device): # --- Tests for status() and derived_data() logic --- @pytest.mark.asyncio -async def test_status_when_not_connected(airos_device): +async def test_status_when_not_connected(airos_device: AirOS) -> None: """Test calling status() before a successful login.""" airos_device.connected = False # Ensure connected state is false with pytest.raises(airos.exceptions.AirOSDeviceConnectionError): @@ -51,7 +52,7 @@ async def test_status_when_not_connected(airos_device): @pytest.mark.asyncio -async def test_status_non_200_response(airos_device): +async def test_status_non_200_response(airos_device: AirOS) -> None: """Test status() with a non-successful HTTP response.""" airos_device.connected = True mock_status_response = MagicMock() @@ -67,7 +68,7 @@ async def test_status_non_200_response(airos_device): @pytest.mark.asyncio -async def test_status_invalid_json_response(airos_device): +async def test_status_invalid_json_response(airos_device: AirOS) -> None: """Test status() with a response that is not valid JSON.""" airos_device.connected = True mock_status_response = MagicMock() @@ -83,7 +84,7 @@ async def test_status_invalid_json_response(airos_device): @pytest.mark.asyncio -async def test_status_missing_interface_key_data(airos_device): +async def test_status_missing_interface_key_data(airos_device: AirOS) -> None: """Test status() with a response missing critical data fields.""" airos_device.connected = True # The derived_data() function is called with a mocked response @@ -102,7 +103,7 @@ async def test_status_missing_interface_key_data(airos_device): @pytest.mark.asyncio -async def test_derived_data_no_interfaces_key(airos_device): +async def test_derived_data_no_interfaces_key(airos_device: AirOS) -> None: """Test derived_data() with a response that has no 'interfaces' key.""" # This will directly test the 'if not interfaces:' branch (line 206) with pytest.raises(airos.exceptions.AirOSKeyDataMissingError): @@ -110,7 +111,7 @@ async def test_derived_data_no_interfaces_key(airos_device): @pytest.mark.asyncio -async def test_derived_data_no_br0_eth0_ath0(airos_device): +async def test_derived_data_no_br0_eth0_ath0(airos_device: AirOS) -> None: """Test derived_data() with an unexpected interface list, to test the fallback logic.""" fixture_data = { "interfaces": [ @@ -125,7 +126,7 @@ async def test_derived_data_no_br0_eth0_ath0(airos_device): # --- Tests for stakick() --- @pytest.mark.asyncio -async def test_stakick_when_not_connected(airos_device): +async def test_stakick_when_not_connected(airos_device: AirOS) -> None: """Test stakick() before a successful login.""" airos_device.connected = False with pytest.raises(airos.exceptions.AirOSDeviceConnectionError): @@ -133,7 +134,7 @@ async def test_stakick_when_not_connected(airos_device): @pytest.mark.asyncio -async def test_stakick_no_mac_address(airos_device): +async def test_stakick_no_mac_address(airos_device: AirOS) -> None: """Test stakick() with a None mac_address.""" airos_device.connected = True with pytest.raises(airos.exceptions.AirOSDataMissingError): @@ -141,7 +142,7 @@ async def test_stakick_no_mac_address(airos_device): @pytest.mark.asyncio -async def test_stakick_non_200_response(airos_device): +async def test_stakick_non_200_response(airos_device: AirOS) -> None: """Test stakick() with a non-successful HTTP response.""" airos_device.connected = True mock_stakick_response = MagicMock() @@ -154,7 +155,7 @@ async def test_stakick_non_200_response(airos_device): @pytest.mark.asyncio -async def test_stakick_connection_error(airos_device): +async def test_stakick_connection_error(airos_device: AirOS) -> None: """Test aiohttp ClientError during stakick.""" airos_device.connected = True with ( @@ -166,7 +167,7 @@ async def test_stakick_connection_error(airos_device): # --- Tests for provmode() (Complete Coverage) --- @pytest.mark.asyncio -async def test_provmode_when_not_connected(airos_device): +async def test_provmode_when_not_connected(airos_device: AirOS) -> None: """Test provmode() before a successful login.""" airos_device.connected = False with pytest.raises(airos.exceptions.AirOSDeviceConnectionError): @@ -174,7 +175,7 @@ async def test_provmode_when_not_connected(airos_device): @pytest.mark.asyncio -async def test_provmode_activate_success(airos_device): +async def test_provmode_activate_success(airos_device: AirOS) -> None: """Test successful activation of provisioning mode.""" airos_device.connected = True mock_provmode_response = MagicMock() @@ -188,7 +189,7 @@ async def test_provmode_activate_success(airos_device): @pytest.mark.asyncio -async def test_provmode_deactivate_success(airos_device): +async def test_provmode_deactivate_success(airos_device: AirOS) -> None: """Test successful deactivation of provisioning mode.""" airos_device.connected = True mock_provmode_response = MagicMock() @@ -202,7 +203,7 @@ async def test_provmode_deactivate_success(airos_device): @pytest.mark.asyncio -async def test_provmode_non_200_response(airos_device): +async def test_provmode_non_200_response(airos_device: AirOS) -> None: """Test provmode() with a non-successful HTTP response.""" airos_device.connected = True mock_provmode_response = MagicMock() @@ -217,7 +218,7 @@ async def test_provmode_non_200_response(airos_device): @pytest.mark.asyncio -async def test_provmode_connection_error(airos_device): +async def test_provmode_connection_error(airos_device: AirOS) -> None: """Test aiohttp ClientError during provmode.""" airos_device.connected = True with ( @@ -228,7 +229,7 @@ async def test_provmode_connection_error(airos_device): @pytest.mark.asyncio -async def test_status_missing_required_key_in_json(airos_device): +async def test_status_missing_required_key_in_json(airos_device: AirOS) -> None: """Test status() with a response missing a key required by the dataclass.""" airos_device.connected = True # Fixture is valid JSON, but is missing the entire 'wireless' block, diff --git a/tests/test_data.py b/tests/test_data.py index c4aad42..dc319d8 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -7,7 +7,7 @@ @pytest.mark.asyncio -async def test_unknown_enum_values(): +async def test_unknown_enum_values() -> None: """Test that unknown enum values are handled gracefully.""" # 1. Test for Host.netrole host_data = {"netrole": "unsupported_role", "other_field": "value"} diff --git a/tests/test_discovery.py b/tests/test_discovery.py index fe20bb0..1857c4f 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -1,8 +1,10 @@ """Test discovery of Ubiquiti airOS devices.""" import asyncio +from collections.abc import Callable import os -import socket # Add this import +import socket +from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch from airos.discovery import ( @@ -21,7 +23,8 @@ async def _read_binary_fixture(fixture_name: str) -> bytes: path = os.path.join(fixture_dir, fixture_name) try: - def _read_file(): + def _read_file() -> bytes: + """Read the fixture file.""" with open(path, "rb") as f: return f.read() @@ -39,7 +42,7 @@ async def mock_airos_packet() -> bytes: @pytest.mark.asyncio -async def test_parse_airos_packet_success(mock_airos_packet): +async def test_parse_airos_packet_success(mock_airos_packet: bytes) -> None: """Test parse_airos_packet with a valid packet containing scrubbed data.""" protocol = AirOSDiscoveryProtocol( AsyncMock() @@ -63,7 +66,7 @@ async def test_parse_airos_packet_success(mock_airos_packet): @pytest.mark.asyncio -async def test_parse_airos_packet_invalid_header(): +async def test_parse_airos_packet_invalid_header() -> None: """Test parse_airos_packet with an invalid header.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) invalid_data = b"\x00\x00\x00\x00\x00\x00" + b"someotherdata" @@ -81,7 +84,7 @@ async def test_parse_airos_packet_invalid_header(): @pytest.mark.asyncio -async def test_parse_airos_packet_too_short(): +async def test_parse_airos_packet_too_short() -> None: """Test parse_airos_packet with data too short for header.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) too_short_data = b"\x01\x06\x00" @@ -99,7 +102,7 @@ async def test_parse_airos_packet_too_short(): @pytest.mark.asyncio -async def test_parse_airos_packet_truncated_tlv(): +async def test_parse_airos_packet_truncated_tlv() -> None: """Test parse_airos_packet with a truncated TLV.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) # Header + MAC TLV (valid) + then a truncated TLV_IP @@ -117,7 +120,7 @@ async def test_parse_airos_packet_truncated_tlv(): @pytest.mark.asyncio -async def test_datagram_received_calls_callback(mock_airos_packet): +async def test_datagram_received_calls_callback(mock_airos_packet: bytes) -> None: """Test that datagram_received correctly calls the callback.""" mock_callback = AsyncMock() protocol = AirOSDiscoveryProtocol(mock_callback) @@ -141,7 +144,7 @@ async def test_datagram_received_calls_callback(mock_airos_packet): @pytest.mark.asyncio -async def test_datagram_received_handles_parsing_error(): +async def test_datagram_received_handles_parsing_error() -> None: """Test datagram_received handles exceptions during parsing.""" mock_callback = AsyncMock() protocol = AirOSDiscoveryProtocol(mock_callback) @@ -157,7 +160,7 @@ async def test_datagram_received_handles_parsing_error(): @pytest.mark.asyncio -async def test_connection_made_sets_transport(): +async def test_connection_made_sets_transport() -> None: """Test connection_made sets up transport and socket options.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) mock_transport = MagicMock(spec=asyncio.DatagramTransport) @@ -174,7 +177,7 @@ async def test_connection_made_sets_transport(): @pytest.mark.asyncio -async def test_connection_lost_without_exception(): +async def test_connection_lost_without_exception() -> None: """Test connection_lost without an exception.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) with patch("airos.discovery._LOGGER.debug") as mock_log_debug: @@ -185,7 +188,7 @@ async def test_connection_lost_without_exception(): @pytest.mark.asyncio -async def test_connection_lost_with_exception(): +async def test_connection_lost_with_exception() -> None: """Test connection_lost with an exception.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) test_exception = Exception("Test connection lost error") @@ -200,7 +203,7 @@ async def test_connection_lost_with_exception(): @pytest.mark.asyncio -async def test_error_received(): +async def test_error_received() -> None: """Test error_received logs the error.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) test_exception = Exception("Test network error") @@ -216,15 +219,17 @@ async def test_error_received(): @pytest.mark.asyncio async def test_async_discover_devices_success( - mock_airos_packet, mock_datagram_endpoint -): + mock_airos_packet: bytes, + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: """Test the high-level discovery function on a successful run.""" mock_transport, mock_protocol_instance = mock_datagram_endpoint discovered_devices = {} - def mock_protocol_factory(callback): - def inner_callback(device_info): + def mock_protocol_factory(callback: Callable[[Any], None]) -> MagicMock: + def inner_callback(device_info: dict[str, Any]) -> None: + """Inner callback to process discovered device info.""" mac_address = device_info.get("mac_address") if mac_address: discovered_devices[mac_address] = device_info @@ -236,7 +241,8 @@ def inner_callback(device_info): new=MagicMock(side_effect=mock_protocol_factory), ): - async def _simulate_discovery(): + async def _simulate_discovery() -> None: + """Simulate the discovery process by sending a mock packet.""" await asyncio.sleep(0.1) protocol = AirOSDiscoveryProtocol( @@ -255,11 +261,14 @@ async def _simulate_discovery(): assert "01:23:45:67:89:CD" in discovered_devices assert discovered_devices["01:23:45:67:89:CD"]["hostname"] == "name" - mock_transport.close.assert_called_once() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_called_once() @pytest.mark.asyncio -async def test_async_discover_devices_no_devices(mock_datagram_endpoint): +async def test_async_discover_devices_no_devices( + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: """Test discovery returns an empty dict if no devices are found.""" mock_transport, _ = mock_datagram_endpoint @@ -267,11 +276,14 @@ async def test_async_discover_devices_no_devices(mock_datagram_endpoint): result = await airos_discover_devices(timeout=1) assert result == {} - mock_transport.close.assert_called_once() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_called_once() @pytest.mark.asyncio -async def test_async_discover_devices_oserror(mock_datagram_endpoint): +async def test_async_discover_devices_oserror( + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: """Test discovery handles OSError during endpoint creation.""" mock_transport, _ = mock_datagram_endpoint @@ -287,11 +299,14 @@ async def test_async_discover_devices_oserror(mock_datagram_endpoint): await airos_discover_devices(timeout=1) assert "address_in_use" in str(excinfo.value) - mock_transport.close.assert_not_called() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_not_called() @pytest.mark.asyncio -async def test_async_discover_devices_cancelled(mock_datagram_endpoint): +async def test_async_discover_devices_cancelled( + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: """Test discovery handles CancelledError during the timeout.""" mock_transport, _ = mock_datagram_endpoint @@ -303,11 +318,12 @@ async def test_async_discover_devices_cancelled(mock_datagram_endpoint): await airos_discover_devices(timeout=1) assert "cannot_connect" in str(excinfo.value) - mock_transport.close.assert_called_once() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_called_once() @pytest.mark.asyncio -async def test_datagram_received_handles_general_exception(): +async def test_datagram_received_handles_general_exception() -> None: """Test datagram_received handles a generic exception during parsing.""" mock_callback = AsyncMock() protocol = AirOSDiscoveryProtocol(mock_callback) @@ -345,7 +361,9 @@ async def test_datagram_received_handles_general_exception(): ], ) @pytest.mark.asyncio -async def test_parse_airos_packet_tlv_edge_cases(packet_fragment, error_message): +async def test_parse_airos_packet_tlv_edge_cases( + packet_fragment: bytes, error_message: str +) -> None: """Test parsing of various malformed TLV entries.""" protocol = AirOSDiscoveryProtocol(AsyncMock()) # A valid header is required to get to the TLV parsing stage @@ -360,7 +378,9 @@ async def test_parse_airos_packet_tlv_edge_cases(packet_fragment, error_message) @pytest.mark.asyncio -async def test_async_discover_devices_generic_oserror(mock_datagram_endpoint): +async def test_async_discover_devices_generic_oserror( + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: """Test discovery handles a generic OSError during endpoint creation.""" mock_transport, _ = mock_datagram_endpoint @@ -376,4 +396,5 @@ async def test_async_discover_devices_generic_oserror(mock_datagram_endpoint): await airos_discover_devices(timeout=1) assert "cannot_connect" in str(excinfo.value) - mock_transport.close.assert_not_called() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_not_called() diff --git a/tests/test_stations.py b/tests/test_stations.py index 2c64ba7..5435e3f 100644 --- a/tests/test_stations.py +++ b/tests/test_stations.py @@ -3,11 +3,18 @@ from http.cookies import SimpleCookie import json import os +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch +from airos.airos8 import AirOS from airos.data import AirOS8Data as AirOSData, Wireless -import airos.exceptions -from airos.exceptions import AirOSDeviceConnectionError, AirOSKeyDataMissingError +from airos.exceptions import ( + AirOSConnectionAuthenticationError, + AirOSConnectionSetupError, + AirOSDataMissingError, + AirOSDeviceConnectionError, + AirOSKeyDataMissingError, +) import pytest import aiofiles @@ -15,7 +22,7 @@ from mashumaro.exceptions import MissingField -async def _read_fixture(fixture: str = "loco5ac_ap-ptp"): +async def _read_fixture(fixture: str = "loco5ac_ap-ptp") -> Any: """Read fixture file per device type.""" fixture_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures", "userdata") path = os.path.join(fixture_dir, f"{fixture}.json") @@ -30,7 +37,9 @@ async def _read_fixture(fixture: str = "loco5ac_ap-ptp"): @patch("airos.airos8._LOGGER") @pytest.mark.asyncio -async def test_status_logs_redacted_data_on_invalid_value(mock_logger, airos_device): +async def test_status_logs_redacted_data_on_invalid_value( + mock_logger: MagicMock, airos_device: AirOS +) -> None: """Test that the status method correctly logs redacted data when it encounters an InvalidFieldValue during deserialization.""" # --- Prepare fake POST /api/auth response with cookies --- cookie = SimpleCookie() @@ -93,7 +102,9 @@ async def test_status_logs_redacted_data_on_invalid_value(mock_logger, airos_dev @patch("airos.airos8._LOGGER") @pytest.mark.asyncio -async def test_status_logs_exception_on_missing_field(mock_logger, airos_device): +async def test_status_logs_exception_on_missing_field( + mock_logger: MagicMock, airos_device: AirOS +) -> None: """Test that the status method correctly logs a full exception when it encounters a MissingField during deserialization.""" # --- Prepare fake POST /api/auth response with cookies --- cookie = SimpleCookie() @@ -143,7 +154,9 @@ async def test_status_logs_exception_on_missing_field(mock_logger, airos_device) ], ) @pytest.mark.asyncio -async def test_ap_object(airos_device, base_url, mode, fixture): +async def test_ap_object( + airos_device: AirOS, base_url: str, mode: str, fixture: str +) -> None: """Test device operation.""" cookie = SimpleCookie() cookie["session_id"] = "test-cookie" @@ -174,12 +187,13 @@ async def test_ap_object(airos_device, base_url, mode, fixture): status: AirOSData = await airos_device.status() # Implies return_json = False # Verify the fixture returns the correct mode + assert status.wireless.mode assert status.wireless.mode.value == mode assert status.derived.mac_interface == "br0" @pytest.mark.asyncio -async def test_reconnect(airos_device, base_url): +async def test_reconnect(airos_device: AirOS, base_url: str) -> None: """Test reconnect client.""" # --- Prepare fake POST /api/stakick response --- mock_stakick_response = MagicMock() @@ -194,13 +208,14 @@ async def test_reconnect(airos_device, base_url): @pytest.mark.asyncio -async def test_ap_corners(airos_device, base_url, mode="ap-ptp"): +async def test_ap_corners( + airos_device: AirOS, base_url: str, mode: str = "ap-ptp" +) -> None: """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 = AsyncMock(return_value="{}") @@ -208,56 +223,60 @@ async def test_ap_corners(airos_device, base_url, mode="ap-ptp"): mock_login_response.cookies = cookie mock_login_response.headers = {"X-CSRF-ID": "test-csrf-token"} + # Test case 1: Successful login with ( patch.object(airos_device.session, "post", return_value=mock_login_response), patch.object(airos_device, "_use_json_for_login_post", return_value=True), ): assert await airos_device.login() + # Test case 2: Login fails with missing cookies (expects an exception) mock_login_response.cookies = {} with ( patch.object(airos_device.session, "post", return_value=mock_login_response), + patch.object(airos_device, "_use_json_for_login_post", return_value=True), + pytest.raises(AirOSConnectionSetupError), ): - try: - assert await airos_device.login() - assert False - except airos.exceptions.AirOSConnectionSetupError: - assert True + # Only call the function; no return value to assert. + await airos_device.login() + # Test case 3: Login successful, returns None due to missing headers mock_login_response.cookies = cookie mock_login_response.headers = {} with ( patch.object(airos_device.session, "post", return_value=mock_login_response), + patch.object(airos_device, "_use_json_for_login_post", return_value=True), ): result = await airos_device.login() - assert result is None + assert result is False + # Test case 4: Login fails with bad data from the API (expects an exception) mock_login_response.headers = {"X-CSRF-ID": "test-csrf-token"} mock_login_response.text = AsyncMock(return_value="abc123") with ( patch.object(airos_device.session, "post", return_value=mock_login_response), + patch.object(airos_device, "_use_json_for_login_post", return_value=True), + pytest.raises(AirOSDataMissingError), ): - try: - assert await airos_device.login() - assert False - except airos.exceptions.AirOSDataMissingError: - assert True + # Only call the function; no return value to assert. + await airos_device.login() + # Test case 5: Login fails due to HTTP status code (expects an exception) mock_login_response.text = AsyncMock(return_value="{}") mock_login_response.status = 400 with ( patch.object(airos_device.session, "post", return_value=mock_login_response), + patch.object(airos_device, "_use_json_for_login_post", return_value=True), + pytest.raises(AirOSConnectionAuthenticationError), ): - try: - assert await airos_device.login() - assert False - except airos.exceptions.AirOSConnectionAuthenticationError: - assert True + # Only call the function; no return value to assert. + await airos_device.login() + # Test case 6: Login fails due to client-level connection error (expects an exception) mock_login_response.status = 200 - with patch.object(airos_device.session, "post", side_effect=aiohttp.ClientError): - try: - assert await airos_device.login() - assert False - except airos.exceptions.AirOSDeviceConnectionError: - assert True + with ( + patch.object(airos_device.session, "post", side_effect=aiohttp.ClientError), + pytest.raises(AirOSDeviceConnectionError), + ): + # Only call the function; no return value to assert. + await airos_device.login() From fdd00e6b91a3eb375c7c3f3d0e81d8e65345cbe3 Mon Sep 17 00:00:00 2001 From: Tom Scholten Date: Tue, 12 Aug 2025 20:19:37 +0200 Subject: [PATCH 2/4] Add discovery tests --- tests/test_discovery.py | 128 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) diff --git a/tests/test_discovery.py b/tests/test_discovery.py index 1857c4f..d550fab 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -398,3 +398,131 @@ async def test_async_discover_devices_generic_oserror( assert "cannot_connect" in str(excinfo.value) close_mock = cast(MagicMock, mock_transport.close) close_mock.assert_not_called() + + +@pytest.mark.asyncio +async def test_parse_airos_packet_short_for_next_tlv() -> None: + """Test parsing stops gracefully after the MAC TLV when no more data exists.""" + protocol = AirOSDiscoveryProtocol(AsyncMock()) + # Header + valid MAC TLV, but then abruptly ends + data_with_fragment = ( + b"\x01\x06\x00\x00\x00\x00" + b"\x06" + bytes.fromhex("0123456789CD") + ) + host_ip = "192.168.1.100" + + with patch("airos.discovery._LOGGER.debug") as mock_log_debug: + parsed_data = protocol.parse_airos_packet(data_with_fragment, host_ip) + + assert parsed_data is not None + assert parsed_data["mac_address"] == "01:23:45:67:89:CD" + # The debug log for the successfully parsed MAC address should be called exactly once. + mock_log_debug.assert_called_once_with( + "Parsed MAC from type 0x06: 01:23:45:67:89:CD" + ) + + +@pytest.mark.asyncio +async def test_parse_airos_packet_truncated_two_byte_tlv() -> None: + """Test parsing with a truncated 2-byte length field TLV.""" + protocol = AirOSDiscoveryProtocol(AsyncMock()) + # Header + valid MAC TLV, then a valid type (0x0a) but a truncated length field + data_with_fragment = ( + b"\x01\x06\x00\x00\x00\x00" + + b"\x06" + + bytes.fromhex("0123456789CD") + + b"\x0a\x00" # TLV type 0x0a, followed by only 1 byte for length (should be 2) + ) + host_ip = "192.168.1.100" + + with patch("airos.discovery._LOGGER.warning") as mock_log_warning: + with pytest.raises(AirOSEndpointError): + protocol.parse_airos_packet(data_with_fragment, host_ip) + + mock_log_warning.assert_called_once() + assert "no 2-byte length field" in mock_log_warning.call_args[0][0] + + +@pytest.mark.asyncio +async def test_parse_airos_packet_malformed_tlv_length() -> None: + """Test parsing with a malformed TLV length field.""" + protocol = AirOSDiscoveryProtocol(AsyncMock()) + # Header + valid MAC TLV, then a valid type (0x02) but a truncated length field + data_with_fragment = ( + b"\x01\x06\x00\x00\x00\x00" + + b"\x06" + + bytes.fromhex("0123456789CD") + + b"\x02\x00" # TLV type 0x02, followed by only 1 byte for length (should be 2) + ) + host_ip = "192.168.1.100" + + with patch("airos.discovery._LOGGER.warning") as mock_log_warning: + with pytest.raises(AirOSEndpointError): + protocol.parse_airos_packet(data_with_fragment, host_ip) + + mock_log_warning.assert_called_once() + assert "no 2-byte length field" in mock_log_warning.call_args[0][0] + + +@pytest.mark.parametrize( + "packet_fragment, unhandled_type", + [ + (b"\x0e\x00\x02\x01\x02", "0xe"), # Unhandled 2-byte length TLV + (b"\x10\x00\x02\x01\x02", "0x10"), # Unhandled 2-byte length TLV + ], +) +@pytest.mark.asyncio +async def test_parse_airos_packet_unhandled_tlv_continues_parsing( + packet_fragment: bytes, unhandled_type: str +) -> None: + """Test that the parser logs an unhandled TLV type but continues parsing the packet.""" + protocol = AirOSDiscoveryProtocol(AsyncMock()) + + # Construct a packet with a valid MAC TLV followed by the unhandled TLV + valid_mac_tlv = b"\x06" + bytes.fromhex("0123456789CD") + base_packet = b"\x01\x06\x00\x00\x00\x00" + + # This new packet structure ensures two TLVs are present + malformed_packet = base_packet + valid_mac_tlv + packet_fragment + host_ip = "192.168.1.100" + + with patch("airos.discovery._LOGGER.debug") as mock_log_debug: + parsed_data = protocol.parse_airos_packet(malformed_packet, host_ip) + + assert parsed_data is not None + assert parsed_data["mac_address"] == "01:23:45:67:89:CD" + + # Now, two debug logs are expected: one for the MAC and one for the unhandled TLV. + assert mock_log_debug.call_count == 2 + + # Check the first log call for the MAC address + assert ( + mock_log_debug.call_args_list[0][0][0] + == "Parsed MAC from type 0x06: 01:23:45:67:89:CD" + ) + + # Check the second log call for the unhandled TLV + log_message = mock_log_debug.call_args_list[1][0][0] + assert f"Unhandled TLV type: {unhandled_type}" in log_message + assert "with length" in log_message + + +@pytest.mark.asyncio +async def test_async_discover_devices_generic_exception( + mock_datagram_endpoint: tuple[asyncio.DatagramTransport, AirOSDiscoveryProtocol], +) -> None: + """Test discovery handles a generic exception during the main execution.""" + mock_transport, _ = mock_datagram_endpoint + + with ( + patch( + "asyncio.sleep", new=AsyncMock(side_effect=Exception("Unexpected error")) + ), + patch("airos.discovery._LOGGER.exception") as mock_log_exception, + pytest.raises(AirOSListenerError) as excinfo, + ): + await airos_discover_devices(timeout=1) + + assert "cannot_connect" in str(excinfo.value) + mock_log_exception.assert_called_once() + close_mock = cast(MagicMock, mock_transport.close) + close_mock.assert_called_once() From 7cbe4bc173e4c5e9d098d4be68d6605f3c8c944f Mon Sep 17 00:00:00 2001 From: Tom Scholten Date: Tue, 12 Aug 2025 20:22:59 +0200 Subject: [PATCH 3/4] mypy actions --- requirements-test.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/requirements-test.txt b/requirements-test.txt index df67449..2c593d7 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -7,4 +7,5 @@ aioresponses aioresponses==0.7.8 aiofiles==24.1.0 radon==6.0.1 -types-aiofiles==24.1.0.20250809 \ No newline at end of file +types-aiofiles==24.1.0.20250809 +mypy==1.17.1 From 58d50bfa0d6462404859e5e64779c52f7bcc28d9 Mon Sep 17 00:00:00 2001 From: Tom Scholten Date: Tue, 12 Aug 2025 20:38:30 +0200 Subject: [PATCH 4/4] Update changelog and bump for release --- CHANGELOG.md | 13 +++++++++++++ pyproject.toml | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eaf8e10..62d2a8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ All notable changes to this project will be documented in this file. +## [0.2.9] - 2025-08-12 + +### Changed + +- Bug Fixes + - More consistent error handling across login, status, stakick, and provmode; login now returns False when an auth token is missing. Improved discovery transport setup and resilience. +- Refactor + - Tightened type hints and clarified method signatures for predictable return types and safer usage. +- Tests + - Substantially expanded coverage, especially for discovery edge cases and error paths. +- Chores + - Enabled type checking in CI and gated coverage on it; added pre-commit hook and supporting environment script; updated test dependencies. + ## [0.2.8] - 2025-08-12 ### Changed diff --git a/pyproject.toml b/pyproject.toml index 2e3f95a..5283b42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "airos" -version = "0.2.9a0" +version = "0.2.9" license = "MIT" description = "Ubiquity airOS module(s) for Python 3." readme = "README.md"