[mock_uss] Set of typing fixes

Author: the-glu

.basedpyright/baseline.json

@@ -10,7 +10,7 @@
     KEY_TRACER_OUTPUT_FOLDER,
 )
 from monitoring.mock_uss.tracer.observation_areas import ObservationAreaID
-from monitoring.mock_uss.tracer.tracerlog import Logger
+from monitoring.mock_uss.tracer.tracerlog import DummyLogger, Logger
 from monitoring.monitorlib import infrastructure
 from monitoring.monitorlib.auth import make_auth_adapter
 from monitoring.monitorlib.fetch import scd
@@ -23,7 +23,7 @@
 scd_cache: dict[ObservationAreaID, dict[str, scd.FetchedEntity]] = {}
 
 
-def _get_tracer_logger() -> Logger | None:
+def _get_tracer_logger() -> Logger:
     kml_server = webapp.config[KEY_TRACER_KML_SERVER]
     kml_folder = webapp.config[KEY_TRACER_KML_FOLDER]
     output_folder = webapp.config[KEY_TRACER_OUTPUT_FOLDER]
@@ -36,7 +36,7 @@ def _get_tracer_logger() -> Logger | None:
         if kml_server
         else None
     )
-    return Logger(output_folder, kml_session) if output_folder else None
+    return Logger(output_folder, kml_session) if output_folder else DummyLogger()
 
 
 tracer_logger: Logger = _get_tracer_logger()
@@ -57,14 +57,16 @@ def resolve_auth_spec(requested_auth_spec: AuthSpec | None) -> AuthSpec:
         return requested_auth_spec
 
 
-def resolve_rid_dss_base_url(dss_base_url: str, rid_version: RIDVersion) -> str:
+def resolve_rid_dss_base_url(dss_base_url: str | None, rid_version: RIDVersion) -> str:
     if not dss_base_url:
-        if KEY_DSS_URL not in webapp.config or not webapp.config[KEY_DSS_URL]:
+
+        dss_base_url = webapp.config.get(KEY_DSS_URL)
+
+        if not dss_base_url:
             raise ValueError(
                 "DSS base URL was not specified explicitly nor in mock_uss_configuration"
             )
-        else:
-            dss_base_url = webapp.config[KEY_DSS_URL]
+
     if rid_version == RIDVersion.f3411_19:
         return dss_base_url
     elif rid_version == RIDVersion.f3411_22a:
@@ -75,14 +77,15 @@ def resolve_rid_dss_base_url(dss_base_url: str, rid_version: RIDVersion) -> str:
         )
 
 
-def resolve_scd_dss_base_url(dss_base_url: str) -> str:
+def resolve_scd_dss_base_url(dss_base_url: str | None) -> str:
     if not dss_base_url:
-        if KEY_DSS_URL not in webapp.config or not webapp.config[KEY_DSS_URL]:
+        dss_base_url = webapp.config.get(KEY_DSS_URL)
+
+        if not dss_base_url:
             raise ValueError(
                 "DSS base URL was not specified explicitly nor in mock_uss_configuration"
             )
-        else:
-            dss_base_url = webapp.config[KEY_DSS_URL]
+
     return dss_base_url
 
 

monitoring/mock_uss/tracer/context.py

@@ -1,13 +1,14 @@
+from typing import cast
 from monitoring.monitorlib import formatting
 from monitoring.monitorlib.fetch import rid, scd, summarize
 
 
 def isa_diff_text(a: rid.FetchedISAs | None, b: rid.FetchedISAs | None) -> str:
     """Create text to display to a real-time user describing a change in ISAs."""
     a_summary = summarize.isas(a) if a else {}
-    a_summary = summarize.limit_long_arrays(a_summary, 6)
+    a_summary = cast(dict, summarize.limit_long_arrays(a_summary, 6))
     b_summary = summarize.isas(b) if b else {}
-    b_summary = summarize.limit_long_arrays(b_summary, 6)
+    b_summary = cast(dict, summarize.limit_long_arrays(b_summary, 6))
     if b is not None and b.success and a is not None and not a.success:
         a_summary = {}
     if a is not None and a.success and b is not None and not b.success:
@@ -26,9 +27,9 @@ def entity_diff_text(
     if entity_type and "_" in entity_type:
         entity_type = entity_type[0 : entity_type.index("_")]
     a_summary = summarize.entities(a, entity_type) if a else {}
-    a_summary = summarize.limit_long_arrays(a_summary, 6)
+    a_summary = cast(dict, summarize.limit_long_arrays(a_summary, 6))
     b_summary = summarize.entities(b, entity_type) if b else {}
-    b_summary = summarize.limit_long_arrays(b_summary, 6)
+    b_summary = cast(dict, summarize.limit_long_arrays(b_summary, 6))
     if b is not None and b.success and a is not None and not a.success:
         a_summary = {}
     if a is not None and a.success and b is not None and not b.success:

monitoring/mock_uss/tracer/diff.py

@@ -7,6 +7,7 @@
 from datetime import UTC, datetime, timedelta
 from enum import Enum
 from typing import Protocol
+from abc import abstractmethod
 
 import yaml
 from implicitdict import ImplicitDict
@@ -40,7 +41,8 @@ def __enter__(self):
         self._start_time = datetime.now(UTC)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        self.elapsed_time += datetime.now(UTC) - self._start_time
+        if self._start_time:
+            self.elapsed_time += datetime.now(UTC) - self._start_time
 
 
 class VolumeType(str, Enum):
@@ -57,10 +59,12 @@ class HistoricalVolumesCollection:
     active_at: datetime
 
 
-class HistoricalVolumesRenderer(Protocol):
+class HistoricalVolumesRenderer[T](Protocol):
+
+    @abstractmethod
     def __call__(
         self,
-        log_entry: TracerLogEntry,
+        log_entry: T,
         existing_volume_collections: list[HistoricalVolumesCollection],
     ) -> list[HistoricalVolumesCollection]:
         """Function that generates named collections of 4D volumes from a tracer log entry.
@@ -71,6 +75,7 @@ def __call__(
 
         Returns: Collection of 4D volume collections.
         """
+        raise NotImplementedError
 
 
 @dataclass
@@ -126,6 +131,10 @@ def _historical_volumes_op_intent_notification(
     existing_volume_collections: list[HistoricalVolumesCollection],
 ) -> list[HistoricalVolumesCollection]:
     try:
+
+        if log_entry.request.json is None:
+            raise ValueError("No json data in log entry")
+
         req = ImplicitDict.parse(
             log_entry.request.json, PutOperationalIntentDetailsParameters
         )
@@ -136,7 +145,7 @@ def _historical_volumes_op_intent_notification(
         return []
     assert isinstance(req, PutOperationalIntentDetailsParameters)
 
-    claims = get_token_claims(log_entry.request.headers)
+    claims = get_token_claims(log_entry.request.headers or {})
     manager = claims.get("sub", "[Unknown manager]")
     name = f"{manager} {req.operational_intent_id}"
 
@@ -148,7 +157,7 @@ def _historical_volumes_op_intent_notification(
     else:
         version = "[deleted]"
         state = "Ended"
-        volumes = []
+        volumes = Volume4DCollection()
 
     # See if this op intent version already has a volumes collection
     already_defined = False
@@ -183,6 +192,10 @@ def _historical_volumes_op_intent_poll(
     # Add newly-polled operational intents
     for op_intent_id, query in log_entry.poll.uss_queries.items():
         try:
+
+            if query.json_result is None:
+                raise ValueError("No json result in query")
+
             resp = ImplicitDict.parse(
                 query.json_result, GetOperationalIntentDetailsResponse
             )
@@ -222,6 +235,10 @@ def _historical_volumes_op_intent_poll(
     # Remove any existing operational intents that no longer exist as of this poll
     for cached_op_intent_id, cached_query in log_entry.poll.cached_uss_queries.items():
         try:
+
+            if cached_query.json_result is None:
+                raise ValueError("No json result in query")
+
             resp = ImplicitDict.parse(
                 cached_query.json_result, GetOperationalIntentDetailsResponse
             )
@@ -274,17 +291,19 @@ class VolumesFolder:
     def truncate(self, latest_time: Time) -> None:
         to_remove = []
         for v in self.volumes:
-            if v.volume.time_start.datetime > latest_time.datetime:
+            if v.volume.time_start and v.volume.time_start.datetime > latest_time.datetime:
                 to_remove.append(v)
-            elif v.volume.time_end.datetime > latest_time.datetime:
+            elif v.volume.time_end and v.volume.time_end.datetime > latest_time.datetime:
                 v.volume.time_end = latest_time
         for v in to_remove:
             self.volumes.remove(v)
         for c in self.children:
             c.truncate(latest_time)
 
-    def to_kml_folder(self) -> kml.Folder:
+    def to_kml_folder(self):
         def dt(t: Time) -> int:
+            if self.reference_time is None:
+                return -1
             return round((t.datetime - self.reference_time.datetime).total_seconds())
 
         if self.reference_time:
@@ -293,7 +312,7 @@ def dt(t: Time) -> int:
         else:
             folder = kml.Folder(kml.name(self.name))
         for v in self.volumes:
-            name = f"{v.name} {dt(v.volume.time_start)}s-{dt(v.volume.time_end)}s"
+            name = f"{v.name} {dt(v.volume.time_start) if v.volume.time_start else '?'}s-{dt(v.volume.time_end) if v.volume.time_end else '?'}s"
             folder.append(
                 make_placemark_from_volume(v.volume, name=name, style_url=v.style)
             )
@@ -408,13 +427,13 @@ def render_historical_kml(log_folder: str) -> str:
             version_folder.children.append(future_folder)
 
             for i, v in enumerate(hvc.volumes):
-                if v.time_end.datetime <= hvc.active_at:
+                if v.time_end and v.time_end.datetime <= hvc.active_at:
                     # This volume ended before the collection was declared, so it never actually existed
                     continue
-                if v.time_start.datetime < hvc.active_at:
+                if v.time_start and v.time_start.datetime < hvc.active_at:
                     # Volume is declared in the past, but it's only visible starting now
                     v.time_start = t_hvc
-                elif v.time_start.datetime > hvc.active_at:
+                elif v.time_start and v.time_start.datetime > hvc.active_at:
                     # Add a "future" volume between when this volume was declared and its start time
                     future_v = Volume4D(v)
                     future_v.time_end = v.time_start

monitoring/mock_uss/tracer/kml.py

@@ -67,7 +67,7 @@ class ObservationArea(ImplicitDict):
     @property
     def polls(self) -> bool:
         """Whether any of the observation activity involves periodic polling."""
-        return (self.f3411 and self.f3411.poll) or (self.f3548 and self.f3548.poll)
+        return bool(self.f3411 and self.f3411.poll) or bool(self.f3548 and self.f3548.poll)
 
 
 class F3411ObservationAreaRequest(ImplicitDict):
@@ -134,7 +134,7 @@ class ObservationAreaRequest(ImplicitDict):
     @property
     def polls(self) -> bool:
         """Whether any of the observation activity requested involves periodic polling."""
-        return (self.f3411 and self.f3411.poll) or (self.f3548 and self.f3548.poll)
+        return bool(self.f3411 and self.f3411.poll) or bool(self.f3548 and self.f3548.poll)
 
 
 class ListObservationAreasResponse(ImplicitDict):

monitoring/mock_uss/tracer/observation_areas.py

@@ -36,7 +36,7 @@
 
 
 @webapp.route("/tracer/observation_areas", methods=["GET"])
-@ui_auth.login_required
+@ui_auth.login_required()
 def tracer_list_observation_areas() -> flask.Response:
     with db as tx:
         result = ListObservationAreasResponse(
@@ -131,7 +131,7 @@ def tracer_import_observation_areas() -> tuple[str, int] | flask.Response:
         elif request.area.volume.outline_polygon:
             points = [
                 s2sphere.LatLng.from_degrees(p.lat, p.lng)
-                for p in request.area.volume.outline_polygon.vertices
+                for p in request.area.volume.outline_polygon.vertices or []
             ]
         else:
             raise NotImplementedError(

monitoring/mock_uss/tracer/routes/observation_areas.py

@@ -39,6 +39,10 @@ def tracer_scd_v21_operation_notification(observation_area_id: str) -> tuple[str
     label = colored("Operation", "blue")
     try:
         json = flask.request.json
+
+        if json is None:
+            raise ValueError("No json in request")
+
         id = json.get("operational_intent_id", "<Unknown ID>")
         if json.get("operational_intent"):
             op = json["operational_intent"]
@@ -94,6 +98,10 @@ def tracer_scd_v21_constraint_notification(observation_area_id: str) -> tuple[st
     label = colored("Constraint", "magenta")
     try:
         json = flask.request.json
+
+        if json is None:
+            raise ValueError("No json in request")
+
         id = json.get("constraint_id", "<Unknown ID>")
         if json.get("constraint"):
             constraint = json["constraint"]