Fix linting errors

python/nav/dhcp/generic_metrics.py

@@ -10,8 +10,9 @@
 class DhcpMetricKey(Enum):
     TOTAL = "total"  # total addresses managed by dhcp
     ASSIGNED = "assigned"  # assigned addresses
+
     def __str__(self):
-        return self.value # graphite key
+        return self.value  # graphite key
 
 
 @dataclass(frozen=True)
@@ -42,9 +43,7 @@ def fetch_metrics(self) -> Iterator[DhcpMetric]:
         raise NotImplementedError
 
     def fetch_metrics_to_graphite(
-            self,
-            host=CONFIG.get("carbon", "host"),
-            port=CONFIG.getint("carbon", "port")
+        self, host=CONFIG.get("carbon", "host"), port=CONFIG.getint("carbon", "port")
     ):
         """
         Fetch metrics describing total amount of addresses

python/nav/dhcp/kea_metrics.py

@@ -13,6 +13,7 @@
 
 _logger = logging.getLogger(__name__)
 
+
 class KeaDhcpMetricSource(DhcpMetricSource):
     """
     Communicates with a Kea Control Agent and fetches metrics for each
@@ -105,7 +106,6 @@ def fetch_metrics(self) -> list[DhcpMetric]:
                         )
                         metrics.append(metric)
 
-
         newsubnets = _subnets_of_config(self._fetch_config(s), self.dhcp_version)
         if sorted(subnets) != sorted(newsubnets):
             _logger.error(
@@ -116,7 +116,6 @@ def fetch_metrics(self) -> list[DhcpMetric]:
 
         return metrics
 
-
     def _fetch_config(self, session: requests.Session) -> dict:
         """
         Returns the current config of the Kea DHCP server that the Kea
@@ -133,7 +132,7 @@ def _fetch_config(self, session: requests.Session) -> dict:
             except KeyError as err:
                 raise KeaException(
                     "Unrecognizable response to the 'config-get' request",
-                    {"Response": response}
+                    {"Response": response},
                 ) from err
         return self.dhcp_config
 
@@ -152,7 +151,6 @@ def _fetch_config_hash(self, session: requests.Session) -> Optional[str]:
             _logger.debug(str(err))
             return None
 
-
     def _send_query(self, session: requests.Session, command: str, **kwargs) -> dict:
         """
         Returns the response from the Kea Control Agent to the query
@@ -199,27 +197,25 @@ def _send_query(self, session: requests.Session, command: str, **kwargs) -> dict
             raise KeaException(
                 "Server does not look like a Kea Control Agent; "
                 "expected response content to be JSON",
-                request_summary
+                request_summary,
             ) from err
         except RequestException as err:
             raise KeaException(
-                "HTTP-related error during request to server",
-                request_summary
+                "HTTP-related error during request to server", request_summary
             ) from err
 
         if not isinstance(responses, list):
             # See https://kea.readthedocs.io/en/kea-2.6.0/arm/ctrl-channel.html#control-agent-command-response-format
             raise KeaException(
-                "Invalid response; server has likely rejected a query",
-                request_summary
+                "Invalid response; server has likely rejected a query", request_summary
             )
         if not (len(responses) == 1 and "result" in responses[0]):
             # `post-data` queries *one* service. Thus `responses` should contain *one* response.
             raise KeaException(
                 "Server does not look like a Kea Control Agent; "
                 "expected response content to be a JSON list "
                 "of a single object that has 'result' as one of its keys. ",
-                request_summary
+                request_summary,
             )
         request_summary["Validity"] = "Valid Kea response"
 
@@ -240,7 +236,6 @@ def _send_query(self, session: requests.Session, command: str, **kwargs) -> dict
             raise KeaConflict(details=request_summary)
         raise KeaException("Kea returned an unkown status response", request_summary)
 
-
     def _parsetime(self, timestamp: str) -> int:
         """Parse the timestamp string used in Kea's timeseries into unix time"""
         fmt = "%Y-%m-%d %H:%M:%S.%f"
@@ -262,8 +257,7 @@ def _subnets_of_config(config: dict, ip_version: int) -> list[tuple[int, IP]]:
         prefix = subnet.get("subnet", None)
         if id is None or prefix is None:
             _logger.warning(
-                "id or prefix missing from a subnet's configuration: %r",
-                subnet
+                "id or prefix missing from a subnet's configuration: %r", subnet
             )
             continue
         subnets.append((id, IP(prefix)))
@@ -272,6 +266,7 @@ def _subnets_of_config(config: dict, ip_version: int) -> list[tuple[int, IP]]:
 
 class KeaException(GeneralException):
     """Error related to interaction with a Kea Control Agent"""
+
     def __init__(self, message: str = "", details: dict[str, str] = {}):
         self.message = message
         self.details = details
@@ -284,20 +279,31 @@ def __str__(self) -> str:
             message = f": {self.message}"
         if self.details:
             details = "\nDetails:\n"
-            details += "\n".join(f"{label}: {info}" for label, info in self.details.items())
+            details += "\n".join(
+                f"{label}: {info}" for label, info in self.details.items()
+            )
         return "".join([doc, message, details])
 
+
 class KeaError(KeaException):
     """Kea failed during command processing"""
+
+
 class KeaUnsupported(KeaException):
     """Unsupported command"""
+
+
 class KeaEmpty(KeaException):
     """Requested resource not found"""
+
+
 class KeaConflict(KeaException):
     """Kea failed to apply requested changes due to conflicts with its server state"""
 
+
 class KeaStatus(IntEnum):
     """Status of a response sent from a Kea Control Agent"""
+
     SUCCESS = 0
     ERROR = 1
     UNSUPPORTED = 2

tests/unittests/dhcp/kea_metrics_test.py

@@ -120,7 +120,7 @@ def test_all_responses_is_empty_but_valid_should_yield_no_metrics(
 
 
 def test_response_with_http_error_status_code_should_cause_KeaException_to_be_raised(
-        valid_dhcp4, responsequeue
+    valid_dhcp4, responsequeue
 ):
     config, statistics, _ = valid_dhcp4
     responsequeue.autofill("dhcp4", config, statistics, attrs={"status_code": 403})
@@ -537,6 +537,7 @@ def kearesponse(val, status=KeaStatus.SUCCESS):
 ]
     '''
 
+
 @pytest.fixture(autouse=True)
 def responsequeue(monkeypatch):
     """
@@ -645,12 +646,19 @@ def add_command_response(command_name, text, attrs={}):
     def clear_command_responses():
         command_responses.clear()
 
-    def autofill_command_responses(expected_service, config=None, statistics=None, attrs={}):
+    def autofill_command_responses(
+        expected_service, config=None, statistics=None, attrs={}
+    ):
         def config_get_response(arguments, service):
-            assert service == [expected_service], f"KeaDhcpSource for service [{expected_service}] should not send requests to {service}"
+            assert service == [
+                expected_service
+            ], f"KeaDhcpSource for service [{expected_service}] should not send requests to {service}"
             return kearesponse(config)
+
         def statistic_get_response(arguments, service):
-            assert service == [expected_service], f"KeaDhcpSource for service [{expected_service}] should not send requests to {service}"
+            assert service == [
+                expected_service
+            ], f"KeaDhcpSource for service [{expected_service}] should not send requests to {service}"
             return kearesponse({arguments["name"]: statistics[arguments["name"]]})
 
         if config is not None: