[typing] Set of typing fixes

monitoring/monitorlib/idempotency.py

@@ -46,7 +46,8 @@ def _set_responses(responses: dict[str, Response]) -> bytes:
                 oldest_id = request_id
                 oldest_timestamp = t
 
-        del responses[oldest_id]
+        if oldest_id:
+            del responses[oldest_id]
     return s.encode("utf-8")
 
 

monitoring/monitorlib/infrastructure.py

@@ -37,11 +37,15 @@ def issue_token(self, intended_audience: str, scopes: list[str]) -> str:
 
         raise NotImplementedError()
 
-    def get_headers(self, url: str, scopes: list[str] = None) -> dict[str, str]:
+    def get_headers(self, url: str, scopes: list[str] | None = None) -> dict[str, str]:
         if scopes is None:
             scopes = ALL_SCOPES
         scopes = [s.value if isinstance(s, Enum) else s for s in scopes]
         intended_audience = urllib.parse.urlparse(url).hostname
+
+        if not intended_audience:
+            return {}
+
         scope_string = " ".join(scopes)
         if intended_audience not in self._tokens:
             self._tokens[intended_audience] = {}
@@ -57,8 +61,9 @@ def get_headers(self, url: str, scopes: list[str] = None) -> dict[str, str]:
         return {"Authorization": "Bearer " + token}
 
     def add_headers(self, request: requests.PreparedRequest, scopes: list[str]):
-        for k, v in self.get_headers(request.url, scopes).items():
-            request.headers[k] = v
+        if request.url:
+            for k, v in self.get_headers(request.url, scopes).items():
+                request.headers[k] = v
 
     def get_sub(self) -> str | None:
         """Retrieve `sub` claim from one of the existing tokens"""
