diff --git a/.github/workflows/codechecker.yml b/.github/workflows/codechecker.yml index 5f3fa95..87a4a69 100644 --- a/.github/workflows/codechecker.yml +++ b/.github/workflows/codechecker.yml @@ -26,7 +26,9 @@ jobs: python-version: "3.12" - name: Install ruff - run: pip install ruff + # Pinned so a new ruff release can't turn CI red on an unrelated PR. + # Bump together with the rules in pyproject.toml. + run: pip install "ruff==0.16.6" - name: Auto-fix and format Python code run: | diff --git a/custom_components/pfsense/__init__.py b/custom_components/pfsense/__init__.py index 2f0bd37..416d0af 100644 --- a/custom_components/pfsense/__init__.py +++ b/custom_components/pfsense/__init__.py @@ -1,32 +1,28 @@ -"""Support for pfSense REST API""" +"""Support for the pfSense REST API integration.""" from __future__ import annotations import asyncio +from collections.abc import Callable import copy from datetime import timedelta import logging import re import time -from typing import Callable from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ( - CONF_SCAN_INTERVAL, - CONF_URL, - CONF_VERIFY_SSL, -) +from homeassistant.const import CONF_SCAN_INTERVAL, CONF_URL, CONF_VERIFY_SSL from homeassistant.core import HomeAssistant, callback -from homeassistant.exceptions import ConfigEntryAuthFailed +from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.restore_state import RestoreEntity +from homeassistant.helpers.storage import Store from homeassistant.helpers.update_coordinator import ( CoordinatorEntity, DataUpdateCoordinator, UpdateFailed, ) -from homeassistant.helpers.storage import Store from .const import ( CONF_API_KEY, @@ -47,8 +43,7 @@ SHOULD_RELOAD, UNDO_UPDATE_LISTENER, ) -from .pypfsense import Client as pfSenseClient -from .pypfsense import PfSenseAuthError, PfSensePrivilegeError +from .pypfsense import Client as pfSenseClient, PfSenseAuthError, PfSensePrivilegeError from .services import ServiceRegistrar _LOGGER = logging.getLogger(__name__) @@ -62,8 +57,8 @@ async def async_save_cache(hass: HomeAssistant, entry_id: str, data: dict): store = Store(hass, STORAGE_VERSION, f"{DOMAIN}_{entry_id}_cache") try: await store.async_save(data) - except Exception as e: - _LOGGER.error(f"Failed to save pfSense cache: {e}") + except (OSError, HomeAssistantError, ValueError) as err: + _LOGGER.error("Failed to save pfSense cache: %s", err) async def async_load_cache(hass: HomeAssistant, entry_id: str): @@ -71,8 +66,8 @@ async def async_load_cache(hass: HomeAssistant, entry_id: str): store = Store(hass, STORAGE_VERSION, f"{DOMAIN}_{entry_id}_cache") try: return await store.async_load() - except Exception as e: - _LOGGER.error(f"Failed to load pfSense cache: {e}") + except (OSError, HomeAssistantError, ValueError) as err: + _LOGGER.error("Failed to load pfSense cache: %s", err) return None @@ -80,15 +75,14 @@ async def async_load_cache(hass: HomeAssistant, entry_id: str): def dict_get(data: dict, path: str, default=None): - pathList = re.split(r"\.", path, flags=re.IGNORECASE) + """Traverse a nested dict/list by a dotted path; numeric segments index lists.""" result = data - for key in pathList: - try: + try: + for key in re.split(r"\.", path, flags=re.IGNORECASE): key = int(key) if key.isnumeric() else key result = result[key] - except Exception: - result = default - break + except (KeyError, IndexError, TypeError): + return default return result @@ -124,11 +118,11 @@ def _compute_interface_rates(new_state, elapsed_time, scan_interval): label, value = "kilobytes_per_second", rate / 1000 new_property = f"{prop}_{label}" if elapsed_time >= scan_interval: - interface[new_property] = int(round(value)) + interface[new_property] = round(value) else: previous_value = previous_interface.get(new_property) - interface[new_property] = int( - round(previous_value if previous_value is not None else value) + interface[new_property] = round( + previous_value if previous_value is not None else value ) @@ -144,7 +138,7 @@ def _compute_openvpn_rates(new_state, elapsed_time): for prop in ("total_bytes_recv", "total_bytes_sent"): change = abs(server.get(prop, 0) - previous_server.get(prop, 0)) rate = change / elapsed_time if elapsed_time > 0 else 0 - server[f"{prop}_kilobytes_per_second"] = int(round(rate / 1000)) + server[f"{prop}_kilobytes_per_second"] = round(rate / 1000) async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry): @@ -177,24 +171,26 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry): async def async_update_data(): """Fetch data from pfSense, falling back to the on-disk cache on failure.""" + new_state = None try: async with asyncio.timeout(scan_interval - 1): new_state = await data.update() - if not new_state: - raise UpdateFailed("no data received from pfSense") - await async_save_cache(hass, entry.entry_id, new_state) - return new_state except (PfSenseAuthError, PfSensePrivilegeError) as err: raise ConfigEntryAuthFailed(str(err)) from err - except Exception as err: + except Exception: _LOGGER.warning( - "pfSense poll failed (%s); trying the local cache", err + "pfSense poll failed; trying the local cache", exc_info=True ) - cached_data = await async_load_cache(hass, entry.entry_id) - if cached_data: - data._state = cached_data - return cached_data - raise UpdateFailed(f"poll failed and no cache available: {err}") + else: + if new_state: + await async_save_cache(hass, entry.entry_id, new_state) + return new_state + + cached_data = await async_load_cache(hass, entry.entry_id) + if cached_data: + data.restore_state(cached_data) + return cached_data + raise UpdateFailed("pfSense poll failed and no usable cache is available") coordinator = DataUpdateCoordinator( hass, @@ -216,21 +212,23 @@ async def async_update_data(): async def async_update_device_tracker_data(): """Fetch the ARP table from pfSense.""" + new_dt_state = None try: async with asyncio.timeout(device_tracker_scan_interval - 1): new_dt_state = await device_tracker_data.update( {"scope": "device_tracker"} ) - if not new_dt_state: - raise UpdateFailed("no device tracker data received") - return new_dt_state except (PfSenseAuthError, PfSensePrivilegeError) as err: raise ConfigEntryAuthFailed(str(err)) from err - except Exception as err: - _LOGGER.warning("pfSense device tracker update failed: %s", err) - if device_tracker_data._state: - return device_tracker_data._state - raise UpdateFailed(err) + except Exception: + _LOGGER.warning("pfSense device tracker update failed", exc_info=True) + else: + if new_dt_state: + return new_dt_state + + if device_tracker_data.state: + return device_tracker_data.state + raise UpdateFailed("pfSense device tracker update failed") device_tracker_coordinator = DataUpdateCoordinator( hass, @@ -303,6 +301,8 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> class PfSenseData: + """Fetches and holds the pfSense poll state for a config entry.""" + def __init__( self, client: pfSenseClient, config_entry: ConfigEntry, hass: HomeAssistant ): @@ -315,8 +315,13 @@ def __init__( @property def state(self): + """Return the most recently fetched (or restored) poll state.""" return self._state + def restore_state(self, state: dict) -> None: + """Adopt a state dict loaded from the on-disk cache.""" + self._state = state + async def update(self, opts=None): """Fetch the latest state from pfSense over the REST API.""" opts = opts or {} @@ -420,6 +425,7 @@ def __init__( process_entities_callback: Callable, async_add_entities: AddEntitiesCallback, ) -> None: + """Initialize the data holder.""" self.hass = hass self.coordinator = coordinator self.config_entry = config_entry @@ -434,6 +440,7 @@ def __init__( @callback def process_entities(self): + """Build entities from the current coordinator data and add new ones.""" entities = self.process_entities_callback(self.hass, self.config_entry) new_entities = [] @@ -447,14 +454,16 @@ def process_entities(self): class PfSenseEntity(CoordinatorEntity, RestoreEntity): - """base entity for pfSense""" + """Base entity for pfSense.""" @property def coordinator_context(self): + """Return the coordinator context.""" return None @property def device_info(self): + """Return device registry information.""" state = self.coordinator.data if not state or "host_firmware_version" not in state: return None @@ -470,12 +479,14 @@ def device_info(self): @property def pfsense_device_name(self): + """Return the pfsense device name.""" if self.config_entry.title: return self.config_entry.title return f"{self._get_pfsense_state_value('system_info.hostname')}.{self._get_pfsense_state_value('system_info.domain')}" @property def pfsense_device_unique_id(self): + """Return the pfsense device unique id.""" return self._get_pfsense_state_value("system_info.netgate_device_id") def _get_pfsense_state_value(self, path, default=None): @@ -487,19 +498,22 @@ def _get_pfsense_client(self) -> pfSenseClient: async def service_start_service( self, service_name: str, service: dict | str | None = None ): + """Handle the pfsense.start_service service call.""" await self._get_pfsense_client().start_service(service_name, service) async def service_stop_service( self, service_name: str, service: dict | str | None = None ): + """Handle the pfsense.stop_service service call.""" await self._get_pfsense_client().stop_service(service_name, service) async def service_restart_service( self, service_name: str, - only_if_running: int | str | None | bool = False, + only_if_running: int | str | bool | None = False, service: dict | str | None = None, ): + """Handle the pfsense.restart_service service call.""" client = self._get_pfsense_client() if str(only_if_running).lower() in ["true", "1"]: await client.restart_service_if_running(service_name, service) @@ -507,22 +521,29 @@ async def service_restart_service( await client.restart_service(service_name, service) async def service_reset_state_table(self): + """Handle the pfsense.reset_state_table service call.""" await self._get_pfsense_client().reset_state_table() - async def service_kill_states(self, source: str, destination: str = None): + async def service_kill_states(self, source: str, destination: str | None = None): + """Handle the pfsense.kill_states service call.""" await self._get_pfsense_client().kill_states(source, destination) async def service_system_halt(self): + """Handle the pfsense.system_halt service call.""" await self._get_pfsense_client().system_halt() async def service_system_reboot(self): + """Handle the pfsense.system_reboot service call.""" await self._get_pfsense_client().system_reboot() async def service_send_wol(self, interface: str, mac: str): + """Handle the pfsense.send_wol service call.""" await self._get_pfsense_client().send_wol(interface, mac) async def service_set_default_gateway(self, gateway: str, ip_version: str): + """Handle the pfsense.set_default_gateway service call.""" await self._get_pfsense_client().set_default_gateway(gateway, ip_version) async def service_exec_command(self, command: str, background: bool = False): + """Handle the pfsense.exec_command service call.""" await self._get_pfsense_client().exec_command(command, background) diff --git a/custom_components/pfsense/binary_sensor.py b/custom_components/pfsense/binary_sensor.py index d2bd94f..05b8bb0 100644 --- a/custom_components/pfsense/binary_sensor.py +++ b/custom_components/pfsense/binary_sensor.py @@ -30,7 +30,7 @@ async def async_setup_entry( def process_entities_callback(hass, config_entry): data = hass.data[DOMAIN][config_entry.entry_id] coordinator = data[COORDINATOR] - entities = [ + return [ PfSenseCarpStatusBinarySensor( config_entry, coordinator, @@ -42,7 +42,6 @@ def process_entities_callback(hass, config_entry): False, ) ] - return entities cem = CoordinatorEntityManager( hass, @@ -55,6 +54,8 @@ def process_entities_callback(hass, config_entry): class PfSenseBinarySensor(PfSenseEntity, BinarySensorEntity): + """Base class for pfSense binary sensors.""" + def __init__( self, config_entry, @@ -74,20 +75,26 @@ def __init__( @property def is_on(self): + """Return true if the entity is on.""" return False @property def device_class(self): + """Return the device class.""" return None @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" return None class PfSenseCarpStatusBinarySensor(PfSenseBinarySensor): + """Binary sensor for the CARP maintenance/enable state.""" + @property def is_on(self): + """Return true if the entity is on.""" state = self.coordinator.data try: return state["carp_status"] diff --git a/custom_components/pfsense/button.py b/custom_components/pfsense/button.py index f165921..1ca6cbe 100644 --- a/custom_components/pfsense/button.py +++ b/custom_components/pfsense/button.py @@ -27,7 +27,7 @@ def process_entities_callback(hass, config_entry): data = hass.data[DOMAIN][config_entry.entry_id] coordinator = data[COORDINATOR] - entities = [ + return [ PfSenseRebootButton( config_entry, coordinator, @@ -52,7 +52,6 @@ def process_entities_callback(hass, config_entry): ), ), ] - return entities cem = CoordinatorEntityManager( hass, @@ -73,6 +72,7 @@ def __init__( coordinator: DataUpdateCoordinator, entity_description: ButtonEntityDescription, ) -> None: + """Initialize the object.""" self.config_entry = config_entry self.entity_description = entity_description self.coordinator = coordinator @@ -83,15 +83,24 @@ def __init__( class PfSenseRebootButton(PfSenseButton): + """Button that reboots the firewall.""" + async def async_press(self) -> None: + """Handle the button press.""" await self.service_system_reboot() class PfSenseHaltButton(PfSenseButton): + """Button that halts the firewall.""" + async def async_press(self) -> None: + """Handle the button press.""" await self.service_system_halt() class PfSenseResetStatesButton(PfSenseButton): + """Button that flushes the firewall state table.""" + async def async_press(self) -> None: + """Handle the button press.""" await self.service_reset_state_table() diff --git a/custom_components/pfsense/config_flow.py b/custom_components/pfsense/config_flow.py index 82a04b5..ee9eff6 100644 --- a/custom_components/pfsense/config_flow.py +++ b/custom_components/pfsense/config_flow.py @@ -5,18 +5,14 @@ import logging from urllib.parse import urlparse +import voluptuous as vol + from homeassistant import config_entries -from homeassistant.const import ( - CONF_NAME, - CONF_SCAN_INTERVAL, - CONF_URL, - CONF_VERIFY_SSL, -) +from homeassistant.const import CONF_NAME, CONF_SCAN_INTERVAL, CONF_URL, CONF_VERIFY_SSL from homeassistant.core import callback from homeassistant.helpers.aiohttp_client import async_get_clientsession import homeassistant.helpers.config_validation as cv from homeassistant.util import slugify -import voluptuous as vol from .const import ( CONF_API_KEY, @@ -69,6 +65,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): VERSION = 3 def __init__(self) -> None: + """Initialize the flow state.""" self._reauth_entry: config_entries.ConfigEntry | None = None async def async_step_user(self, user_input=None): @@ -116,7 +113,7 @@ async def async_step_user(self, user_input=None): errors["base"] = "cannot_connect_ssl" else: errors["base"] = "cannot_connect" - except Exception: # noqa: BLE001 + except Exception: _LOGGER.exception("Unexpected error validating pfSense connection") errors["base"] = "unknown" @@ -169,7 +166,7 @@ async def async_step_reauth_confirm(self, user_input=None): errors["base"] = "privilege_missing" except (PfSenseConnectionError, PfSenseNotFoundError): errors["base"] = "cannot_connect" - except Exception: # noqa: BLE001 + except Exception: _LOGGER.exception("Unexpected error during pfSense reauth") errors["base"] = "unknown" else: @@ -195,6 +192,7 @@ async def async_step_reauth_confirm(self, user_input=None): @staticmethod @callback def async_get_options_flow(config_entry): + """Return the options flow handler.""" return OptionsFlowHandler() @@ -202,9 +200,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow): """Handle the pfSense options flow.""" def __init__(self) -> None: + """Initialize the object.""" self.new_options: dict | None = None async def async_step_init(self, user_input=None): + """Handle the options form.""" if user_input is not None: if user_input.get(CONF_DEVICE_TRACKER_ENABLED): self.new_options = user_input diff --git a/custom_components/pfsense/const.py b/custom_components/pfsense/const.py index d0cdd92..33f742a 100644 --- a/custom_components/pfsense/const.py +++ b/custom_components/pfsense/const.py @@ -9,11 +9,7 @@ SensorEntityDescription, SensorStateClass, ) -from homeassistant.const import ( - PERCENTAGE, - UnitOfTemperature, - UnitOfTime, -) +from homeassistant.const import PERCENTAGE, UnitOfTemperature, UnitOfTime DEFAULT_USERNAME = "admin" DOMAIN = "pfsense" diff --git a/custom_components/pfsense/device_tracker.py b/custom_components/pfsense/device_tracker.py index f2775b0..5f210b9 100644 --- a/custom_components/pfsense/device_tracker.py +++ b/custom_components/pfsense/device_tracker.py @@ -2,9 +2,12 @@ from __future__ import annotations +from collections.abc import Mapping import logging import time -from typing import Any, Mapping +from typing import Any + +from mac_vendor_lookup import AsyncMacLookup from homeassistant.components.device_tracker import SourceType from homeassistant.components.device_tracker.config_entry import ScannerEntity @@ -18,7 +21,6 @@ from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.util import slugify -from mac_vendor_lookup import AsyncMacLookup from . import CoordinatorEntityManager, PfSenseEntity, dict_get from .const import ( @@ -35,6 +37,7 @@ def lookup_mac(mac_vendor_lookup: AsyncMacLookup, mac: str) -> str: + """Return the OUI vendor string for a MAC address.""" mac = mac_vendor_lookup.sanitise(mac) if isinstance(mac, str): mac = mac.encode("utf8") @@ -55,11 +58,11 @@ async def async_setup_entry( mac_vendor_lookup = AsyncMacLookup() try: await mac_vendor_lookup.update_vendors() - except Exception: + except Exception: # noqa: BLE001 - OUI lookup is optional try: await mac_vendor_lookup.load_vendors() - except Exception: - pass + except Exception as err: # noqa: BLE001 - continue without vendors + _LOGGER.debug("MAC vendor database unavailable: %s", err) dev_reg = async_get_dev_reg(hass) @@ -86,24 +89,23 @@ def process_entities_callback( if configured_mac_addresses: mac_addresses = configured_mac_addresses enabled_default = True - else: - if device_per_arp_entry: - arp_entries = dict_get(state, "arp_table") - if not arp_entries: - return [] - - mac_addresses = [ - mac_address.lower() - for arp_entry in arp_entries - if (mac_address := arp_entry.get("mac_address")) - ] + elif device_per_arp_entry: + arp_entries = dict_get(state, "arp_table") + if not arp_entries: + return [] + + mac_addresses = [ + mac_address.lower() + for arp_entry in arp_entries + if (mac_address := arp_entry.get("mac_address")) + ] for mac_address in mac_addresses: mac_vendor = None try: mac_vendor = lookup_mac(mac_vendor_lookup, mac_address) - except Exception: - pass + except Exception as err: # noqa: BLE001 - unknown OUI, leave vendor unset + _LOGGER.debug("MAC vendor lookup failed for %s: %s", mac_address, err) entity = PfSenseScannerEntity( hass, @@ -183,6 +185,7 @@ def _get_pfsense_arp_entry(self) -> dict[str, str]: @property def available(self) -> bool: + """Return whether the entity is available.""" state = self.coordinator.data arp_table = dict_get(state, "arp_table") if arp_table is None: @@ -204,8 +207,8 @@ def _extra_state_attributes(self) -> Mapping[str, Any] | None: """Return extra state attributes.""" entry = self._get_pfsense_arp_entry() if entry is not None: - for property in ["interface", "expires", "type"]: - self._extra_state[property] = entry.get(property) + for prop in ["interface", "expires", "type"]: + self._extra_state[prop] = entry.get(prop) if self._last_known_hostname is not None: self._extra_state["last_known_hostname"] = self._last_known_hostname @@ -280,7 +283,7 @@ def icon(self) -> str: """Return device icon.""" try: return "mdi:lan-connect" if self.is_connected else "mdi:lan-disconnect" - except Exception: + except (KeyError, TypeError): return "mdi:lan-disconnect" @property diff --git a/custom_components/pfsense/manifest.json b/custom_components/pfsense/manifest.json index 1dc6868..0898b1e 100644 --- a/custom_components/pfsense/manifest.json +++ b/custom_components/pfsense/manifest.json @@ -12,5 +12,5 @@ "requirements": [ "mac-vendor-lookup>=0.1.11" ], - "version": "0.9.3" + "version": "0.10.0" } diff --git a/custom_components/pfsense/pypfsense/__init__.py b/custom_components/pfsense/pypfsense/__init__.py index 7aa09e5..6d5a844 100644 --- a/custom_components/pfsense/pypfsense/__init__.py +++ b/custom_components/pfsense/pypfsense/__init__.py @@ -13,10 +13,11 @@ from __future__ import annotations import asyncio +import contextlib import ipaddress import logging import re -from typing import Any +from typing import Any, ClassVar from urllib.parse import urlparse import aiohttp @@ -30,12 +31,12 @@ def dict_get(data: dict, path: str, default=None): """Traverse a nested dict/list by a dotted path. Numeric segments index lists.""" result = data - for key in re.split(r"\.", path): - try: + try: + for key in re.split(r"\.", path): key = int(key) if key.isnumeric() else key result = result[key] - except (KeyError, IndexError, TypeError): - return default + except (KeyError, IndexError, TypeError): + return default return result @@ -63,6 +64,7 @@ class PfSenseAPIError(PfSenseError): """Any other non-2xx response. Carries ``code`` / ``response_id`` / ``message``.""" def __init__(self, code: int, response_id: str, message: str) -> None: + """Store the HTTP code, machine response id and human message.""" self.code = code self.response_id = response_id self.message = message @@ -95,6 +97,7 @@ def __init__( session: aiohttp.ClientSession, opts: dict | None = None, ) -> None: + """Store the base URL, API key, aiohttp session and options.""" opts = opts or {} parts = urlparse(url.rstrip("/")) self._base = f"{parts.scheme}://{parts.netloc}{API_BASE}" @@ -163,7 +166,7 @@ async def _apply(self, area: str) -> None: # -------------------------------------------------------------- identity async def get_system_info(self) -> dict: - """hostname / domain / serial / netgate id / platform, old-client shape.""" + """Hostname / domain / serial / netgate id / platform, old-client shape.""" status, hostname = await asyncio.gather( self._get("/status/system"), self._get("/system/hostname"), @@ -199,6 +202,7 @@ async def get_firmware_update_info(self) -> dict | None: return None async def get_dns_servers(self) -> list[str]: + """Return the configured system DNS servers.""" data = await self._get("/system/dns") return data.get("dnsserver", []) if isinstance(data, dict) else [] @@ -211,14 +215,18 @@ async def get_telemetry(self) -> dict: ``interfaces.{name}.{counter}``, ``gateways.{name}.{prop}``, ``openvpn.servers.{vpnid}.{prop}``, ``cpu.*``, ``system.*``, ``wan_ip``. """ - system, interfaces, gateways, ovpn_servers, gateways_detail = ( - await asyncio.gather( - self._get("/status/system"), - self._get("/status/interfaces"), - self._get("/status/gateways"), - self._get("/status/openvpn/servers"), - self.get_gateways_detail(), - ) + ( + system, + interfaces, + gateways, + ovpn_servers, + gateways_detail, + ) = await asyncio.gather( + self._get("/status/system"), + self._get("/status/interfaces"), + self._get("/status/gateways"), + self._get("/status/openvpn/servers"), + self.get_gateways_detail(), ) return _build_telemetry( system, interfaces, gateways, ovpn_servers, gateways_detail @@ -227,6 +235,7 @@ async def get_telemetry(self) -> dict: # ------------------------------------------------------------- services async def get_services(self) -> list[dict]: + """Return the list of pfSense services and their status.""" data = await self._get("/status/services") return data or [] @@ -243,13 +252,19 @@ async def _find_service(self, service_name: str) -> dict | None: return svc return None - async def start_service(self, service_name: str, service: dict | None = None) -> None: + async def start_service( + self, service_name: str, service: dict | None = None + ) -> None: + """Start a pfSense service by name.""" svc = service if isinstance(service, dict) and "id" in service else None svc = svc or await self._find_service(service_name) if svc: await self._service_action(svc, "start") - async def stop_service(self, service_name: str, service: dict | None = None) -> None: + async def stop_service( + self, service_name: str, service: dict | None = None + ) -> None: + """Stop a pfSense service by name.""" svc = service if isinstance(service, dict) and "id" in service else None svc = svc or await self._find_service(service_name) if svc: @@ -258,6 +273,7 @@ async def stop_service(self, service_name: str, service: dict | None = None) -> async def restart_service( self, service_name: str, service: dict | None = None ) -> None: + """Restart a pfSense service by name.""" svc = service if isinstance(service, dict) and "id" in service else None svc = svc or await self._find_service(service_name) if svc: @@ -266,6 +282,7 @@ async def restart_service( async def restart_service_if_running( self, service_name: str, service: dict | None = None ) -> None: + """Restart a pfSense service only if it is currently running.""" svc = service if isinstance(service, dict) and "id" in service else None svc = svc or await self._find_service(service_name) if svc and svc.get("status"): @@ -274,16 +291,19 @@ async def restart_service_if_running( # ---------------------------------------------------------------- dhcp async def get_dhcp_leases(self, dns_lookups=None) -> list[dict]: + """Return the current DHCP leases.""" data = await self._get("/status/dhcp_server/leases") return data or [] # ----------------------------------------------------------------- arp async def get_arp_table(self, resolve_hostnames: bool = False) -> list[dict]: + """Return the ARP table entries.""" data = await self._get("/diagnostics/arp_table") return data or [] async def delete_arp_entry(self, ip: str) -> None: + """Delete the ARP entry for an IP address.""" if not ip: return entry_id: Any = ip @@ -292,12 +312,10 @@ async def delete_arp_entry(self, ip: str) -> None: if entry.get("ip_address") == ip: entry_id = entry.get("id", ip) break - try: + with contextlib.suppress(PfSenseNotFoundError): await self._request( "DELETE", "/diagnostics/arp_table/entry", params={"id": entry_id} ) - except PfSenseNotFoundError: - pass # ------------------------------------------------------------ gateways @@ -323,6 +341,7 @@ async def get_gateways_detail(self) -> dict: return out async def set_default_gateway(self, gateway: str, ip_version: str = "4") -> None: + """Set the default IPv4 or IPv6 gateway and apply routing.""" key = "defaultgw6" if "6" in str(ip_version) else "defaultgw4" async with self._write_lock: await self._request( @@ -333,14 +352,17 @@ async def set_default_gateway(self, gateway: str, ip_version: str = "4") -> None # -------------------------------------------------------- firewall rules async def get_filter_rules(self) -> list[dict]: + """Return the firewall filter rules.""" data = await self._get("/firewall/rules") return data or [] async def get_nat_port_forward_rules(self) -> list[dict]: + """Return the NAT port-forward rules.""" data = await self._get("/firewall/nat/port_forwards") return data or [] async def get_nat_outbound_rules(self) -> list[dict]: + """Return the NAT outbound mappings.""" data = await self._get("/firewall/nat/outbound/mappings") return data or [] @@ -350,7 +372,7 @@ async def get_nat_outbound_rules(self) -> list[dict]: # fields make an otherwise unrelated toggle fail with # ``FIELD_EMPTY_NOT_ALLOWED``. Re-send them with the value pfSense would # have defaulted to, which is a no-op for the rule's behaviour. - _RULE_REQUIRED_DEFAULTS = { + _RULE_REQUIRED_DEFAULTS: ClassVar[dict[str, dict[str, str]]] = { "/firewall/rule": {"statetype": "keep state"}, } @@ -363,25 +385,34 @@ async def _set_rule_disabled( if bool(rule.get("disabled")) == disabled: return payload = {"id": rule["id"], "disabled": disabled} - for field, fallback in self._RULE_REQUIRED_DEFAULTS.get(path, {}).items(): - if not rule.get(field): - payload[field] = fallback + payload.update( + { + field: fallback + for field, fallback in self._RULE_REQUIRED_DEFAULTS.get( + path, {} + ).items() + if not rule.get(field) + } + ) async with self._write_lock: await self._request("PATCH", path, payload=payload) await self._apply("firewall") return async def enable_filter_rule_by_tracker(self, tracker) -> None: + """Enable the firewall rule with the given tracker id.""" await self._set_rule_disabled( "/firewall/rule", await self.get_filter_rules(), "tracker", tracker, False ) async def disable_filter_rule_by_tracker(self, tracker) -> None: + """Disable the firewall rule with the given tracker id.""" await self._set_rule_disabled( "/firewall/rule", await self.get_filter_rules(), "tracker", tracker, True ) async def enable_nat_port_forward_rule_by_created_time(self, created_time) -> None: + """Enable the NAT port-forward rule with the given created_time.""" await self._set_rule_disabled( "/firewall/nat/port_forward", await self.get_nat_port_forward_rules(), @@ -391,6 +422,7 @@ async def enable_nat_port_forward_rule_by_created_time(self, created_time) -> No ) async def disable_nat_port_forward_rule_by_created_time(self, created_time) -> None: + """Disable the NAT port-forward rule with the given created_time.""" await self._set_rule_disabled( "/firewall/nat/port_forward", await self.get_nat_port_forward_rules(), @@ -400,6 +432,7 @@ async def disable_nat_port_forward_rule_by_created_time(self, created_time) -> N ) async def enable_nat_outbound_rule_by_created_time(self, created_time) -> None: + """Enable the NAT outbound mapping with the given created_time.""" await self._set_rule_disabled( "/firewall/nat/outbound/mapping", await self.get_nat_outbound_rules(), @@ -409,6 +442,7 @@ async def enable_nat_outbound_rule_by_created_time(self, created_time) -> None: ) async def disable_nat_outbound_rule_by_created_time(self, created_time) -> None: + """Disable the NAT outbound mapping with the given created_time.""" await self._set_rule_disabled( "/firewall/nat/outbound/mapping", await self.get_nat_outbound_rules(), @@ -426,6 +460,7 @@ async def update_alias_address( action: str = "add", kill_states: bool = True, ) -> None: + """Add or remove an address in a firewall alias and apply.""" aliases = await self._get("/firewall/aliases") or [] target = next((a for a in aliases if a.get("name") == alias_name), None) @@ -477,12 +512,14 @@ async def update_alias_address( # ----------------------------------------------------------- carp / vip async def get_carp_status(self) -> bool: + """Return True when CARP is enabled and not in maintenance mode.""" data = await self._get("/status/carp") if not isinstance(data, dict): return False return bool(data.get("enable")) and not data.get("maintenance_mode") async def get_carp_interfaces(self) -> list[dict]: + """Return the CARP virtual IPs with their status.""" data = await self._get("/firewall/virtual_ips") or [] carp = [] for vip in data: @@ -496,11 +533,11 @@ async def get_carp_interfaces(self) -> list[dict]: # --------------------------------------------------------- state table async def reset_state_table(self) -> None: - await self._request( - "DELETE", "/firewall/states", params={"limit": 0} - ) + """Flush the entire firewall state table.""" + await self._request("DELETE", "/firewall/states", params={"limit": 0}) async def kill_states(self, source: str, destination: str | None = None) -> None: + """Kill states for a source (and optional destination) via pfctl.""" cmd = f"/sbin/pfctl -k {_shq(source)}" if destination: cmd += f" -k {_shq(destination)}" @@ -528,14 +565,12 @@ async def kill_states_for_rule(self, rule: dict) -> None: async with self._write_lock: for prefix in prefixes: for field in ("source", "destination"): - try: + with contextlib.suppress(PfSenseAPIError): await self._request( "DELETE", "/firewall/states", params={f"{field}__startswith": prefix, "limit": 0}, ) - except PfSenseAPIError: - pass async def _rule_match_networks(self, rule: dict) -> list: """Concrete ``ip_network`` objects for a rule's source + destination.""" @@ -563,20 +598,21 @@ async def _rule_match_networks(self, rule: dict) -> list: # ------------------------------------------------------- system control async def system_reboot(self, type: str = "normal") -> None: - try: + """Reboot the firewall.""" + # The connection drops as the box goes down -- that is success. + with contextlib.suppress(PfSenseConnectionError): await self._request("POST", "/diagnostics/reboot", payload={}) - except PfSenseConnectionError: - pass # connection drops as the box goes down async def system_halt(self) -> None: - try: + """Halt (power off) the firewall.""" + # The connection drops as the box goes down -- that is success. + with contextlib.suppress(PfSenseConnectionError): await self._request("POST", "/diagnostics/halt_system", payload={}) - except PfSenseConnectionError: - pass # ------------------------------------------------------------------ wol async def send_wol(self, interface: str, mac: str) -> None: + """Send a Wake-on-LAN magic packet on an interface.""" await self._request( "POST", "/services/wake_on_lan/send", @@ -586,6 +622,7 @@ async def send_wol(self, interface: str, mac: str) -> None: # -------------------------------------------------------------- exec_* async def exec_command(self, command: str, background: bool = False) -> str: + """Run a shell command via the diagnostics endpoint and return its output.""" if background: command = f"{command} &" data = await self._request( @@ -611,11 +648,11 @@ def _as_network(value: str): def _states_prefix(net) -> str | None: - """``str`` that ``firewall/state`` source/destination values start with for - every address in ``net``, or ``None`` if ``net`` can't be expressed that way. + """Return the string every ``firewall/state`` endpoint in ``net`` starts with. - States render endpoints as ``ip:port`` (IPv4) so a host becomes ``"ip:"`` - and an octet-aligned network becomes its leading octets plus a dot. + Returns ``None`` when ``net`` can't be expressed as such a prefix. States + render endpoints as ``ip:port`` (IPv4), so a host becomes ``"ip:"`` and an + octet-aligned network becomes its leading octets plus a dot. """ if net.version != 4: return None @@ -648,7 +685,7 @@ def _expand_alias_networks(aliases: list[dict], name: str, _depth: int = 3) -> l def _flatten_params(params: dict | None) -> dict | None: - """aiohttp needs str values; drop ``None`` and stringify the rest.""" + """Aiohttp needs str values; drop ``None`` and stringify the rest.""" if not params: return None return {k: str(v) for k, v in params.items() if v is not None} diff --git a/custom_components/pfsense/sensor.py b/custom_components/pfsense/sensor.py index 5127480..52dcbae 100644 --- a/custom_components/pfsense/sensor.py +++ b/custom_components/pfsense/sensor.py @@ -34,7 +34,7 @@ _LOGGER = logging.getLogger(__name__) -async def async_setup_entry( +async def async_setup_entry( # noqa: C901 - flat per-telemetry-type entity builder hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: entity_platform.AddEntitiesCallback, @@ -42,11 +42,11 @@ async def async_setup_entry( """Set up the pfSense sensors.""" @callback - def process_entities_callback(hass, config_entry): + def process_entities_callback(hass, config_entry): # noqa: C901 - see above data = hass.data[DOMAIN][config_entry.entry_id] coordinator = data[COORDINATOR] state = coordinator.data - resources = [sensor_id for sensor_id in SENSOR_TYPES] + resources = list(SENSOR_TYPES) entities = [] @@ -82,7 +82,7 @@ def process_entities_callback(hass, config_entry): coordinator, SensorEntityDescription( key=f"telemetry.filesystems.{device_clean}", - name="Filesystem Used Percentage {}".format(mountpoint_clean), + name=f"Filesystem Used Percentage {mountpoint_clean}", native_unit_of_measurement=PERCENTAGE, icon="mdi:harddisk", state_class=SensorStateClass.MEASUREMENT, @@ -113,9 +113,9 @@ def process_entities_callback(hass, config_entry): ) entities.append(entity) - for interface_name in dict_get(state, "telemetry.interfaces", {}).keys(): + for interface_name in dict_get(state, "telemetry.interfaces", {}): interface = state["telemetry"]["interfaces"][interface_name] - for property in [ + for prop in [ "status", "inerrs", "outerrs", @@ -142,7 +142,7 @@ def process_entities_callback(hass, config_entry): icon = None enabled_default = False - if property in [ + if prop in [ "status", "inbytes_kilobytes_per_second", "outbytes_kilobytes_per_second", @@ -151,33 +151,30 @@ def process_entities_callback(hass, config_entry): ]: enabled_default = True - if ( - "_packets_per_second" in property - or "_kilobytes_per_second" in property - ): + if "_packets_per_second" in prop or "_kilobytes_per_second" in prop: state_class = SensorStateClass.MEASUREMENT - if "_packets_per_second" in property: + if "_packets_per_second" in prop: native_unit_of_measurement = DATA_RATE_PACKETS_PER_SECOND - if "_kilobytes_per_second" in property: + if "_kilobytes_per_second" in prop: native_unit_of_measurement = UnitOfDataRate.KILOBYTES_PER_SECOND if native_unit_of_measurement is None: - if "bytes" in property: + if "bytes" in prop: native_unit_of_measurement = UnitOfInformation.BYTES state_class = SensorStateClass.TOTAL_INCREASING - if "pkts" in property: + if "pkts" in prop: native_unit_of_measurement = DATA_PACKETS state_class = SensorStateClass.TOTAL_INCREASING - if property in ["inerrs", "outerrs", "collisions"]: + if prop in ["inerrs", "outerrs", "collisions"]: native_unit_of_measurement = COUNT - if "pkts" in property or "bytes" in property: + if "pkts" in prop or "bytes" in prop: icon = "mdi:server-network" - if property == "status": + if prop == "status": icon = "mdi:check-network-outline" if icon is None: @@ -188,9 +185,9 @@ def process_entities_callback(hass, config_entry): coordinator, SensorEntityDescription( key="telemetry.interface.{}.{}".format( - interface["ifname"], property + interface["ifname"], prop ), - name="Interface {} {}".format(interface["descr"], property), + name="Interface {} {}".format(interface["descr"], prop), native_unit_of_measurement=native_unit_of_measurement, icon=icon, state_class=state_class, @@ -199,29 +196,29 @@ def process_entities_callback(hass, config_entry): ) entities.append(entity) - for gateway_name in dict_get(state, "telemetry.gateways", {}).keys(): + for gateway_name in dict_get(state, "telemetry.gateways", {}): gateway = state["telemetry"]["gateways"][gateway_name] - for property in ["status", "delay", "stddev", "loss"]: + for prop in ["status", "delay", "stddev", "loss"]: state_class = None native_unit_of_measurement = None icon = "mdi:router-network" enabled_default = True - if property == "loss": + if prop == "loss": native_unit_of_measurement = PERCENTAGE - if property in ["delay", "stddev"]: + if prop in ["delay", "stddev"]: native_unit_of_measurement = UnitOfTime.MILLISECONDS - if property == "status": + if prop == "status": icon = "mdi:check-network-outline" entity = PfSenseGatewaySensor( config_entry, coordinator, SensorEntityDescription( - key="telemetry.gateway.{}.{}".format(gateway["name"], property), - name="Gateway {} {}".format(gateway["name"], property), + key="telemetry.gateway.{}.{}".format(gateway["name"], prop), + name="Gateway {} {}".format(gateway["name"], prop), native_unit_of_measurement=native_unit_of_measurement, icon=icon, state_class=state_class, @@ -230,10 +227,10 @@ def process_entities_callback(hass, config_entry): ) entities.append(entity) - for vpnid in dict_get(state, "telemetry.openvpn.servers", {}).keys(): + for vpnid in dict_get(state, "telemetry.openvpn.servers", {}): servers = dict_get(state, "telemetry.openvpn.servers", {}) server = servers[vpnid] - for property in [ + for prop in [ "connected_client_count", "total_bytes_recv", "total_bytes_sent", @@ -245,26 +242,25 @@ def process_entities_callback(hass, config_entry): icon = None enabled_default = False - if "_kilobytes_per_second" in property: + if "_kilobytes_per_second" in prop: state_class = SensorStateClass.MEASUREMENT - if property == "connected_client_count": + if prop == "connected_client_count": state_class = SensorStateClass.MEASUREMENT - if "_kilobytes_per_second" in property: + if "_kilobytes_per_second" in prop: native_unit_of_measurement = UnitOfDataRate.KILOBYTES_PER_SECOND - if native_unit_of_measurement is None: - if "bytes" in property: - native_unit_of_measurement = UnitOfInformation.BYTES + if native_unit_of_measurement is None and "bytes" in prop: + native_unit_of_measurement = UnitOfInformation.BYTES - if property in ["connected_client_count"]: + if prop == "connected_client_count": native_unit_of_measurement = "clients" - if "bytes" in property: + if "bytes" in prop: icon = "mdi:server-network" - if property == "connected_client_count": + if prop == "connected_client_count": icon = "mdi:ip-network-outline" if icon is None: @@ -274,9 +270,9 @@ def process_entities_callback(hass, config_entry): config_entry, coordinator, SensorEntityDescription( - key="telemetry.openvpn.servers.{}.{}".format(vpnid, property), + key=f"telemetry.openvpn.servers.{vpnid}.{prop}", name="OpenVPN Server {} ({}) {}".format( - vpnid, server["name"], property + vpnid, server["name"], prop ), native_unit_of_measurement=native_unit_of_measurement, icon=icon, @@ -299,6 +295,7 @@ def process_entities_callback(hass, config_entry): def normalize_filesystem_device_name(device_name): + """Return a slug-safe token for a filesystem device or mountpoint.""" return device_name.replace("/", "_slash_").strip("_") @@ -325,8 +322,11 @@ def __init__( class PfSenseStaticKeySensor(PfSenseSensor): + """Sensor backed by a fixed telemetry key from SENSOR_TYPES.""" + @property def available(self) -> bool: + """Return whether the entity is available.""" value = self._get_pfsense_state_value(self.entity_description.key) if value is None: return False @@ -336,6 +336,7 @@ def available(self) -> bool: @property def native_value(self): + """Return the entity's current value.""" value = self._get_pfsense_state_value(self.entity_description.key) if value is None: return STATE_UNKNOWN @@ -361,6 +362,8 @@ def extra_state_attributes(self): class PfSenseFilesystemSensor(PfSenseSensor): + """Sensor for a filesystem's used-space percentage.""" + def _pfsense_get_filesystem(self): state = self.coordinator.data found = None @@ -373,6 +376,7 @@ def _pfsense_get_filesystem(self): @property def available(self) -> bool: + """Return whether the entity is available.""" filesystem = self._pfsense_get_filesystem() if filesystem is None: return False @@ -380,11 +384,13 @@ def available(self) -> bool: @property def native_value(self): + """Return the entity's current value.""" filesystem = self._pfsense_get_filesystem() return filesystem["percent_used"] @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" attributes = {} filesystem = self._pfsense_get_filesystem() for attr in ["device", "type", "total_size", "mountpoint"]: @@ -393,6 +399,8 @@ def extra_state_attributes(self): class PfSenseInterfaceSensor(PfSenseSensor): + """Sensor for a network interface counter or status.""" + def _pfsense_get_interface_property_name(self): return self.entity_description.key.split(".")[3] @@ -403,7 +411,7 @@ def _pfsense_get_interface(self): state = self.coordinator.data found = None interface_name = self._pfsense_get_interface_name() - for i_interface_name in state["telemetry"]["interfaces"].keys(): + for i_interface_name in state["telemetry"]["interfaces"]: if i_interface_name == interface_name: found = state["telemetry"]["interfaces"][i_interface_name] break @@ -411,14 +419,16 @@ def _pfsense_get_interface(self): @property def available(self) -> bool: + """Return whether the entity is available.""" interface = self._pfsense_get_interface() - property = self._pfsense_get_interface_property_name() - if interface is None or property not in interface.keys(): + prop = self._pfsense_get_interface_property_name() + if interface is None or prop not in interface: return False return super().available @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" attributes = {} interface = self._pfsense_get_interface() for attr in ["hwif", "enable", "if", "macaddr", "mtu", "media"]: @@ -428,22 +438,26 @@ def extra_state_attributes(self): @property def icon(self): - property = self._pfsense_get_interface_property_name() - if property == "status" and self.native_value != "up": + """Return the entity icon.""" + prop = self._pfsense_get_interface_property_name() + if prop == "status" and self.native_value != "up": return "mdi:close-network-outline" return super().icon @property def native_value(self): + """Return the entity's current value.""" interface = self._pfsense_get_interface() - property = self._pfsense_get_interface_property_name() + prop = self._pfsense_get_interface_property_name() try: - return interface[property] + return interface[prop] except KeyError: return STATE_UNKNOWN class PfSenseCarpInterfaceSensor(PfSenseSensor): + """Sensor for a CARP virtual IP's status.""" + def _pfsense_get_interface_name(self): return self.entity_description.key.split(".")[2] @@ -459,6 +473,7 @@ def _pfsense_get_interface(self): @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" attributes = {} interface = self._pfsense_get_interface() for attr in [ @@ -475,6 +490,7 @@ def extra_state_attributes(self): @property def available(self) -> bool: + """Return whether the entity is available.""" interface = self._pfsense_get_interface() if interface is None: return False @@ -482,12 +498,14 @@ def available(self) -> bool: @property def icon(self): + """Return the entity icon.""" if self.native_value != "MASTER": return "mdi:close-network-outline" return super().icon @property def native_value(self): + """Return the entity's current value.""" interface = self._pfsense_get_interface() try: return interface["status"] @@ -496,6 +514,8 @@ def native_value(self): class PfSenseGatewaySensor(PfSenseSensor): + """Sensor for a gateway's status or latency metric.""" + def _pfsense_get_gateway_property_name(self): return self.entity_description.key.split(".")[3] @@ -506,7 +526,7 @@ def _pfsense_get_gateway(self): state = self.coordinator.data found = None gateway_name = self._pfsense_get_gateway_name() - for i_gateway_name in state["telemetry"]["gateways"].keys(): + for i_gateway_name in state["telemetry"]["gateways"]: if i_gateway_name == gateway_name: found = state["telemetry"]["gateways"][i_gateway_name] break @@ -516,7 +536,7 @@ def _pfsense_get_gateway_details(self): state = self.coordinator.data found = None gateway_name = self._pfsense_get_gateway_name() - for i_gateway_name in state["telemetry"]["gateways_detail"].keys(): + for i_gateway_name in state["telemetry"]["gateways_detail"]: if i_gateway_name == gateway_name: found = state["telemetry"]["gateways_detail"][i_gateway_name] break @@ -524,13 +544,14 @@ def _pfsense_get_gateway_details(self): @property def available(self) -> bool: + """Return whether the entity is available.""" gateway = self._pfsense_get_gateway() - property = self._pfsense_get_gateway_property_name() - if gateway is None or property not in gateway.keys(): + prop = self._pfsense_get_gateway_property_name() + if gateway is None or prop not in gateway: return False - if property in ["stddev", "delay", "loss"]: - value = gateway[property] + if prop in ["stddev", "delay", "loss"]: + value = gateway[prop] if isinstance(value, str): value = re.sub(r"[^0-9\.]*", "", value) if len(value) < 1: @@ -539,6 +560,7 @@ def available(self) -> bool: @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" attributes = {} gateway = self._pfsense_get_gateway() gateway_detail = self._pfsense_get_gateway_details() @@ -553,44 +575,46 @@ def extra_state_attributes(self): if attr in gateway_detail: value = gateway_detail[attr] attributes[attr] = value - else: - if attr == "isdefaultgw": - value = False - attributes[attr] = value + elif attr == "isdefaultgw": + value = False + attributes[attr] = value return attributes @property def icon(self): - property = self._pfsense_get_gateway_property_name() - if property == "status" and self.native_value != "online": + """Return the entity icon.""" + prop = self._pfsense_get_gateway_property_name() + if prop == "status" and self.native_value != "online": return "mdi:close-network-outline" return super().icon @property def native_value(self): + """Return the entity's current value.""" gateway = self._pfsense_get_gateway() - property = self._pfsense_get_gateway_property_name() + prop = self._pfsense_get_gateway_property_name() if gateway is None: return STATE_UNKNOWN try: - value = gateway[property] - if property in ["stddev", "delay", "loss"]: - if isinstance(value, str): - value = re.sub(r"[^0-9\.]*", "", value) - if len(value) > 0: - value = float(value) + value = gateway[prop] + except KeyError: + return STATE_UNKNOWN - if isinstance(value, str) and len(value) < 1: - return STATE_UNKNOWN + if prop in ["stddev", "delay", "loss"] and isinstance(value, str): + value = re.sub(r"[^0-9\.]*", "", value) + if len(value) > 0: + value = float(value) - return value - except KeyError: + if isinstance(value, str) and len(value) < 1: return STATE_UNKNOWN + return value class PfSenseOpenVPNServerSensor(PfSenseSensor): + """Sensor for an OpenVPN server's client/throughput stats.""" + def _pfsense_get_server_property_name(self): return self.entity_description.key.split(".")[4] @@ -601,7 +625,7 @@ def _pfsense_get_server(self): state = self.coordinator.data found = None vpnid = self._pfsense_get_server_vpnid() - for server_vpnid in dict_get(state, "telemetry.openvpn.servers", {}).keys(): + for server_vpnid in dict_get(state, "telemetry.openvpn.servers", {}): if vpnid == server_vpnid: found = state["telemetry"]["openvpn"]["servers"][vpnid] break @@ -609,14 +633,16 @@ def _pfsense_get_server(self): @property def available(self) -> bool: + """Return whether the entity is available.""" server = self._pfsense_get_server() - property = self._pfsense_get_server_property_name() - if server is None or property not in server.keys(): + prop = self._pfsense_get_server_property_name() + if server is None or prop not in server: return False return super().available @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" attributes = {} server = self._pfsense_get_server() if server is None: @@ -628,13 +654,14 @@ def extra_state_attributes(self): @property def native_value(self): + """Return the entity's current value.""" server = self._pfsense_get_server() - property = self._pfsense_get_server_property_name() + prop = self._pfsense_get_server_property_name() if server is None: return STATE_UNKNOWN try: - return server[property] + return server[prop] except KeyError: return STATE_UNKNOWN diff --git a/custom_components/pfsense/services.py b/custom_components/pfsense/services.py index 3eef26f..7e86c7a 100644 --- a/custom_components/pfsense/services.py +++ b/custom_components/pfsense/services.py @@ -1,11 +1,14 @@ +"""Home Assistant service registration for the pfSense integration.""" + import logging +import voluptuous as vol + from homeassistant.core import HomeAssistant, ServiceCall, callback from homeassistant.helpers import config_validation as cv from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity_platform import async_get_platforms from homeassistant.helpers.service import entity_service_call -import voluptuous as vol from .const import ( DOMAIN, @@ -35,6 +38,8 @@ def async_get_entities(hass: HomeAssistant) -> dict[str, Entity]: class ServiceRegistrar: + """Registers the integration's Home Assistant services once.""" + def __init__( self, hass: HomeAssistant, @@ -44,26 +49,25 @@ def __init__( @callback def async_register(self): + """Register the integration's services once.""" if "loaded" in _data: return _data.add("loaded") - # --- DEFERRED RUNTIME INJECTION PATCH --- - from . import PfSenseEntity + # Deferred to break the services <-> __init__ import cycle. + from . import PfSenseEntity # noqa: PLC0415 async def service_update_alias( - self_entity, + self, alias_name: str, address: str, action: str, kill_states: bool = True, ): - """Dynamic extension mapping runtime command parameters directly to the Client interface.""" - client = self_entity._get_pfsense_client() - await client.update_alias_address( - alias_name, address, action, kill_states - ) + """Bind the update_alias service onto PfSenseEntity at runtime.""" + client = self._get_pfsense_client() + await client.update_alias_address(alias_name, address, action, kill_states) if not hasattr(PfSenseEntity, "service_update_alias"): PfSenseEntity.service_update_alias = service_update_alias diff --git a/custom_components/pfsense/switch.py b/custom_components/pfsense/switch.py index ffb9ef6..8d28a7d 100644 --- a/custom_components/pfsense/switch.py +++ b/custom_components/pfsense/switch.py @@ -60,7 +60,7 @@ def process_entities_callback(hass, config_entry): config_entry, coordinator, SwitchEntityDescription( - key="filter.{}".format(tracker), + key=f"filter.{tracker}", name="Filter Rule {} ({})".format( tracker, rule.get("descr", "") ), @@ -99,7 +99,7 @@ def process_entities_callback(hass, config_entry): config_entry, coordinator, SwitchEntityDescription( - key="{}.{}".format(rule_type, tracker), + key=f"{rule_type}.{tracker}", name="{} {} ({})".format( label, tracker, rule.get("descr", "") ), @@ -112,7 +112,7 @@ def process_entities_callback(hass, config_entry): # services for service in state["services"]: - for property in ["status"]: + for prop in ["status"]: icon = "mdi:application-cog-outline" # likely only want very specific services to manipulate from actions enabled_default = False @@ -122,14 +122,14 @@ def process_entities_callback(hass, config_entry): if service["name"] == "openvpn" and service.get("vpnid"): key = "service.{}.{}".format( service["name"] + "-" + str(service["vpnid"]), - property, + prop, ) name = "Service {} {}".format( - service["name"] + " " + service.get("description", ""), property + service["name"] + " " + service.get("description", ""), prop ) else: - key = "service.{}.{}".format(service["name"], property) - name = "Service {} {}".format(service["name"], property) + key = "service.{}.{}".format(service["name"], prop) + name = "Service {} {}".format(service["name"], prop) entity = PfSenseServiceSwitch( config_entry, @@ -157,6 +157,8 @@ def process_entities_callback(hass, config_entry): class PfSenseSwitch(PfSenseEntity, SwitchEntity): + """Base class for pfSense switches.""" + def __init__( self, config_entry, @@ -174,10 +176,12 @@ def __init__( @property def is_on(self): + """Return true if the entity is on.""" return False @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" return None async def _maybe_kill_rule_states(self, rule): @@ -190,11 +194,13 @@ async def _maybe_kill_rule_states(self, rule): return try: await self._get_pfsense_client().kill_states_for_rule(rule) - except Exception: # best effort - never fail the toggle over this + except Exception: _LOGGER.warning("failed to kill states for toggled rule", exc_info=True) class PfSenseFilterSwitch(PfSenseSwitch): + """Switch that enables or disables a firewall rule.""" + def _pfsense_get_tracker(self): return self.entity_description.key.split(".")[1] @@ -208,6 +214,7 @@ def _pfsense_get_rule(self): @property def available(self) -> bool: + """Return whether the entity is available.""" rule = self._pfsense_get_rule() if rule is None: return False @@ -216,6 +223,7 @@ def available(self) -> bool: @property def is_on(self): + """Return true if the entity is on.""" rule = self._pfsense_get_rule() if rule is None: return STATE_UNKNOWN @@ -245,6 +253,8 @@ async def async_turn_off(self, **kwargs): class PfSenseNatSwitch(PfSenseSwitch): + """Switch that enables or disables a NAT rule.""" + def _pfsense_get_rule_type(self): return self.entity_description.key.split(".")[0] @@ -267,6 +277,7 @@ def _pfsense_get_rule(self): @property def available(self) -> bool: + """Return whether the entity is available.""" rule = self._pfsense_get_rule() if rule is None: return False @@ -275,6 +286,7 @@ def available(self) -> bool: @property def is_on(self): + """Return true if the entity is on.""" rule = self._pfsense_get_rule() if rule is None: return STATE_UNKNOWN @@ -316,6 +328,8 @@ async def async_turn_off(self, **kwargs): class PfSenseServiceSwitch(PfSenseSwitch): + """Switch that starts or stops a pfSense service.""" + def _pfsense_get_property_name(self): return self.entity_description.key.split(".")[2] @@ -330,9 +344,10 @@ def _pfsense_get_service(self): if service_name.startswith("openvpn"): # [ "openvpn", """] parts = service_name.split("-") - if service["name"] == parts[0] and str( - service.get("vpnid") - ) == parts[1]: + if ( + service["name"] == parts[0] + and str(service.get("vpnid")) == parts[1] + ): found = service elif service["name"] == service_name: found = service @@ -341,20 +356,21 @@ def _pfsense_get_service(self): @property def available(self) -> bool: + """Return whether the entity is available.""" service = self._pfsense_get_service() - property = self._pfsense_get_property_name() - if service is None or property not in service.keys(): + prop = self._pfsense_get_property_name() + if service is None or prop not in service: return False return super().available @property def is_on(self): + """Return true if the entity is on.""" service = self._pfsense_get_service() - property = self._pfsense_get_property_name() + prop = self._pfsense_get_property_name() try: - value = service[property] - return value + return service[prop] except KeyError: return STATE_UNKNOWN diff --git a/custom_components/pfsense/update.py b/custom_components/pfsense/update.py index d16e3da..6ee32b2 100644 --- a/custom_components/pfsense/update.py +++ b/custom_components/pfsense/update.py @@ -57,6 +57,8 @@ def process_entities_callback(hass, config_entry): class PfSenseUpdate(PfSenseEntity, UpdateEntity): + """Base class for the pfSense update entity.""" + def __init__( self, config_entry, @@ -79,12 +81,16 @@ def __init__( @property def device_class(self): + """Return the device class.""" return UpdateDeviceClass.FIRMWARE class PfSenseFirmwareUpdatesAvailableUpdate(PfSenseUpdate): + """Update entity reporting available package upgrades.""" + @property def available(self): + """Return whether the entity is available.""" state = self.coordinator.data if ( state["firmware_update_info"] is None @@ -97,6 +103,7 @@ def available(self): @property def title(self): + """Return the title.""" return "pfSense" @property @@ -128,6 +135,7 @@ def in_progress(self): @property def extra_state_attributes(self): + """Return the entity's extra state attributes.""" state = self.coordinator.data attrs = {} info = dict_get(state, "firmware_update_info.base", {}) @@ -135,7 +143,7 @@ def extra_state_attributes(self): if not info: return attrs - for key in info.keys(): + for key in info: attrs[f"pfsense_base_{key}"] = dict_get( state, f"firmware_update_info.base.{key}" ) @@ -144,4 +152,5 @@ def extra_state_attributes(self): @property def release_url(self): + """Return the release notes URL.""" return "https://docs.netgate.com/pfsense/en/latest/releases/index.html" diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6be9ced --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,175 @@ +[tool.ruff] +# Mirrors the Home Assistant core ruff configuration so this custom +# integration is held to the same standard. Kept in sync manually with +# https://github.com/home-assistant/core/blob/dev/pyproject.toml +required-version = ">=0.16.5" + +[tool.ruff.lint] +select = [ + "A001", # Variable {name} is shadowing a Python builtin + "ASYNC", # flake8-async + "B", # flake8-bugbear + "BLE", + "C", # complexity + "COM818", # Trailing comma on bare tuple prohibited + "D", # docstrings + "DTZ003", # Use datetime.now(tz=) instead of datetime.utcnow() + "DTZ004", # Use datetime.fromtimestamp(ts, tz=) instead of datetime.utcfromtimestamp(ts) + "DTZ011", # Use datetime.now(tz=).date() instead of date.today() + "E", # pycodestyle + "F", # pyflakes/autoflake + "F541", # f-string without any placeholders + "FLY", # flynt + "FURB", # refurb + "G", # flake8-logging-format + "I", # isort + "INP", # flake8-no-pep420 + "ISC", # flake8-implicit-str-concat + "ICN001", # import concentions; {name} should be imported as {asname} + "LOG", # flake8-logging + "N804", # First argument of a class method should be named cls + "N805", # First argument of a method should be named self + "N806", # Variable {name} in function should be snake_case + "N815", # Variable {name} in class scope should not be mixedCase + "PERF", # Perflint + "PGH", # pygrep-hooks + "PIE", # flake8-pie + "PL", # pylint + "PT", # flake8-pytest-style + "PTH", # flake8-pathlib + "PYI", # flake8-pyi + "RET", # flake8-return + "RSE", # flake8-raise + "RUF", # Ruff-specific rules (see `ignore` for exclusions) + "S107", # Possible hardcoded password assigned to function default + "S102", # Use of exec detected + "S103", # bad-file-permissions + "S108", # hardcoded-temp-file + "S301", # suspicious-pickle-usage + "S306", # suspicious-mktemp-usage + "S307", # suspicious-eval-usage + "S313", # suspicious-xmlc-element-tree-usage + "S314", # suspicious-xml-element-tree-usage + "S315", # suspicious-xml-expat-reader-usage + "S316", # suspicious-xml-expat-builder-usage + "S317", # suspicious-xml-sax-usage + "S318", # suspicious-xml-mini-dom-usage + "S319", # suspicious-xml-pull-dom-usage + "S601", # paramiko-call + "S602", # subprocess-popen-with-shell-equals-true + "S604", # call-with-shell-equals-true + "S608", # hardcoded-sql-expression + "S609", # unix-command-wildcard-injection + "SIM", # flake8-simplify + "SLF", # flake8-self + "SLOT", # flake8-slots + "T100", # Trace found: {name} used + "T20", # flake8-print + "TC", # flake8-type-checking + "TID", # Tidy imports + "TRY", # tryceratops + "UP", # pyupgrade + "UP031", # Use format specifiers instead of percent format + "UP032", # Use f-string instead of `format` call + "W", # pycodestyle +] + +ignore = [ + "ASYNC109", # Async function definition with a `timeout` parameter Use `asyncio.timeout` instead + "ASYNC110", # Use `asyncio.Event` instead of awaiting `asyncio.sleep` in a `while` loop + "ASYNC240", # Use an async function for entering the file system + "B008", # Do not perform function call in argument defaults; commonly used in Home Assistant (e.g. cv.* validators) + "B019", # Use of functools.lru_cache or functools.cache on methods can lead to memory leaks + "D202", # No blank lines allowed after function docstring + "D203", # 1 blank line required before class docstring + "D213", # Multi-line docstring summary should start at the second line + "D406", # Section name should end with a newline + "D407", # Section name underlining + "D417", # Missing argument descriptions in docstring - to allow documenting only non-obvious parameters + "E501", # line too long + + "PLC1901", # {existing} can be simplified to {replacement} as an empty string is falsey; too many false positives + "PLR0911", # Too many return statements ({returns} > {max_returns}) + "PLR0912", # Too many branches ({branches} > {max_branches}) + "PLR0913", # Too many arguments to function call ({c_args} > {max_args}) + "PLR0915", # Too many statements ({statements} > {max_statements}) + "PLR0917", # Too many positional arguments defined for a function ({p_args} > {max_args}) + "PLR2004", # Magic value used in comparison, consider replacing {value} with a constant variable + "PLW0108", # Unnecessary lambda wrapping a function call; can often be replaced by the function itself + "PLW1641", # __eq__ without __hash__ + "PLW2901", # Outer {outer_kind} variable {name} overwritten by inner {inner_kind} target + "PT011", # pytest.raises({exception}) is too broad, set the `match` parameter or use a more specific exception + "PT018", # Assertion should be broken down into multiple parts + "RUF001", # String contains ambiguous unicode character. + "RUF012", # Mutable class attributes should be annotated with typing.ClassVar + "RUF015", # Prefer next(...) over single element slice + "RUF043", # Pattern passed to match= contains metacharacters but is neither escaped nor raw + "SIM102", # Use a single if statement instead of nested if statements + "SIM103", # Return the condition {condition} directly + "SIM108", # Use ternary operator {contents} instead of if-else-block + "SIM115", # Use context handler for opening files + + # Moving imports into type-checking blocks can mess with pytest.patch() + "TC001", # Move application import {} into a type-checking block + "TC002", # Move third-party import {} into a type-checking block + "TC003", # Move standard library import {} into a type-checking block + # Quotes for typing.cast generally not necessary, only for performance critical paths + "TC006", # Add quotes to type expression in typing.cast() + + "TRY003", # Avoid specifying long messages outside the exception class + "TRY400", # Use `logging.exception` instead of `logging.error` + + "UP047", # Non PEP 696 generic function + "UP049", # Avoid private type parameter names + + # May conflict with the formatter, https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules + "W191", + "E111", + "E114", + "E117", + "D206", + "D300", + "Q", + "COM812", + "COM819", + + # Disabled because ruff does not understand type of __all__ generated by a function + "PLE0605", + + "FURB116", + + # Disabled to implement in follow up PRs after ruff 0.16 bump + "ISC004", + "LOG004", +] + +[tool.ruff.lint.flake8-pytest-style] +fixture-parentheses = false +mark-parentheses = false + +[tool.ruff.lint.flake8-tidy-imports.banned-api] +"async_timeout".msg = "use asyncio.timeout instead" +"pytz".msg = "use zoneinfo instead" + +[tool.ruff.lint.isort] +force-sort-within-sections = true +known-first-party = ["homeassistant"] +combine-as-imports = true +split-on-trailing-comma = false + +[tool.ruff.lint.per-file-ignores] +# Relative imports are the norm inside a custom-component package +"custom_components/*/*" = ["TID252"] +"custom_components/*/*/*" = ["TID252"] + +# Tests reach into client / entity internals on purpose +"tests/**" = ["PTH", "SLF001"] + +# Temporary, mirroring Home Assistant core +"custom_components/**" = ["PTH"] + +[tool.ruff.lint.mccabe] +max-complexity = 25 + +[tool.ruff.lint.pydocstyle] +convention = "google" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..683bcd9 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test suite for the pfSense integration.""" diff --git a/tests/conftest.py b/tests/conftest.py index e874b4e..5dcd67d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,9 @@ +"""Shared pytest fixtures for the pfSense test suite.""" + import pytest @pytest.fixture(autouse=True) def auto_enable_custom_integrations(enable_custom_integrations): """Enable custom integrations automatically for all tests.""" - yield + return diff --git a/tests/test_binary_sensor.py b/tests/test_binary_sensor.py index d4439a1..b701298 100644 --- a/tests/test_binary_sensor.py +++ b/tests/test_binary_sensor.py @@ -3,20 +3,22 @@ from unittest.mock import AsyncMock, patch import pytest -from homeassistant.const import CONF_URL, CONF_VERIFY_SSL -from homeassistant.core import HomeAssistant from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.pfsense.const import CONF_API_KEY, COORDINATOR, DOMAIN +from homeassistant.const import CONF_URL, CONF_VERIFY_SSL +from homeassistant.core import HomeAssistant @pytest.fixture(autouse=True) def auto_enable_custom_integrations(enable_custom_integrations): - yield + """Test helper.""" + return @pytest.fixture def mock_pfsense_client(): + """Test helper.""" client = AsyncMock() client.get_system_info.return_value = { "hostname": "router", @@ -70,9 +72,7 @@ def _entry(entry_id): async def _setup(hass, entry, client): entry.add_to_hass(hass) with ( - patch( - "custom_components.pfsense.pfSenseClient", return_value=client - ), + patch("custom_components.pfsense.pfSenseClient", return_value=client), patch("custom_components.pfsense.async_load_cache", return_value=None), patch("custom_components.pfsense.async_save_cache"), ): @@ -82,20 +82,20 @@ async def _setup(hass, entry, client): @pytest.mark.asyncio async def test_carp_sensor_on(hass: HomeAssistant, mock_pfsense_client): + """Test carp sensor on.""" mock_pfsense_client.get_carp_status.return_value = True await _setup(hass, _entry("carp_on"), mock_pfsense_client) coordinator = hass.data[DOMAIN]["carp_on"][COORDINATOR] assert coordinator.data["carp_status"] is True # Notices have no REST endpoint; that binary sensor no longer exists. - assert ( - hass.states.get("binary_sensor.router_local_pending_notices_present") is None - ) + assert hass.states.get("binary_sensor.router_local_pending_notices_present") is None assert hass.states.get("binary_sensor.router_local_carp_status") is not None @pytest.mark.asyncio async def test_carp_sensor_off(hass: HomeAssistant, mock_pfsense_client): + """Test carp sensor off.""" mock_pfsense_client.get_carp_status.return_value = False await _setup(hass, _entry("carp_off"), mock_pfsense_client) diff --git a/tests/test_config_flow.py b/tests/test_config_flow.py index 91814ec..0e8403f 100644 --- a/tests/test_config_flow.py +++ b/tests/test_config_flow.py @@ -3,9 +3,6 @@ from unittest.mock import AsyncMock, patch import pytest -from homeassistant.core import HomeAssistant -from homeassistant.const import CONF_URL, CONF_VERIFY_SSL -from homeassistant.data_entry_flow import FlowResultType from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.pfsense.const import ( @@ -15,6 +12,9 @@ DOMAIN, ) from custom_components.pfsense.pypfsense import PfSenseAuthError +from homeassistant.const import CONF_URL, CONF_VERIFY_SSL +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType def _client_mock(**overrides): @@ -31,6 +31,7 @@ def _client_mock(**overrides): @pytest.mark.asyncio async def test_form_user_success(hass: HomeAssistant): + """Test form user success.""" result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": "user"} ) @@ -60,6 +61,7 @@ async def test_form_user_success(hass: HomeAssistant): @pytest.mark.asyncio async def test_form_user_invalid_auth(hass: HomeAssistant): + """Test form user invalid auth.""" result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": "user"} ) @@ -76,6 +78,7 @@ async def test_form_user_invalid_auth(hass: HomeAssistant): @pytest.mark.asyncio async def test_reauth_flow_updates_key(hass: HomeAssistant): + """Test reauth flow updates key.""" entry = MockConfigEntry( domain=DOMAIN, version=3, @@ -104,6 +107,7 @@ async def test_reauth_flow_updates_key(hass: HomeAssistant): @pytest.mark.asyncio async def test_options_flow(hass: HomeAssistant): + """Test options flow.""" entry = MockConfigEntry( domain=DOMAIN, version=3, diff --git a/tests/test_device_tracker.py b/tests/test_device_tracker.py index 0080d35..434b455 100644 --- a/tests/test_device_tracker.py +++ b/tests/test_device_tracker.py @@ -1,3 +1,5 @@ +"""Tests for the pfSense device tracker.""" + from unittest.mock import MagicMock from custom_components.pfsense.device_tracker import lookup_mac diff --git a/tests/test_init.py b/tests/test_init.py index 4f77918..cf163bc 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -3,17 +3,18 @@ from unittest.mock import AsyncMock, patch import pytest -from homeassistant.const import CONF_URL, CONF_VERIFY_SSL -from homeassistant.core import HomeAssistant from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.pfsense import async_setup_entry, async_unload_entry from custom_components.pfsense.const import CONF_API_KEY, DOMAIN +from homeassistant.const import CONF_URL, CONF_VERIFY_SSL +from homeassistant.core import HomeAssistant @pytest.fixture(autouse=True) def auto_enable_custom_integrations(enable_custom_integrations): - yield + """Test helper.""" + return def _full_client_mock(): @@ -55,6 +56,7 @@ def _full_client_mock(): @pytest.mark.asyncio async def test_setup_and_unload_entry(hass: HomeAssistant): + """Test setup and unload entry.""" entry = MockConfigEntry( domain=DOMAIN, version=3, diff --git a/tests/test_pypfsense.py b/tests/test_pypfsense.py index 2d6f5c2..e664c1c 100644 --- a/tests/test_pypfsense.py +++ b/tests/test_pypfsense.py @@ -3,15 +3,15 @@ import re import aiohttp -import pytest from aioresponses import aioresponses +import pytest from custom_components.pfsense.pypfsense import ( Client, + PfSenseAPIError, PfSenseAuthError, PfSenseNotFoundError, PfSensePrivilegeError, - PfSenseAPIError, _build_telemetry, dict_get, ) @@ -32,11 +32,13 @@ def _envelope(data, code=200, status="ok", response_id="SUCCESS", message=""): @pytest.fixture async def client(): + """Test helper.""" async with aiohttp.ClientSession() as session: yield Client(BASE, "test-key", session, {"verify_ssl": False}) def test_dict_get(): + """Test dict get.""" data = {"a": {"b": [{"c": 1}]}, "n": {2: "x"}} assert dict_get(data, "a.b.0.c") == 1 assert dict_get(data, "n.2") == "x" @@ -45,18 +47,26 @@ def test_dict_get(): def test_base_url_strips_path(): + """Test base url strips path.""" c = Client("https://pf.example:8444/ui/", "k", object()) assert c._base == "https://pf.example:8444/api/v2" async def test_request_unwraps_data(client): + """Test request unwraps data.""" with aioresponses() as m: - m.get(f"{API}/system/hostname", payload=_envelope({"hostname": "pf", "domain": "lan"})) - assert await client._get("/system/hostname") == {"hostname": "pf", "domain": "lan"} + m.get( + f"{API}/system/hostname", + payload=_envelope({"hostname": "pf", "domain": "lan"}), + ) + assert await client._get("/system/hostname") == { + "hostname": "pf", + "domain": "lan", + } @pytest.mark.parametrize( - "code,exc", + ("code", "exc"), [ (401, PfSenseAuthError), (403, PfSensePrivilegeError), @@ -65,17 +75,21 @@ async def test_request_unwraps_data(client): ], ) async def test_error_codes_map_to_exceptions(client, code, exc): + """Test error codes map to exceptions.""" with aioresponses() as m: m.get( f"{API}/system/hostname", status=code, - payload=_envelope([], code=code, status="err", response_id="X", message="nope"), + payload=_envelope( + [], code=code, status="err", response_id="X", message="nope" + ), ) with pytest.raises(exc): await client._get("/system/hostname") async def test_get_system_info_merges_endpoints(client): + """Test get system info merges endpoints.""" with aioresponses() as m: m.get( f"{API}/status/system", @@ -98,6 +112,7 @@ async def test_get_system_info_merges_endpoints(client): async def test_carp_status_reduces_to_bool(client): + """Test carp status reduces to bool.""" with aioresponses() as m: m.get( f"{API}/status/carp", @@ -114,16 +129,24 @@ async def test_carp_status_reduces_to_bool(client): def _patch_body(m): return next( - r for (method, url), reqs in m.requests.items() - for r in reqs if method == "PATCH" + r + for (method, url), reqs in m.requests.items() + for r in reqs + if method == "PATCH" ).kwargs["json"] async def test_disable_filter_rule_patches_then_applies(client): + """Test disable filter rule patches then applies.""" rules = [ {"id": 4, "tracker": 111, "disabled": False, "descr": "r"}, - {"id": 5, "tracker": 222, "disabled": False, "descr": "r2", - "statetype": "keep state"}, + { + "id": 5, + "tracker": 222, + "disabled": False, + "descr": "r2", + "statetype": "keep state", + }, ] with aioresponses() as m: m.get(f"{API}/firewall/rules", payload=_envelope(rules)) @@ -134,6 +157,7 @@ async def test_disable_filter_rule_patches_then_applies(client): async def test_disable_filter_rule_backfills_empty_statetype(client): + """Test disable filter rule backfills empty statetype.""" # pfSense's GUI writes ````; a bare disabled PATCH then # fails FIELD_EMPTY_NOT_ALLOWED, so the client re-sends the default. rules = [{"id": 5, "tracker": 222, "disabled": False, "statetype": ""}] @@ -150,6 +174,7 @@ async def test_disable_filter_rule_backfills_empty_statetype(client): async def test_kill_states_for_rule_resolves_alias_to_prefix(client): + """Test kill states for rule resolves alias to prefix.""" rule = {"source": "kids", "destination": "any"} aliases = [{"name": "kids", "type": "network", "address": ["10.0.10.0/24"]}] with aioresponses() as m: @@ -171,6 +196,7 @@ async def test_kill_states_for_rule_resolves_alias_to_prefix(client): async def test_kill_states_for_rule_skips_unresolvable_endpoints(client): + """Test kill states for rule skips unresolvable endpoints.""" # ``any`` / ``(self)`` / a /25 network have no usable prefix -> no request. rule = {"source": "any", "destination": "(self)"} with aioresponses() as m: @@ -179,6 +205,7 @@ async def test_kill_states_for_rule_skips_unresolvable_endpoints(client): async def test_build_telemetry_shape(): + """Test build telemetry shape.""" system = { "cpu_usage": 12.5, "cpu_count": 4, @@ -193,9 +220,7 @@ async def test_build_telemetry_shape(): {"name": "lan", "descr": "LAN", "inbytes": 5}, ] gateways = [{"name": "WAN_DHCP", "delay": 1.2, "status": "online"}] - ovpn = [ - {"vpnid": 1, "name": "S", "conns": [{"bytes_recv": 100, "bytes_sent": 50}]} - ] + ovpn = [{"vpnid": 1, "name": "S", "conns": [{"bytes_recv": 100, "bytes_sent": 50}]}] t = _build_telemetry(system, interfaces, gateways, ovpn) assert t["wan_ip"] == "1.2.3.4" assert t["cpu"]["used_percent"] == 12.5 diff --git a/tests/test_sensor.py b/tests/test_sensor.py index 9a895cf..7a5fb43 100644 --- a/tests/test_sensor.py +++ b/tests/test_sensor.py @@ -1,5 +1,8 @@ +"""Tests for the pfSense sensors.""" + +from unittest.mock import MagicMock, PropertyMock, patch + import pytest -from unittest.mock import MagicMock, patch, PropertyMock from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.pfsense.const import DOMAIN @@ -9,6 +12,7 @@ @pytest.fixture def mock_coordinator(): + """Test helper.""" coord = MagicMock() coord.data = { "telemetry": { @@ -31,6 +35,7 @@ def mock_coordinator(): return_value="pfSense", ) def test_openvpn_sensor(mock_name, mock_uid, mock_coordinator): + """Test openvpn sensor.""" config_entry = MockConfigEntry(domain=DOMAIN) desc = SensorEntityDescription( key="telemetry.openvpn.servers.1.status", name="VPN Status" diff --git a/tests/test_switch.py b/tests/test_switch.py index f729691..baf6e88 100644 --- a/tests/test_switch.py +++ b/tests/test_switch.py @@ -3,14 +3,15 @@ from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch import pytest -from homeassistant.components.switch import SwitchEntityDescription from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.pfsense.switch import PfSenseServiceSwitch +from homeassistant.components.switch import SwitchEntityDescription @pytest.fixture def mock_coordinator(): + """Test helper.""" coord = MagicMock() coord.data = { "services": [ @@ -33,6 +34,7 @@ def mock_coordinator(): return_value="pfSense", ) async def test_service_switch_turn_off(mock_name, mock_uid, mock_coordinator): + """Test service switch turn off.""" desc = SwitchEntityDescription(key="service.unbound.status", name="unbound") switch = PfSenseServiceSwitch(MockConfigEntry(), mock_coordinator, desc)