style: ruff format and the lint rules of the driver standard
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff cannot fix comes in the next commit.
This commit is contained in:
1 parent
8145a9f479
commit
b23967417c
3 files changed
+54
-26
No files matched your search
+39
-20
@@ -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({
|
||||
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
|
||||
|
||||
|
||||
+8
-1
@@ -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"]
|
||||
@@ -6,7 +6,6 @@ import pytest
|
||||
from fritzconnection.core.exceptions import (
|
||||
FritzActionError,
|
||||
FritzArrayIndexError,
|
||||
FritzServiceError,
|
||||
)
|
||||
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in new issue
Block a user