fix(ground): validate UDP port and window name for GStreamer receiver

ground/ground_resources/comms/test_udp_gst_validate.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+import unittest
+
+from udp_gst_validate import validate_udp_port, validate_window_name
+
+
+class TestUdpGstValidate(unittest.TestCase):
+    def test_port_ok(self):
+        self.assertEqual(validate_udp_port(5600), 5600)
+        self.assertEqual(validate_udp_port(1), 1)
+        self.assertEqual(validate_udp_port(65535), 65535)
+
+    def test_port_bad(self):
+        for bad in (0, -1, 65536, 1.5, True, "5600", None):
+            with self.subTest(bad=bad):
+                with self.assertRaises(ValueError):
+                    validate_udp_port(bad)
+
+    def test_name_ok(self):
+        self.assertEqual(validate_window_name("cam0"), "cam0")
+        self.assertEqual(validate_window_name("  left  "), "left")
+
+    def test_name_bad(self):
+        for bad in ("", "   ", None, 1):
+            with self.subTest(bad=bad):
+                with self.assertRaises(ValueError):
+                    validate_window_name(bad)
+
+
+if __name__ == "__main__":
+    unittest.main()

ground/ground_resources/comms/udp_gst_receiver.py

@@ -2,14 +2,19 @@
 import argparse
 import sys
 
+from udp_gst_validate import validate_udp_port, validate_window_name
+
 def main():
     parser = argparse.ArgumentParser(description="Receive UDP GStreamer video and display it.")
     parser.add_argument('--port', type=int, required=True, help="UDP port to listen on.")
     parser.add_argument('--name', type=str, required=True, help="Window name to display.")
     args = parser.parse_args()
 
-    port = args.port
-    name = args.name
+    try:
+        port = validate_udp_port(args.port)
+        name = validate_window_name(args.name)
+    except ValueError as exc:
+        parser.error(str(exc))
 
     pipeline = (
         f"udpsrc port={port} ! "

ground/ground_resources/comms/udp_gst_validate.py

@@ -0,0 +1,22 @@
+"""Pure validators for UDP GStreamer receiver CLI (offline-testable)."""
+
+from __future__ import annotations
+
+
+def validate_udp_port(port) -> int:
+    """Return port if it is an int in 1..65535; else raise ValueError."""
+    if isinstance(port, bool) or not isinstance(port, int):
+        raise ValueError(f"port must be an int, got {type(port).__name__}")
+    if port < 1 or port > 65535:
+        raise ValueError(f"port must be in 1..65535, got {port}")
+    return port
+
+
+def validate_window_name(name) -> str:
+    """Return non-empty stripped window name; else raise ValueError."""
+    if not isinstance(name, str):
+        raise ValueError(f"name must be a str, got {type(name).__name__}")
+    cleaned = name.strip()
+    if not cleaned:
+        raise ValueError("name must be a non-empty string")
+    return cleaned