Cleanup the code a little. Protect against pod being None in status_data

Author: alfredeen

serve_event_listener/el_types.py

@@ -5,13 +5,13 @@
 # Canonical status/app types
 Status = Literal["Running", "Pending", "Deleted", "Failed", "Unknown", "Succeeded"]
 AppType = Literal["shiny", "shiny-proxy"]
-ProbeClass = Literal["Running", "Unknown", "NotFound"]
+ProbeStatus = Literal["Running", "Unknown", "NotFound"]
 
 # Result block produced by the AppAvailabilityProbe (optional on payload)
-ProbeBlock = TypedDict(
-    "ProbeBlock",
+ProbeBlockDict = TypedDict(
+    "ProbeBlockDict",
     {
-        "status": ProbeClass,  # e.g., "Running" | "Unknown" | "NotFound"
+        "status": ProbeStatus,  # e.g., "Running" | "Unknown" | "NotFound"
         "port80_status": Optional[int],  # HTTP status code (if any)
         "note": str,  # short diagnostic
         "url": str,  # probed URL
@@ -34,7 +34,7 @@
         "message": NotRequired[str],
         "app-type": NotRequired[AppType],
         "app-url": NotRequired[str],
-        "curl-probe": NotRequired[ProbeBlock],
+        "curl-probe": NotRequired[ProbeBlockDict],
     },
     total=False,
 )

serve_event_listener/event_listener.py

@@ -100,7 +100,7 @@ def status_data(self) -> StatusData:
         return self._status_data
 
     @property
-    def client_api_ping_endpoint(self) -> str:
+    def client_api_health_endpoint(self) -> str:
         """A URL of the client target API for checking the status of the API is UP or DOWN"""
         return BASE_URL + "/openapi/v1/are-you-there"
 
@@ -133,8 +133,8 @@ def setup(self, **kwargs: Optional[Any]) -> None:
 
             try:
                 # Verify that the prober works for at least a known URL,
-                test_url = self.client_api_ping_endpoint
-                _ = self._prober.probe_url(test_url)
+                health_url = self.client_api_health_endpoint
+                _ = self._prober.probe_url(health_url)
             except Exception as e:
                 # Otherwise disable the probing feature
                 logger.warning("Probing disabled: baseline probe failed (%s)", e)
@@ -279,7 +279,7 @@ def check_serve_api_status(self) -> bool:
         Returns:
         - bool: True if the status is okay, False otherwise.
         """
-        url = self.client_api_ping_endpoint
+        url = self.client_api_health_endpoint
         logger.debug("Verifying that the server API is up and available via %s", url)
 
         # Using the new http get function

serve_event_listener/main.py

@@ -67,7 +67,8 @@ def parse_args():
         "--mode",
         choices=["normal", "diagnostics", "probetest"],
         default="normal",
-        help="run mode (default: normal)",
+        help="run mode (default: normal). diagnostics prints out configuration info and exists. \
+              probetest runs a probe against a specified URL.",
     )
     parser.add_argument(
         "--probe-url",

serve_event_listener/probing.py

@@ -8,6 +8,7 @@
 
 import requests
 
+from serve_event_listener.el_types import ProbeStatus
 from serve_event_listener.http_client import get as http_get
 
 logger = logging.getLogger(__name__)
@@ -17,7 +18,7 @@
 class ProbeResult:
     """Outcome of an availability probe."""
 
-    status: str  # "Running" | "Unknown" | "NotFound"
+    status: ProbeStatus  # "Running" | "Unknown" | "NotFound"
     port80_status: Optional[int] = None
     note: str = ""  # short diagnostic
 

serve_event_listener/status_data.py

@@ -30,7 +30,15 @@
 }
 
 
-def _detect_app_type(pod) -> Optional[AppType]:
+def _detect_app_type(pod: Optional[object]) -> Optional[AppType]:
+    """Best-effort app-type detection from a k8s Pod-like object.
+
+    Accepts None to match upstream callers.
+    Returns None when the shape is incomplete or labels are missing.
+    """
+    if pod is None:
+        return None
+
     labels = getattr(pod.metadata, "labels", None) or {}
     ann = getattr(pod.metadata, "annotations", None) or {}
     values = " ".join(

tests/integration/test_status_data_integration.py

@@ -72,7 +72,7 @@ def test_update_sets_app_url_for_shinyproxy(self):
                 f"No shiny-proxy pod with a 'release' label found in namespace {namespace}"
             )
 
-        print("/nFound a shiny pod: ", pod.metadata.name)
+        print("\nFound a shiny pod: ", pod.metadata.name)
         sd = StatusData(namespace=namespace)
         # Simulate a k8s watch event shape
         t0 = time.time()