Improve tests and coverage

Author: CoMPaTech

.gitignore

@@ -11,3 +11,4 @@ tests/__pycache__
 .coverage
 tmp
 todo
+.DS_Store

airos/discovery.py

@@ -175,14 +175,13 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                     offset += 2
 
                     if tlv_length > (len(data) - offset):
-                        log = f"TLV type {tlv_type:#x} length {tlv_length} exceeds remaining data "
-                        _LOGGER.warning(log)
-                        log = f"({len(data) - offset} bytes left). Packet malformed. "
-                        _LOGGER.warning(log)
-                        log = f"Data from TLV start: {data[offset - 3 :].hex()}"
+                        log = (
+                            f"TLV type {tlv_type:#x} length {tlv_length} exceeds remaining data "
+                            f"({len(data) - offset} bytes left). Packet malformed. "
+                            f"Data from TLV start: {data[offset - 3 :].hex()}"
+                        )
                         _LOGGER.warning(log)
-                        log = f"Malformed packet: {log}"
-                        raise AirOSEndpointError(log)
+                        raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     tlv_value: bytes = data[offset : offset + tlv_length]
 
@@ -195,6 +194,7 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                         else:
                             log = f"Unexpected length for 0x02 TLV (MAC+IP). Expected 10, got {tlv_length}. Value: {tlv_value.hex()}"
                             _LOGGER.warning(log)
