test: optimize and de-brittle test suite

aiopnsense/__init__.py

@@ -5,8 +5,8 @@
     OPNsenseBelowMinFirmware,
     OPNsenseConnectionError,
     OPNsenseError,
-    OPNsenseInvalidAuth,
     OPNsenseInvalidArgument,
+    OPNsenseInvalidAuth,
     OPNsenseInvalidURL,
     OPNsenseMissingDeviceUniqueID,
     OPNsensePrivilegeMissing,
@@ -21,8 +21,8 @@
     "OPNsenseClient",
     "OPNsenseConnectionError",
     "OPNsenseError",
-    "OPNsenseInvalidAuth",
     "OPNsenseInvalidArgument",
+    "OPNsenseInvalidAuth",
     "OPNsenseInvalidURL",
     "OPNsenseMissingDeviceUniqueID",
     "OPNsensePrivilegeMissing",

aiopnsense/client.py

@@ -9,11 +9,7 @@
 from .client_base import ClientBaseMixin
 from .const import OPNSENSE_LTD_FIRMWARE, OPNSENSE_MIN_FIRMWARE
 from .dhcp import DHCPMixin
-from .exceptions import (
-    OPNsenseBelowMinFirmware,
-    OPNsenseUnknownFirmware,
-    _map_opnsense_exception,
-)
+from .exceptions import OPNsenseBelowMinFirmware, OPNsenseUnknownFirmware, _map_opnsense_exception
 from .firewall import FirewallMixin
 from .firmware import FirmwareMixin
 from .helpers import _LOGGER, firmware_is_at_least

aiopnsense/client_transport.py

@@ -210,8 +210,7 @@ def _drain_buffer(
                             if not line.startswith("data:"):
                                 continue
                             value = line[len("data:") :]
-                            if value.startswith(" "):
-                                value = value[1:]
+                            value = value.removeprefix(" ")
                             data_lines.append(value)
 
                         if not data_lines:

aiopnsense/dhcp.py

@@ -544,10 +544,10 @@ async def _get_isc_dhcpv4_leases(self, opnsense_tz: tzinfo | None = None) -> lis
                 try:
                     dt: datetime = datetime.strptime(
                         lease_info.get("ends", None), "%Y/%m/%d %H:%M:%S"
-                    )
+                    ).replace(tzinfo=opnsense_tz)
                 except TypeError, ValueError:
                     continue
-                lease["expires"] = dt.replace(tzinfo=opnsense_tz)
+                lease["expires"] = dt
                 if lease["expires"] < current_time:
                     continue
             else:
@@ -604,10 +604,10 @@ async def _get_isc_dhcpv6_leases(self, opnsense_tz: tzinfo | None = None) -> lis
                 try:
                     dt: datetime = datetime.strptime(
                         lease_info.get("ends", None), "%Y/%m/%d %H:%M:%S"
-                    )
+                    ).replace(tzinfo=opnsense_tz)
                 except TypeError, ValueError:
                     continue
-                lease["expires"] = dt.replace(tzinfo=opnsense_tz)
+                lease["expires"] = dt
                 if lease["expires"] < current_time:
                     continue
             else:

aiopnsense/firewall.py

@@ -375,10 +375,7 @@ async def toggle_firewall_rule(self, uuid: str, toggle_on_off: str | None = None
             return False
 
         apply_resp = await self._safe_dict_post(FIREWALL_FILTER_APPLY_ENDPOINT)
-        if apply_resp.get("status", "").strip() != "OK":
-            return False
-
-        return True
+        return apply_resp.get("status", "").strip() == "OK"
 
     async def toggle_nat_rule(
         self, nat_rule_type: str, uuid: str, toggle_on_off: str | None = None
@@ -425,10 +422,7 @@ async def toggle_nat_rule(
         apply_resp = await self._safe_dict_post(
             f"{FIREWALL_NAT_TOGGLE_RULE_ENDPOINT_PREFIX}{nat_rule_type}{FIREWALL_NAT_APPLY_ENDPOINT_SUFFIX}"
         )
-        if apply_resp.get("status", "").strip() != "OK":
-            return False
-
-        return True
+        return apply_resp.get("status", "").strip() == "OK"
 
     async def kill_states(self, ip_addr: str) -> MutableMapping[str, Any]:
         """Kill the active states of the IP address.
@@ -513,7 +507,4 @@ async def toggle_alias(self, alias: str, toggle_on_off: str | None = None) -> bo
             return False
 
         reconfigure_resp = await self._safe_dict_post(FIREWALL_ALIAS_RECONFIGURE_ENDPOINT)
-        if reconfigure_resp.get("status") != "ok":
-            return False
-
-        return True
+        return reconfigure_resp.get("status") == "ok"

aiopnsense/nut.py

@@ -1,7 +1,7 @@
 """NUT plugin methods for OPNsenseClient."""
 
-import re
 from collections.abc import Mapping
+import re
 from typing import Any
 
 from ._typing import AiopnsenseClientProtocol

aiopnsense/system.py

@@ -9,11 +9,8 @@
 from dateutil.parser import ParserError, UnknownTimezoneWarning, parse
 
 from ._typing import AiopnsenseClientProtocol
-from .const import (
-    AMBIGUOUS_TZINFOS,
-    OPNSENSE_26_1_11_COMPAT_FIRMWARE,
-)
-from .exceptions import OPNsenseMissingDeviceUniqueID, OPNsenseError
+from .const import AMBIGUOUS_TZINFOS, OPNSENSE_26_1_11_COMPAT_FIRMWARE
+from .exceptions import OPNsenseError, OPNsenseMissingDeviceUniqueID
 from .helpers import (
     _LOGGER,
     _log_errors,
@@ -448,7 +445,7 @@ async def get_device_unique_id(self, expected_id: str | None = None) -> str | No
             )
             return expected_id
 
-        device_unique_id = sorted(mac_addresses)[0]
+        device_unique_id = min(mac_addresses)
         _LOGGER.debug("[get_device_unique_id] device_unique_id (first): %s", device_unique_id)
         return device_unique_id
 
@@ -619,9 +616,7 @@ async def system_reboot(self) -> bool:
         """
         response = await self._safe_dict_post(CORE_SYSTEM_REBOOT_ENDPOINT)
         _LOGGER.debug("[system_reboot] response: %s", response)
-        if response.get("status", "") == "ok":
-            return True
-        return False
+        return response.get("status", "") == "ok"
 
     @_log_errors
     async def system_halt(self) -> None:
@@ -677,9 +672,7 @@ async def send_wol(self, interface: str, mac: str) -> bool:
         _LOGGER.debug("[send_wol] payload: %s", payload)
         response = await self._safe_dict_post(WOL_SET_ENDPOINT, payload)
         _LOGGER.debug("[send_wol] response: %s", response)
-        if response.get("status", "") == "ok":
-            return True
-        return False
+        return response.get("status", "") == "ok"
 
     @_log_errors
     async def get_notices(self) -> dict[str, Any]:

aiopnsense/traffic.py

@@ -5,10 +5,10 @@
 from collections.abc import AsyncIterator, Mapping
 from typing import Any
 
+from ._typing import AiopnsenseClientProtocol
 from .client_transport import _STREAM_JSON_EVENT_RESET_KEY
 from .const import DEFAULT_REQUEST_TIMEOUT_SECONDS
 from .exceptions import OPNsenseError, _map_opnsense_exception
-from ._typing import AiopnsenseClientProtocol
 from .helpers import _LOGGER, try_to_float, try_to_int
 
 DIAGNOSTICS_TRAFFIC_ENDPOINT = "/api/diagnostics/traffic/interface"

aiopnsense/vnstat.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, MutableMapping, Sequence
-from datetime import date, datetime, timedelta, tzinfo
+from datetime import UTC, date, datetime, timedelta, tzinfo
 import re
 from typing import Any
 
@@ -291,7 +291,7 @@ def _to_bytes(self, value: str, unit: str) -> int | None:
         factor = _BYTE_FACTORS.get(unit.upper())
         if parsed_value is None or factor is None:
             return None
-        return int(round(parsed_value * factor))
+        return round(parsed_value * factor)
 
     def _to_bits_per_second(self, value: str, unit: str) -> int | None:
         """Convert vnStat rate strings into integer bits-per-second.
@@ -309,7 +309,7 @@ def _to_bits_per_second(self, value: str, unit: str) -> int | None:
         factor = _RATE_FACTORS.get(unit.upper())
         if parsed_value is None or factor is None:
             return None
-        return int(round(parsed_value * factor))
+        return round(parsed_value * factor)
 
     def _pick_daily_row(
         self,
@@ -488,7 +488,7 @@ def _parse_daily_label(self, label: Any) -> date | None:
             return None
         for fmt in ("%m/%d/%y", "%Y-%m-%d"):
             try:
-                return datetime.strptime(label, fmt).date()
+                return datetime.strptime(label, fmt).replace(tzinfo=UTC).date()
             except ValueError:
                 continue
         return None
@@ -508,7 +508,7 @@ def _parse_month_label(self, label: Any) -> tuple[int, int] | None:
             return None
         for fmt in ("%Y-%m", "%b '%y", "%B '%y"):
             try:
-                parsed = datetime.strptime(label, fmt)
+                parsed = datetime.strptime(label, fmt).replace(tzinfo=UTC)
             except ValueError:
                 continue
             else:

docs/source/_ext/opnsense_client_api.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from importlib import import_module
 import inspect
-from typing import Any
+from typing import Any, ClassVar
 
 from docutils import nodes
 from docutils.parsers.rst import Directive, directives
@@ -59,7 +60,7 @@ class OPNsenseClientAPIDirective(Directive):
     required_arguments = 1
     optional_arguments = 0
     final_argument_whitespace = False
-    option_spec = {
+    option_spec: ClassVar[dict[str, Callable[[str], str]]] = {
         "client": directives.unchanged_required,
     }
 

docs/source/conf.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import UTC, datetime
 import inspect
 import logging
 from pathlib import Path
@@ -21,7 +21,7 @@
 
 project: str = "aiopnsense"
 author: str = "Snuffy2"
-copyright: str = f"{datetime.now():%Y}, {author}"
+copyright: str = f"{datetime.now(UTC):%Y}, {author}"
 
 extensions: list[str] = [
     "sphinx.ext.autodoc",

scripts/_opnsense_live_common.py

@@ -2,18 +2,19 @@
 
 from __future__ import annotations
 
-import json
-import os
-import re
 from collections.abc import Mapping
 from dataclasses import dataclass
 from functools import cache
+import json
+import os
 from pathlib import Path
+import re
 import sys
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     import aiohttp
+
     from aiopnsense import OPNsenseClient
 
 
@@ -255,7 +256,7 @@ def write_output(payload: Any, output_path: Path | None = None) -> None:
 
 
 @cache
-def _get_client_class() -> type["OPNsenseClient"]:
+def _get_client_class() -> type[OPNsenseClient]:
     """Return the lazily imported OPNsense client class.
 
     Returns:
@@ -266,7 +267,7 @@ def _get_client_class() -> type["OPNsenseClient"]:
     return OPNsenseClient
 
 
-def create_client(config: LiveConfig, session: aiohttp.ClientSession) -> "OPNsenseClient":
+def create_client(config: LiveConfig, session: aiohttp.ClientSession) -> OPNsenseClient:
     """Create an OPNsense client for live scripts.
 
     Args:

scripts/aiopnsense_dump.py

@@ -5,11 +5,11 @@
 
 import argparse
 import asyncio
+from dataclasses import dataclass
 import importlib
 import logging
-import time
-from dataclasses import dataclass
 from pathlib import Path
+import time
 from typing import Any
 
 _common = importlib.import_module("_opnsense_live_common")
@@ -25,8 +25,9 @@
 if __name__ == "__main__":
     reexec_with_repo_venv(Path(__file__))
 
-import aiohttp  # noqa: E402
-from aiopnsense.exceptions import OPNsenseError  # noqa: E402
+import aiohttp
+
+from aiopnsense.exceptions import OPNsenseError
 
 _LOGGER = logging.getLogger(__name__)
 

scripts/opnsense_api_call.py

@@ -27,7 +27,7 @@
 if __name__ == "__main__":
     reexec_with_repo_venv(Path(__file__))
 
-import aiohttp  # noqa: E402
+import aiohttp
 
 _NO_PAYLOAD = object()
 
@@ -121,14 +121,16 @@ def _load_json_object(payload_text: str, source: str) -> dict[str, Any]:
 
 
     Raises:
-        ValueError: If the payload is invalid JSON or does not decode to an object.
+        ValueError: If the payload is invalid JSON or does not decode to an
+            object.
     """
     try:
         value = json.loads(payload_text)
     except json.JSONDecodeError as err:
         raise ValueError(f"Invalid JSON in {source}: {err}") from err
     if not isinstance(value, dict):
-        raise ValueError(f"{source} must be a JSON object")
+        # ValueError is the CLI's documented payload-validation contract.
+        raise ValueError(f"{source} must be a JSON object")  # noqa: TRY004
     return value
 
 
@@ -164,7 +166,7 @@ def load_payload(
 
 async def call_api(
     session: aiohttp.ClientSession,
-    config: "LiveConfig",
+    config: LiveConfig,
     endpoint: str,
     method: str,
     payload: dict[str, Any] | object,