@@ -91,7 +96,7 @@ def __init__(
 
         self._prefix_url = prefix_url[0:-1] if prefix_url[-1] == "/" else prefix_url
         self.auth_adapter = auth_adapter
-        self.default_scopes = None
+        self.default_scopes: list[str] | None = None
         self.timeout_seconds = timeout_seconds or CLIENT_TIMEOUT
 
     # Overrides method on requests.Session
@@ -117,7 +122,7 @@ def adjust_request_kwargs(self, kwargs):
             def auth(
                 prepared_request: requests.PreparedRequest,
             ) -> requests.PreparedRequest:
-                if scopes:
+                if scopes and self.auth_adapter:
                     self.auth_adapter.add_headers(prepared_request, scopes)
                 return prepared_request
 
@@ -126,15 +131,21 @@ def auth(
             kwargs["timeout"] = self.timeout_seconds
         return kwargs
 
-    def request(self, method, url, **kwargs):
+    def request(self, method, url, *args, **kwargs):
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(kwargs)
 
-        return super().request(method, url, **kwargs)
+        return super().request(method, url, *args, **kwargs)
 
     def get_prefix_url(self):
         return self._prefix_url
 
+    def get(self, *args, **kwargs):
+        return super().get(*args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        return super().delete(*args, **kwargs)
+
 
 class AsyncUTMTestSession:
     """
@@ -155,15 +166,16 @@ def __init__(
 
         self._prefix_url = prefix_url[0:-1] if prefix_url[-1] == "/" else prefix_url
         self.auth_adapter = auth_adapter
-        self.default_scopes = None
+        self.default_scopes: list[str] | None = None
         self.timeout_seconds = timeout_seconds or CLIENT_TIMEOUT
 
     async def build_session(self):
         self._client = ClientSession()
 
     def close(self):
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(self._client.close())
+        if self._client:
+            loop = asyncio.get_event_loop()
+            loop.run_until_complete(self._client.close())
 
     def adjust_request_kwargs(self, url, method, kwargs):
         if self.auth_adapter:
@@ -196,6 +208,10 @@ async def put(self, url, **kwargs):
         url = self._prefix_url + url
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(url, "PUT", kwargs)
+
+        if not self._client:
+            raise ValueError("Client is not ready")
+
         async with self._client.put(url, **kwargs) as response:
             return (
                 response.status,
@@ -208,6 +224,10 @@ async def get(self, url, **kwargs):
         url = self._prefix_url + url
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(url, "GET", kwargs)
+
+        if not self._client:
+            raise ValueError("Client is not ready")
+
         async with self._client.get(url, **kwargs) as response:
             return (
                 response.status,
@@ -220,6 +240,10 @@ async def post(self, url, **kwargs):
         url = self._prefix_url + url
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(url, "POST", kwargs)
+
+        if not self._client:
+            raise ValueError("Client is not ready")
+
         async with self._client.post(url, **kwargs) as response:
             return (
                 response.status,
@@ -232,6 +256,10 @@ async def delete(self, url, **kwargs):
         url = self._prefix_url + url
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(url, "DELETE", kwargs)
+
+        if not self._client:
+            raise ValueError("Client is not ready")
+
         async with self._client.delete(url, **kwargs) as response:
             return (
                 response.status,

monitoring/monitorlib/kml/f3548v21.py

@@ -21,14 +21,14 @@
 from monitoring.monitorlib.scd import priority_of
 
 
-def full_op_intent(op_intent: OperationalIntent) -> kml.Folder:
+def full_op_intent(op_intent: OperationalIntent):
     """Render operational intent information into Placemarks in a KML folder."""
     ref = op_intent.reference
     details = op_intent.details
     name = f"{ref.manager}'s P{priority_of(details)} {ref.state.value} {ref.id}[{ref.version}] @ {ref.ovn}"
     folder = kml.Folder(kml.name(name))
     if "volumes" in details:
-        for i, v4_f3548 in enumerate(details.volumes):
+        for i, v4_f3548 in enumerate(details.volumes or []):
             v4 = Volume4D.from_f3548v21(v4_f3548)
             folder.append(
                 make_placemark_from_volume(
@@ -38,7 +38,7 @@ def full_op_intent(op_intent: OperationalIntent) -> kml.Folder:
                 )
             )
     if "off_nominal_volumes" in details:
-        for i, v4_f3548 in enumerate(details.off_nominal_volumes):
+        for i, v4_f3548 in enumerate(details.off_nominal_volumes or []):
             v4 = Volume4D.from_f3548v21(v4_f3548)
             folder.append(
                 make_placemark_from_volume(
@@ -53,8 +53,12 @@ def full_op_intent(op_intent: OperationalIntent) -> kml.Folder:
 def op_intent_refs_query(
     req: QueryOperationalIntentReferenceParameters,
     resp: QueryOperationalIntentReferenceResponse,
-) -> kml.Placemark:
+):
     """Render the area of interest and response from an operational intent references query into a KML Placemark."""
+
+    if not req.area_of_interest:
+        raise ValueError("req.area_of_interest is not defined")
+
     v4 = Volume4D.from_f3548v21(req.area_of_interest)
     items = "".join(
         f"<li>{oi.manager}'s {oi.state.value} {oi.id}[{oi.version}]</li>"
@@ -68,7 +72,7 @@ def op_intent_refs_query(
     )
 
 
-def f3548v21_styles() -> list[kml.Style]:
+def f3548v21_styles() -> list:
     """Provides KML styles according to F3548-21 operational intent states."""
     return [
         kml.Style(

monitoring/monitorlib/kml/flight_planning.py

@@ -15,17 +15,15 @@
 )
 
 
-def upsert_flight_plan(
-    req: UpsertFlightPlanRequest, resp: UpsertFlightPlanResponse
-) -> kml.Folder:
+def upsert_flight_plan(req: UpsertFlightPlanRequest, resp: UpsertFlightPlanResponse):
     """Render a flight planning action into a KML folder."""
     basic_info = req.flight_plan.basic_information
     folder = kml.Folder(
         kml.name(
             f"Activity {resp.planning_result.value}, flight {resp.flight_plan_status.value}"
         )
     )
-    for i, v4_flight_planning in enumerate(basic_info.area):
+    for i, v4_flight_planning in enumerate(basic_info.area or []):
         v4 = Volume4D.from_flight_planning_api(v4_flight_planning)
         folder.append(
             make_placemark_from_volume(
@@ -37,7 +35,7 @@ def upsert_flight_plan(
     return folder
 
 
-def flight_planning_styles() -> list[kml.Style]:
+def flight_planning_styles() -> list:
     """Provides KML styles with names in the form {FlightPlanState}_{AirspaceUsageState}."""
     return [
         kml.Style(

monitoring/monitorlib/kml/generation.py

@@ -53,7 +53,7 @@ def make_placemark_from_volume(
     name: str | None = None,
     style_url: str | None = None,
     description: str | None = None,
-) -> kml.Placemark:
+):
     if "outline_polygon" in v4.volume and v4.volume.outline_polygon:
         vertices = v4.volume.outline_polygon.vertices
     elif "outline_circle" in v4.volume and v4.volume.outline_circle:
@@ -70,6 +70,9 @@ def make_placemark_from_volume(
     else:
         raise NotImplementedError("Volume footprint type not supported")
 
+    if not vertices:
+        raise NotImplementedError("No vertices found")
+
     # Create placemark
     args = []
     if name is not None:
@@ -121,7 +124,7 @@ def make_placemark_from_volume(
                     _altitude_mode_of(
                         v4.volume.altitude_lower
                         if v4.volume.altitude_lower
-                        else AltitudeDatum.SFC
+                        else Altitude(reference=AltitudeDatum.SFC)
                     )
                 ),
                 kml.outerBoundaryIs(
@@ -142,7 +145,7 @@ def make_placemark_from_volume(
                     _altitude_mode_of(
                         v4.volume.altitude_upper
                         if v4.volume.altitude_upper
-                        else AltitudeDatum.SFC
+                        else Altitude(reference=AltitudeDatum.SFC)
                     )
                 ),
                 kml.outerBoundaryIs(
@@ -160,6 +163,8 @@ def make_placemark_from_volume(
     # We can only create the sides of the volume if the altitude references are the same
     if (
         make_sides
+        and v4.volume.altitude_lower
+        and v4.volume.altitude_upper
         and v4.volume.altitude_lower.reference == v4.volume.altitude_upper.reference
     ):
         indices = list(range(len(vertices)))
@@ -187,7 +192,7 @@ def make_placemark_from_volume(
     return placemark
 
 
-def query_styles() -> list[kml.Style]:
+def query_styles() -> list:
     """Provides KML styles for query areas."""
     return [
         kml.Style(

monitoring/monitorlib/locality.py

@@ -3,7 +3,6 @@
 import inspect
 import sys
 from abc import ABC, abstractmethod
-from typing import TypeVar
 
 LocalityCode = str
 """Case-sensitive string naming a subclass of the Locality base class"""
@@ -42,7 +41,7 @@ def __str__(self):
         return self.__class__.__name__
 
     @staticmethod
-    def from_locale(locality_code: LocalityCode) -> LocalityType:
+    def from_locale(locality_code: LocalityCode) -> Locality:
         current_module = sys.modules[__name__]
         for name, obj in inspect.getmembers(current_module, inspect.isclass):
             if issubclass(obj, Locality) and obj != Locality:
@@ -53,9 +52,6 @@ def from_locale(locality_code: LocalityCode) -> LocalityType:
         )
 
 
-LocalityType = TypeVar("LocalityType", bound=Locality)
-
-
 class Switzerland(Locality):
     @classmethod
     def locality_code(cls) -> str:

monitoring/monitorlib/multiprocessing.py

@@ -2,6 +2,7 @@
 import multiprocessing
 import multiprocessing.shared_memory
 from collections.abc import Callable
+from multiprocessing.synchronize import RLock as RLockT
 from typing import Any
 
 
@@ -24,7 +25,7 @@ class SynchronizedValue:
     SIZE_BYTES = 4
     """Number of bytes at the beginning of the memory buffer dedicated to defining the size of the content."""
 
-    _lock: multiprocessing.RLock
+    _lock: RLockT
     _shared_memory: multiprocessing.shared_memory.SharedMemory
     _encoder: Callable[[Any], bytes]
     _decoder: Callable[[bytes], Any]
@@ -33,7 +34,7 @@ class SynchronizedValue:
     def __init__(
         self,
         initial_value,
-        capacity_bytes: int = 10e6,
+        capacity_bytes: int = 10000000,
         encoder: Callable[[Any], bytes] | None = None,
         decoder: Callable[[bytes], Any] | None = None,
     ):