core/homeassistant/components/opnsense/device_tracker.py

67 lines
2.2 KiB
Python
Raw Normal View History

Add OPNSense device tracker (#26834) * Add OPNSense device_tracker This commit adds a new component for using an OPNSense router as a device tracker. It uses pyopnsense to query the api to look at the arptable for a list of devices on the network. * Run black formatting locally to appease azure * Apply suggestions from code review Co-Authored-By: Fabian Affolter <mail@fabian-affolter.ch> * Fix issues identified during code review This commit updates several issues found in the module during code review. * Update homeassistant/components/opnsense/__init__.py Co-Authored-By: Fabian Affolter <mail@fabian-affolter.ch> * Update CODEOWNERS for recent changes * Fix lint * Apply suggestions from code review Co-Authored-By: Martin Hjelmare <marhje52@kth.se> * More fixes from review comments This commit fixes several issues from review comments, including abandoning all the use of async code. This also completely reworks the tests to be a bit clearer. * Revert tests to previous format * Add device detection to opnsense device_tracker test This commit adds actual device detection to the unit test for the setup test. A fake api response is added to mocks for both api clients so that they will register devices as expected and asserts are added for that. The pyopnsense import is moved from the module level to be runtime in the class. This was done because it was the only way to make the MockDependency() call work as expected. * Rerun black * Fix lint * Move import back to module level * Return false on configuration errors in setup This commit updates the connection logic to return false if we're unable to connect to the configured OPNsense API endpoint for any reason. Previously we would not catch if an endpoint was incorrectly configured until we first tried to use it. In this case it would raise an unhandled exception. To handle this more gracefully this adds an api call early in the setup and catches any exception raised by that so we can return False to indicate the setup failed. * Update tests * Add pyopnsense to test requirements * Rerun gen_requirements script * Fix failing isort lint job step Since opening the PR originally yet another lint/style checker was added which failed the PR in CI. This commit makes the adjustments to have this pass the additional tool's checks. * Fix comment * Update manifest.json Co-authored-by: Fabian Affolter <mail@fabian-affolter.ch> Co-authored-by: Martin Hjelmare <marhje52@gmail.com> Co-authored-by: Pascal Vizeli <pascal.vizeli@syshack.ch>
2020-01-29 15:20:43 +00:00
"""Device tracker support for OPNSense routers."""
import logging
from homeassistant.components.device_tracker import DeviceScanner
from homeassistant.components.opnsense import CONF_TRACKER_INTERFACE, OPNSENSE_DATA
_LOGGER = logging.getLogger(__name__)
async def async_get_scanner(hass, config, discovery_info=None):
"""Configure the OPNSense device_tracker."""
interface_client = hass.data[OPNSENSE_DATA]["interfaces"]
scanner = OPNSenseDeviceScanner(
interface_client, hass.data[OPNSENSE_DATA][CONF_TRACKER_INTERFACE]
)
return scanner
class OPNSenseDeviceScanner(DeviceScanner):
"""This class queries a router running OPNsense."""
def __init__(self, client, interfaces):
"""Initialize the scanner."""
self.last_results = {}
self.client = client
self.interfaces = interfaces
def _get_mac_addrs(self, devices):
"""Create dict with mac address keys from list of devices."""
out_devices = {}
for device in devices:
if not self.interfaces:
out_devices[device["mac"]] = device
elif device["intf_description"] in self.interfaces:
out_devices[device["mac"]] = device
return out_devices
def scan_devices(self):
"""Scan for new devices and return a list with found device IDs."""
self.update_info()
return list(self.last_results)
def get_device_name(self, device):
"""Return the name of the given device or None if we don't know."""
if device not in self.last_results:
return None
hostname = self.last_results[device].get("hostname") or None
return hostname
def update_info(self):
"""Ensure the information from the OPNSense router is up to date.
Return boolean if scanning successful.
"""
devices = self.client.get_arp()
self.last_results = self._get_mac_addrs(devices)
def get_extra_attributes(self, device):
"""Return the extra attrs of the given device."""
if device not in self.last_results:
return None
mfg = self.last_results[device].get("manufacturer")
if mfg:
return {"manufacturer": mfg}
return {}