diff --git a/napalm_fritzbox/fritzbox.py b/napalm_fritzbox/fritzbox.py index 8f37546..fcd6814 100644 --- a/napalm_fritzbox/fritzbox.py +++ b/napalm_fritzbox/fritzbox.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -44,7 +43,6 @@ from fritzconnection.core.exceptions import ( FritzServiceError, ) from napalm.base.exceptions import ConnectionClosedException, ConnectionException - from napalm_device_types import FingerprintRule, ResidentialGatewayDriver from napalm_device_types.models import ( HostDict, @@ -345,10 +343,18 @@ class FritzBoxDriver(ResidentialGatewayDriver): result["dsl_downstream_max_rate"] = int(_to_float(info.get("NewDownstreamMaxRate"))) result["dsl_upstream_max_rate"] = int(_to_float(info.get("NewUpstreamMaxRate"))) # SNR margins and attenuation are stored as dB×10 by TR-064 - result["dsl_downstream_noise_margin"] = int(_to_float(info.get("NewDownstreamNoiseMargin"))) - result["dsl_upstream_noise_margin"] = int(_to_float(info.get("NewUpstreamNoiseMargin"))) - result["dsl_downstream_attenuation"] = int(_to_float(info.get("NewDownstreamAttenuation"))) - result["dsl_upstream_attenuation"] = int(_to_float(info.get("NewUpstreamAttenuation"))) + result["dsl_downstream_noise_margin"] = int( + _to_float(info.get("NewDownstreamNoiseMargin")) + ) + result["dsl_upstream_noise_margin"] = int( + _to_float(info.get("NewUpstreamNoiseMargin")) + ) + result["dsl_downstream_attenuation"] = int( + _to_float(info.get("NewDownstreamAttenuation")) + ) + result["dsl_upstream_attenuation"] = int( + _to_float(info.get("NewUpstreamAttenuation")) + ) except _OPTIONAL_SERVICE_ERRORS: pass @@ -455,7 +461,12 @@ class FritzBoxDriver(ResidentialGatewayDriver): ext_ip = self._call(svc, "GetExternalIPAddress").get("NewExternalIPAddress", "") wan_service_used = svc break - except (FritzArrayIndexError, FritzServiceError, FritzActionError, FritzConnectionException): + except ( + FritzArrayIndexError, + FritzServiceError, + FritzActionError, + FritzConnectionException, + ): continue except Exception: continue @@ -494,7 +505,9 @@ class FritzBoxDriver(ResidentialGatewayDriver): index = 0 while True: try: - entry = self._call(service, "GetGenericPortMappingEntry", NewPortMappingIndex=index) + entry = self._call( + service, "GetGenericPortMappingEntry", NewPortMappingIndex=index + ) except (FritzArrayIndexError, FritzConnectionException): break @@ -530,15 +543,21 @@ class FritzBoxDriver(ResidentialGatewayDriver): except (FritzArrayIndexError, FritzConnectionException, FritzActionError): break proto_num = int(ph.get("NewProtocol", 0) or 0) - proto = "TCP6" if proto_num == 6 else ("UDP6" if proto_num == 17 else str(proto_num)) - forwards.append({ - "name": ph.get("NewPinholeDescription", ""), - "protocol": proto, - "external_port": int(ph.get("NewRemotePort", 0) or 0), - "internal_ip": ph.get("NewInternalIPAddress", ""), - "internal_port": int(ph.get("NewInternalPort", 0) or 0), - "enabled": True, - }) + proto = ( + "TCP6" + if proto_num == 6 + else ("UDP6" if proto_num == 17 else str(proto_num)) + ) + forwards.append( + { + "name": ph.get("NewPinholeDescription", ""), + "protocol": proto, + "external_port": int(ph.get("NewRemotePort", 0) or 0), + "internal_ip": ph.get("NewInternalIPAddress", ""), + "internal_port": int(ph.get("NewInternalPort", 0) or 0), + "enabled": True, + } + ) index += 1 except _OPTIONAL_SERVICE_ERRORS: pass @@ -552,9 +571,9 @@ class FritzBoxDriver(ResidentialGatewayDriver): ``Hosts1.GetHostNumberOfEntries`` entries. """ try: - count = int(self._call("Hosts1", "GetHostNumberOfEntries").get( - "NewHostNumberOfEntries", 0 - ) or 0) + count = int( + self._call("Hosts1", "GetHostNumberOfEntries").get("NewHostNumberOfEntries", 0) or 0 + ) except _OPTIONAL_SERVICE_ERRORS: return [] @@ -631,9 +650,9 @@ class FritzBoxDriver(ResidentialGatewayDriver): for service in self._wlan_services(): try: ssid = self._call(service, "GetInfo").get("NewSSID", "") - total = int(self._call(service, "GetTotalAssociations").get( - "NewTotalAssociations", 0 - ) or 0) + total = int( + self._call(service, "GetTotalAssociations").get("NewTotalAssociations", 0) or 0 + ) except _OPTIONAL_SERVICE_ERRORS: continue @@ -680,9 +699,9 @@ class FritzBoxDriver(ResidentialGatewayDriver): continue try: - clients = int(self._call(service, "GetTotalAssociations").get( - "NewTotalAssociations", 0 - ) or 0) + clients = int( + self._call(service, "GetTotalAssociations").get("NewTotalAssociations", 0) or 0 + ) except _OPTIONAL_SERVICE_ERRORS: clients = 0 diff --git a/pyproject.toml b/pyproject.toml index 7a73892..0c30675 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,14 @@ include = ["napalm_fritzbox*"] [tool.ruff] line-length = 100 -target-version = "py39" +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] + +[tool.ruff.lint.per-file-ignores] +# Recorded device output keeps its lines. +"tests/**" = ["E501"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index f78d668..30de2e5 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -6,7 +6,6 @@ import pytest from fritzconnection.core.exceptions import ( FritzActionError, FritzArrayIndexError, - FritzServiceError, ) from napalm.base.exceptions import ConnectionClosedException, ConnectionException @@ -135,9 +134,9 @@ DSL_INFO = { "NewDownstreamMaxRate": 120000, "NewUpstreamMaxRate": 50000, "NewDownstreamNoiseMargin": 120, # 12.0 dB - "NewUpstreamNoiseMargin": 80, # 8.0 dB + "NewUpstreamNoiseMargin": 80, # 8.0 dB "NewDownstreamAttenuation": 150, # 15.0 dB - "NewUpstreamAttenuation": 90, # 9.0 dB + "NewUpstreamAttenuation": 90, # 9.0 dB } DSL_STATS = { @@ -639,7 +638,8 @@ class TestPPPoEInterface: class TestAutoMaxBitRate: def test_lan_speed_zero_when_auto(self, driver): driver.fc.call_action.side_effect = lambda svc, act, **kw: ( - LAN_INFO_AUTO_RATE if (svc, act) == ("LANEthernetInterfaceConfig1", "GetInfo") + LAN_INFO_AUTO_RATE + if (svc, act) == ("LANEthernetInterfaceConfig1", "GetInfo") else _call_action(svc, act, **kw) ) interfaces = driver.get_interfaces() @@ -690,7 +690,8 @@ class TestGetSystemConfig: def test_firmware_not_available_when_none(self, driver): driver.fc.call_action.side_effect = lambda svc, act, **kw: ( - UI_INFO_NO_UPDATE if (svc, act) == ("UserInterface1", "X_AVM-DE_GetInfo") + UI_INFO_NO_UPDATE + if (svc, act) == ("UserInterface1", "X_AVM-DE_GetInfo") else _call_action(svc, act, **kw) ) assert driver.get_system_config()["firmware_update_available"] is False @@ -730,8 +731,9 @@ class TestIPv6Pinholes: def test_pinholes_are_not_reported_as_nat_translations(self, driver): """A pinhole opens the firewall for a globally routable address; there is nothing to translate.""" - assert all(":" not in t["inside_local"].rsplit(":", 1)[0] - for t in driver.get_nat_translations()) + assert all( + ":" not in t["inside_local"].rsplit(":", 1)[0] for t in driver.get_nat_translations() + ) def test_no_pinholes_without_ipv6_service(self, driver): del driver.fc.services["WANIPv6Firewall1"]