1.1.0 (#379)

* Add tests and improve code (#372)

* Add tests and improve code

* Fix errors

* Update tools module (#373)

* Improve readers module (#374)

* Improve writers module (#375)

* Remove custom logger module (#376)

* Add some tests and code improvements (#377)

* Add RT-AX88U / Merlin test data (#378)

* Bump version to `1.1.0`

.coveragerc

@@ -0,0 +1,5 @@
+[run]
+source = asusrouter
+omit = 
+    # Constants modules
+    asusrouter/modules/endpoint/devicemap_const.py

.github/workflows/ci.yml

@@ -0,0 +1,64 @@
+name: CI
+
+on:
+  push:
+    branches:
+      - main
+      - dev
+  pull_request: ~
+  schedule:
+    - cron: '0 0 * * *'
+
+env:
+  DEFAULT_PYTHON: 3.11
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/[email protected]
+
+      - name: Set up Python ${{ env.DEFAULT_PYTHON }}
+        uses: actions/[email protected]
+        with:
+          python-version: ${{ env.DEFAULT_PYTHON }}
+
+      - name: Get pip cache directory path
+        id: pip-cache
+        run: echo "PIP_CACHE_DIR=$(pip cache dir)" >> $GITHUB_ENV
+
+      - name: Cache pip dependencies
+        uses: actions/[email protected]
+        with:
+          path: ${{ env.PIP_CACHE_DIR }}
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/pyproject.toml') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install project dependencies
+        run: |
+          pip install .
+
+      - name: Install test dependencies
+        run: |
+          pip install -r requirements_test.txt
+
+      - name: Run unit tests
+        run: |
+          pytest --cov=asusrouter --cov-report=xml:unit-tests-cov.xml -k 'not test_devices'
+
+      - name: Run real-data tests
+        run: |
+          pytest --cov=asusrouter --cov-report=xml:real-data-tests-cov.xml tests/test_devices.py --log-cli-level=INFO
+
+      - name: Upload coverage to Codecov
+        uses: codecov/[email protected]
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          fail_ci_if_error: true
+          files: unit-tests-cov.xml,real-data-tests-cov.xml

.github/workflows/release.yml

@@ -23,16 +23,28 @@ jobs:
       with:
         python-version: ${{ env.DEFAULT_PYTHON }}
 
-    - name: Build package
-      shell: bash
+    - name: Cache pip dependencies
+      uses: actions/cache@v3
+      with:
+        path: ~/.cache/pip
+        key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
+        restore-keys: |
+          ${{ runner.os }}-pip-
+
+    - name: Install dependencies
       run: |
+        pip install -r requirements.txt
+        pip install pytest
         pip install build twine
-        python -m build
+
+    - name: Run tests
+      run: pytest
+
+    - name: Build package
+      run: python -m build
 
     - name: Publish package
-      shell: bash
       run: |
         export TWINE_USERNAME="__token__"
         export TWINE_PASSWORD="${{ secrets.TWINE_TOKEN }}"
-
-        twine upload dist/* --skip-existing
+        twine upload dist/* --skip-existing || exit 1

.gitignore

@@ -5,10 +5,12 @@ __pycache__/
 dist/
 *.egg-info
 
+# Pytest
+.coverage
 
 ### Other
 # Test
-test*.py
+local_test*.py
 local/
 
 # VS Code

asusrouter/__main__.py

@@ -9,6 +9,7 @@
 import aiohttp
 
 from asusrouter import AsusRouter, AsusRouterDump, AsusRouterError
+from asusrouter.modules.data import AsusData
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -43,6 +44,13 @@ async def _connect_and_dump(args: argparse.Namespace) -> None:
 
             _LOGGER.debug("Connected and identified")
 
+            _LOGGER.debug("Checking all known data...")
+
+            for datatype in AsusData:
+                await router.async_get_data(datatype)
+
+            _LOGGER.debug("Finished checking all known data")
+
             # Disconnect from the router
             await router.async_disconnect()
             _LOGGER.debug("Disconnected")

asusrouter/asusrouter.py

@@ -269,8 +269,6 @@ async def _check_flags(self) -> None:
         ):
             flags = state_flags.data
 
-        print("FLAGS: ", flags)
-
         if flags.get("reboot", False) is True:
             _LOGGER.debug("Reboot flag is set")
             await self._async_handle_reboot()

asusrouter/const.py

@@ -7,9 +7,13 @@
 class ContentType(str, Enum):
     """Content type enum."""
 
+    UNKNOWN = "unknown"
+
+    BINARY = "application/octet-stream"
+    HTML = "text/html"
     JSON = "application/json"
     TEXT = "text/plain"
-    HTML = "text/html"
+    XML = "application/xml"
 
 
 # Asus constants

asusrouter/modules/connection.py

@@ -4,8 +4,7 @@
 from typing import Optional
 
 from asusrouter.modules.wlan import Wlan
-from asusrouter.tools import get_enum_key_by_value
-from asusrouter.tools.converters import safe_int
+from asusrouter.tools.converters import get_enum_key_by_value, safe_int
 
 
 class ConnectionState(IntEnum):
@@ -69,4 +68,4 @@ def get_connection_type(value: Optional[int]) -> ConnectionType:
     # Check that it's actually an int
     value = safe_int(value) or 0
 
-    return CONNECTION_TYPE.get(value, ConnectionType.WIRED)
+    return get_enum_key_by_value(ConnectionType, value, ConnectionType.WIRED)

asusrouter/modules/endpoint/__init__.py

@@ -11,7 +11,6 @@
 from asusrouter.error import AsusRouter404Error
 from asusrouter.modules.data import AsusData, AsusDataState
 from asusrouter.modules.firmware import Firmware
-from asusrouter.modules.flags import Flag
 from asusrouter.modules.wlan import Wlan
 
 _LOGGER = logging.getLogger(__name__)
@@ -126,7 +125,7 @@ def data_set(data: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
     return data
 
 
-def data_get(data: dict[str, Any], key: str) -> Any:
+def data_get(data: dict[str, Any], key: str) -> Optional[Any]:
     """Extract value from the data dict and update the data dict."""
 
     # Get the value

asusrouter/modules/endpoint/command.py

@@ -7,7 +7,7 @@
 from asusrouter.tools.readers import read_json_content
 
 
-def read(content: str) -> dict[str, Any]:  # pylint: disable=unused-argument
+def read(content: str) -> dict[str, Any]:
     """Read state data"""
 
     # Read the json content

asusrouter/modules/endpoint/devicemap.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import re
 from datetime import datetime, timedelta
 from typing import Any, Optional, Tuple
@@ -18,6 +19,8 @@
 
 from .devicemap_const import DEVICEMAP_BY_INDEX, DEVICEMAP_BY_KEY, DEVICEMAP_CLEAR
 
+_LOGGER = logging.getLogger(__name__)
+
 REQUIRE_HISTORY = True
 
 
@@ -28,8 +31,13 @@ def read(content: str) -> dict[str, Any]:
     devicemap: dict[str, Any] = {}
 
     # Parse the XML data
-    xml_content: dict[str, Any] = xmltodict.parse(content).get("devicemap", {})
-    if not xml_content:
+    try:
+        xml_content: dict[str, Any] = xmltodict.parse(content).get("devicemap", {})
+        if not xml_content:
+            _LOGGER.debug("Received empty devicemap XML")
+            return devicemap
+    except xmltodict.expat.ExpatError as ex:  # type: ignore
+        _LOGGER.debug("Received invalid devicemap XML: %s", ex)
         return devicemap
 
     # Go through the data and fill the dict
@@ -42,6 +50,8 @@ def read(content: str) -> dict[str, Any]:
 
     # Clear values from useless symbols
     for output_group, clear_map in DEVICEMAP_CLEAR.items():
+        if output_group not in devicemap:
+            continue
         for key, clear_value in clear_map.items():
             # If the key is not in the devicemap, continue
             if key not in devicemap[output_group]:
@@ -61,22 +71,30 @@ def read(content: str) -> dict[str, Any]:
     return devicemap
 
 
-# This method performs reading of the devicemap by index
-# to simplify the original read_devicemap method
 def read_index(xml_content: dict[str, Any]) -> dict[str, Any]:
-    """Read devicemap by index."""
+    """Read devicemap by index.
+
+    This method performs reading of the devicemap by index
+    to simplify the original read_devicemap method."""
 
     # Create a dict to store the data
     devicemap: dict[str, Any] = {}
 
     # Get values for which we only know their order (index)
     for output_group, input_group, input_values in DEVICEMAP_BY_INDEX:
-        # Create a dict to store the data
-        output_group_data: dict[str, Any] = {}
+        # Create an empty dictionary for the output group
+        devicemap[output_group] = {}
 
-        # Go through the input values and fill the dict
-        for index, input_value in enumerate(input_values):
-            output_group_data[input_value] = xml_content[input_group][index]
+        # Check that the input group is in the xml content
+        if input_group not in xml_content:
+            continue
+
+        # Use dict comprehension to build output_group_data
+        output_group_data = {
+            input_value: xml_content[input_group][index]
+            for index, input_value in enumerate(input_values)
+            if index < len(xml_content[input_group])
+        }
 
         # Add the output group data to the devicemap
         devicemap[output_group] = output_group_data
@@ -85,10 +103,11 @@ def read_index(xml_content: dict[str, Any]) -> dict[str, Any]:
     return devicemap
 
 
-# This method performs reading of the devicemap by key
-# to simplify the original read_devicemap method
 def read_key(xml_content: dict[str, Any]) -> dict[str, Any]:
-    """Read devicemap by key."""
+    """Read devicemap by key.
+
+    This method performs reading of the devicemap by key
+    to simplify the original read_devicemap method."""
 
     # Create a dict to store the data
     devicemap: dict[str, Any] = {}
@@ -100,27 +119,25 @@ def read_key(xml_content: dict[str, Any]) -> dict[str, Any]:
 
         # Go through the input values and fill the dict
         for input_value in input_values:
-            # Check if the input group is a string
-            if isinstance(xml_content.get(input_group), str):
-                # Check if the input value is in the input group
-                if input_value in xml_content[input_group]:
-                    # Add the input value to the output group data and remove the key
-                    output_group_data[input_value] = xml_content[input_group].replace(
+            # Get the input group data
+            xml_input_group = xml_content.get(input_group)
+
+            # If the input group data is None, skip this iteration
+            if xml_input_group is None:
+                continue
+
+            # If the input group data is a string, convert it to a list
+            if isinstance(xml_input_group, str):
+                xml_input_group = [xml_input_group]
+
+            # Go through the input group data and check if the input value is in it
+            for value in xml_input_group:
+                if input_value in value:
+                    # Add the input value to the output group data
+                    output_group_data[input_value] = value.replace(
                         f"{input_value}=", ""
                     )
-            # Check if the input group is a list
-            else:
-                # Go through the input group and check if the input value is in it
-                xml_input_group = xml_content.get(input_group)
-                if not xml_input_group:
-                    continue
-                for value in xml_input_group:
-                    if input_value in value:
-                        # Add the input value to the output group data
-                        output_group_data[input_value] = value.replace(
-                            f"{input_value}=", ""
-                        )
-                        break
+                    break
 
         # Add the output group data to the devicemap
         devicemap[output_group] = output_group_data
@@ -142,17 +159,24 @@ def read_special(xml_content: dict[str, Any]) -> dict[str, Any]:
 def read_uptime_string(content: str) -> datetime | None:
     """Read uptime string and return proper datetime object."""
 
+    # Split the content into the date/time part and the seconds part
+    uptime_parts = content.split("(")
+    if len(uptime_parts) < 2:
+        return None
+
+    # Extract the number of seconds from the seconds part
+    seconds_match = re.search("([0-9]+)", uptime_parts[1])
+    if not seconds_match:
+        return None
+
     try:
-        part = content.split("(")
-        match = re.search("([0-9]+)", part[1])
-        if not match:
-            return None
-        seconds = int(match.group())
-        when = dtparse(part[0])
-        uptime = when - timedelta(seconds=seconds)
+        seconds = int(seconds_match.group())
+        when = dtparse(uptime_parts[0])
     except ValueError:
         return None
 
+    uptime = when - timedelta(seconds=seconds)
+
     return uptime
 
 

asusrouter/modules/endpoint/ethernet_ports.py

@@ -25,7 +25,7 @@ def process(data: dict[str, Any]) -> dict[AsusData, Any]:
     """Process ethernet ports data."""
 
     # Ports info
-    ports = {
+    ports: dict[PortType, dict] = {
         PortType.LAN: {},
         PortType.WAN: {},
     }