+                            raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     elif tlv_type == 0x03:
                         parsed_info["firmware_version"] = tlv_value.decode(
@@ -213,6 +213,7 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                         else:
                             log = f"Unexpected length for Uptime (Type 0x0A): {tlv_length}. Value: {tlv_value.hex()}"
                             _LOGGER.warning(log)
+                            raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     elif tlv_type == 0x0B:
                         parsed_info["hostname"] = tlv_value.decode(

tests/test_airos8.py

@@ -8,6 +8,7 @@
 import pytest
 
 import aiohttp
+from mashumaro.exceptions import MissingField
 
 
 # --- Tests for Login and Connection Errors ---
@@ -224,3 +225,36 @@ async def test_provmode_connection_error(airos_device):
         pytest.raises(airos.exceptions.AirOSDeviceConnectionError),
     ):
         await airos_device.provmode(active=True)
+
+
+@pytest.mark.asyncio
+async def test_status_missing_required_key_in_json(airos_device):
+    """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,
+    # which is a required field for the AirOS8Data dataclass.
+    invalid_data = {
+        "host": {"hostname": "test"},
+        "interfaces": [
+            {"ifname": "br0", "hwaddr": "11:22:33:44:55:66", "enabled": True}
+        ],
+    }
+
+    mock_status_response = MagicMock()
+    mock_status_response.__aenter__.return_value = mock_status_response
+    mock_status_response.text = AsyncMock(return_value=json.dumps(invalid_data))
+    mock_status_response.status = 200
+
+    with (
+        patch.object(airos_device.session, "get", return_value=mock_status_response),
+        patch("airos.airos8._LOGGER.exception") as mock_log_exception,
+        pytest.raises(airos.exceptions.AirOSKeyDataMissingError) as excinfo,
+    ):
+        await airos_device.status()
+
+    # Check that the specific mashumaro error is logged and caught
+    mock_log_exception.assert_called_once()
+    assert "Failed to deserialize AirOS data" in mock_log_exception.call_args[0][0]
+    # --- MODIFICATION START ---
+    # Assert that the cause of our exception is the correct type from mashumaro
+    assert isinstance(excinfo.value.__cause__, MissingField)

tests/test_data.py

@@ -0,0 +1,54 @@
+"""Tests for airos data module."""
+
+from unittest.mock import patch
+
+from airos.data import Host, Wireless
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_unknown_enum_values():
+    """Test that unknown enum values are handled gracefully."""
+    # 1. Test for Host.netrole
+    host_data = {"netrole": "unsupported_role", "other_field": "value"}
+    format_string = (
+        "Unknown value '%s' for %s.%s. Please report at "
+        "https://github.com/CoMPaTech/python-airos/issues so we can add support."
+    )
+    with patch("airos.data.logger.warning") as mock_warning:
+        processed_host = Host.__pre_deserialize__(host_data.copy())
+        # Verify the unknown value was removed
+        assert "netrole" not in processed_host
+        # Verify the other fields remain
+        assert "other_field" in processed_host
+        # Verify a warning was logged
+        mock_warning.assert_called_once_with(
+            format_string, "unsupported_role", "Host", "netrole"
+        )
+
+    # 2. Test for Wireless (all enums)
+    wireless_data = {
+        "mode": "unsupported_mode",
+        "ieeemode": "unsupported_ieee",
+        "security": "unsupported_security",
+        "other_field": "value",
+    }
+    with patch("airos.data.logger.warning") as mock_warning:
+        processed_wireless = Wireless.__pre_deserialize__(wireless_data.copy())
+        # Verify the unknown values were removed
+        assert "mode" not in processed_wireless
+        assert "ieeemode" not in processed_wireless
+        assert "security" not in processed_wireless
+        # Verify the other field remains
+        assert "other_field" in processed_wireless
+        # Verify warnings were logged for each unknown enum
+        assert mock_warning.call_count == 3
+        mock_warning.assert_any_call(
+            format_string, "unsupported_mode", "Wireless", "mode"
+        )
+        mock_warning.assert_any_call(
+            format_string, "unsupported_ieee", "Wireless", "ieeemode"
+        )
+        mock_warning.assert_any_call(
+            format_string, "unsupported_security", "Wireless", "security"
+        )

tests/test_discovery.py

@@ -304,3 +304,76 @@ async def test_async_discover_devices_cancelled(mock_datagram_endpoint):
 
     assert "cannot_connect" in str(excinfo.value)
     mock_transport.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_datagram_received_handles_general_exception():
+    """Test datagram_received handles a generic exception during parsing."""
+    mock_callback = AsyncMock()
+    protocol = AirOSDiscoveryProtocol(mock_callback)
+    some_data = b"\x01\x06\x00\x00\x00\x00"
+    host_ip = "192.168.1.100"
+
+    with (
+        patch.object(
+            protocol, "parse_airos_packet", side_effect=ValueError("A generic error")
+        ) as mock_parse,
+        patch("airos.discovery._LOGGER.exception") as mock_log_exception,
+    ):
+        # A generic exception should be caught and re-raised as AirOSDiscoveryError
+        with pytest.raises(AirOSDiscoveryError):
+            protocol.datagram_received(some_data, (host_ip, DISCOVERY_PORT))
+
+        mock_parse.assert_called_once_with(some_data, host_ip)
+        mock_callback.assert_not_called()
+        mock_log_exception.assert_called_once()
+        assert (
+            "Error processing AirOS discovery packet"
+            in mock_log_exception.call_args[0][0]
+        )
+
+
+@pytest.mark.parametrize(
+    "packet_fragment, error_message",
+    [
+        # Case 1: TLV type 0x0A (Uptime) with wrong length
+        (b"\x0a\x00\x02\x01\x02", "Unexpected length for Uptime (Type 0x0A)"),
+        # Case 2: TLV declared length exceeds remaining packet data
+        (b"\x0c\x00\xff\x41\x42", "length 255 exceeds remaining data"),
+        # Case 3: An unknown TLV type
+        (b"\xff\x01\x02", "Unhandled TLV type: 0xff"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_parse_airos_packet_tlv_edge_cases(packet_fragment, error_message):
+    """Test parsing of various malformed TLV entries."""
+    protocol = AirOSDiscoveryProtocol(AsyncMock())
+    # A valid header is required to get to the TLV parsing stage
+    base_packet = b"\x01\x06\x00\x00\x00\x00"
+    malformed_packet = base_packet + packet_fragment
+    host_ip = "192.168.1.100"
+
+    with pytest.raises(AirOSEndpointError) as excinfo:
+        protocol.parse_airos_packet(malformed_packet, host_ip)
+
+    assert error_message in str(excinfo.value)
+
+
+@pytest.mark.asyncio
+async def test_async_discover_devices_generic_oserror(mock_datagram_endpoint):
+    """Test discovery handles a generic OSError during endpoint creation."""
+    mock_transport, _ = mock_datagram_endpoint
+
+    with (
+        patch("asyncio.get_running_loop") as mock_get_loop,
+        pytest.raises(AirOSEndpointError) as excinfo,
+    ):
+        mock_loop = mock_get_loop.return_value
+        # Simulate an OSError that is NOT 'address in use'
+        mock_loop.create_datagram_endpoint = AsyncMock(
+            side_effect=OSError(13, "Permission denied")
+        )
+        await async_discover_devices(timeout=1)
+
+    assert "cannot_connect" in str(excinfo.value)
+    mock_transport.close.assert_not_called()