adding pydantic validation

Author: timmarkhuff

src/groundlight/edge/config.py

@@ -78,7 +78,7 @@ class DetectorConfig(BaseModel):  # pylint: disable=too-few-public-methods
 
     model_config = ConfigDict(extra="ignore")
 
-    detector_id: str = Field(..., description="Detector ID")
+    detector_id: str = Field(..., pattern=r"^det_[A-Za-z0-9]{27}$", description="Detector ID")
     edge_inference_config: str = Field(..., description="Config for edge inference.")
 
 

test/unit/test_edge_config.py

@@ -17,6 +17,11 @@
 CUSTOM_AUDIT_RATE = 0.0
 REFRESH_RATE_SECONDS = 15.0
 
+# Mock detector IDs
+DET_1 = "det_000000000000000000000000001"
+DET_2 = "det_000000000000000000000000002"
+DET_3 = "det_000000000000000000000000003"
+
 
 def _make_detector(detector_id: str) -> Detector:
     return Detector(
@@ -36,15 +41,15 @@ def test_add_detector_allows_equivalent_named_inference_config():
     """Allows reusing the same named inference config with equivalent values."""
     detectors_config = DetectorsConfig()
     detectors_config.add_detector(
-        "det_1",
+        DET_1,
         InferenceConfig(
             name="custom_config",
             always_return_edge_prediction=True,
             min_time_between_escalations=0.5,
         ),
     )
     detectors_config.add_detector(
-        "det_2",
+        DET_2,
         InferenceConfig(
             name="custom_config",
             always_return_edge_prediction=True,
@@ -59,22 +64,22 @@ def test_add_detector_allows_equivalent_named_inference_config():
 def test_add_detector_rejects_different_named_inference_config():
     """Rejects conflicting inference config values under the same name."""
     detectors_config = DetectorsConfig()
-    detectors_config.add_detector("det_1", InferenceConfig(name="custom_config"))
+    detectors_config.add_detector(DET_1, InferenceConfig(name="custom_config"))
 
     with pytest.raises(ValueError, match="different inference config named 'custom_config'"):
         detectors_config.add_detector(
-            "det_2",
+            DET_2,
             InferenceConfig(name="custom_config", always_return_edge_prediction=True),
         )
 
 
 def test_add_detector_rejects_duplicate_detector_id():
     """Rejects adding the same detector ID more than once."""
     detectors_config = DetectorsConfig()
-    detectors_config.add_detector("det_1", DEFAULT)
+    detectors_config.add_detector(DET_1, DEFAULT)
 
     with pytest.raises(ValueError, match="already exists"):
-        detectors_config.add_detector("det_1", DEFAULT)
+        detectors_config.add_detector(DET_1, DEFAULT)
 
 
 def test_constructor_rejects_duplicate_detector_ids():
@@ -83,8 +88,8 @@ def test_constructor_rejects_duplicate_detector_ids():
         DetectorsConfig(
             edge_inference_configs={"default": DEFAULT},
             detectors=[
-                {"detector_id": "det_1", "edge_inference_config": "default"},
-                {"detector_id": "det_1", "edge_inference_config": "default"},
+                {"detector_id": DET_1, "edge_inference_config": "default"},
+                {"detector_id": DET_1, "edge_inference_config": "default"},
             ],
         )
 
@@ -102,18 +107,18 @@ def test_constructor_accepts_matching_inference_config_key_and_name():
     """Accepts constructor input when key/name pairs are consistent."""
     config = DetectorsConfig(
         edge_inference_configs={"default": InferenceConfig(name="default")},
-        detectors=[{"detector_id": "det_1", "edge_inference_config": "default"}],
+        detectors=[{"detector_id": DET_1, "edge_inference_config": "default"}],
     )
 
     assert list(config.edge_inference_configs.keys()) == ["default"]
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_constructor_hydrates_inference_config_name_from_dict_key():
     """Hydrates inference config names from payload dict keys."""
     config = DetectorsConfig(
         edge_inference_configs={"default": {"enabled": True}},
-        detectors=[{"detector_id": "det_1", "edge_inference_config": "default"}],
+        detectors=[{"detector_id": DET_1, "edge_inference_config": "default"}],
     )
 
     assert config.edge_inference_configs["default"].name == "default"
@@ -124,7 +129,7 @@ def test_constructor_rejects_detector_map_input():
     with pytest.raises(ValueError):
         DetectorsConfig(
             edge_inference_configs={"default": {"enabled": True}},
-            detectors={"det_1": {"detector_id": "det_1", "edge_inference_config": "default"}},
+            detectors={DET_1: {"detector_id": DET_1, "edge_inference_config": "default"}},
         )
 
 
@@ -133,33 +138,33 @@ def test_constructor_rejects_undefined_inference_config_reference():
     with pytest.raises(ValueError, match="not defined"):
         DetectorsConfig(
             edge_inference_configs={},
-            detectors=[{"detector_id": "det_1", "edge_inference_config": "does_not_exist"}],
+            detectors=[{"detector_id": DET_1, "edge_inference_config": "does_not_exist"}],
         )
 
 
 def test_edge_endpoint_config_add_detector_uses_shared_config_logic():
     """Adds detectors via EdgeEndpointConfig and preserves inferred config mapping."""
     config = EdgeEndpointConfig()
-    config.add_detector("det_1", NO_CLOUD)
-    config.add_detector("det_2", EDGE_ANSWERS_WITH_ESCALATION)
-    config.add_detector("det_3", DEFAULT)
+    config.add_detector(DET_1, NO_CLOUD)
+    config.add_detector(DET_2, EDGE_ANSWERS_WITH_ESCALATION)
+    config.add_detector(DET_3, DEFAULT)
 
-    assert [detector.detector_id for detector in config.detectors] == ["det_1", "det_2", "det_3"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1, DET_2, DET_3]
     assert set(config.edge_inference_configs.keys()) == {"no_cloud", "edge_answers_with_escalation", "default"}
 
 
 def test_add_detector_accepts_detector_object():
     """Accepts Detector objects in add_detector."""
     config = EdgeEndpointConfig()
-    config.add_detector(_make_detector("det_1"), DEFAULT)
+    config.add_detector(_make_detector(DET_1), DEFAULT)
 
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_disabled_preset_can_be_used():
     """Allows assigning the DISABLED inference preset to a detector."""
     config = EdgeEndpointConfig()
-    config.add_detector("det_1", DISABLED)
+    config.add_detector(DET_1, DISABLED)
 
     assert [detector.edge_inference_config for detector in config.detectors] == ["disabled"]
     assert config.edge_inference_configs["disabled"] == DISABLED
@@ -168,8 +173,8 @@ def test_disabled_preset_can_be_used():
 def test_detectors_config_to_payload_shape():
     """Serializes detector-scoped payload with expected top-level keys."""
     detectors_config = DetectorsConfig()
-    detectors_config.add_detector("det_1", DEFAULT)
-    detectors_config.add_detector("det_2", NO_CLOUD)
+    detectors_config.add_detector(DET_1, DEFAULT)
+    detectors_config.add_detector(DET_2, NO_CLOUD)
 
     payload = detectors_config.to_payload()
 
@@ -182,11 +187,11 @@ def test_edge_endpoint_config_accepts_top_level_payload_shape():
     config = EdgeEndpointConfig.model_validate({
         "global_config": {"refresh_rate": CUSTOM_REFRESH_RATE},
         "edge_inference_configs": {"default": {"enabled": True}},
-        "detectors": [{"detector_id": "det_1", "edge_inference_config": "default"}],
+        "detectors": [{"detector_id": DET_1, "edge_inference_config": "default"}],
     })
 
     assert config.global_config.refresh_rate == CUSTOM_REFRESH_RATE
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_edge_endpoint_config_from_yaml_accepts_yaml_text():
@@ -198,12 +203,12 @@ def test_edge_endpoint_config_from_yaml_accepts_yaml_text():
           default:
             enabled: true
         detectors:
-          - detector_id: det_1
+          - detector_id: {DET_1}
             edge_inference_config: default
         """)
 
     assert config.global_config.refresh_rate == REFRESH_RATE_SECONDS
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_edge_endpoint_config_from_yaml_accepts_filename(tmp_path):
@@ -215,12 +220,12 @@ def test_edge_endpoint_config_from_yaml_accepts_filename(tmp_path):
         "  default:\n"
         "    enabled: true\n"
         "detectors:\n"
-        "  - detector_id: det_1\n"
+        f"  - detector_id: {DET_1}\n"
         "    edge_inference_config: default\n"
     )
     config = EdgeEndpointConfig.from_yaml(filename=str(config_file))
 
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_edge_endpoint_config_from_yaml_requires_exactly_one_input():
@@ -243,23 +248,23 @@ def test_edge_endpoint_config_ignores_extra_fields_at_all_levels():
             "default": {"enabled": True, "unknown_inference_field": 42},
         },
         "detectors": [
-            {"detector_id": "det_1", "edge_inference_config": "default", "unknown_detector_field": [1, 2]},
+            {"detector_id": DET_1, "edge_inference_config": "default", "unknown_detector_field": [1, 2]},
         ],
         "unknown_top_level_field": True,
     })
     assert config.global_config.refresh_rate == REFRESH_RATE_SECONDS
     assert config.edge_inference_configs["default"].enabled is True
-    assert config.detectors[0].detector_id == "det_1"
+    assert config.detectors[0].detector_id == DET_1
 
 
 def test_model_dump_shape_for_edge_endpoint_config():
     """Serializes full edge endpoint config in wire payload shape."""
     config = EdgeEndpointConfig(
         global_config=GlobalConfig(refresh_rate=CUSTOM_REFRESH_RATE, confident_audit_rate=CUSTOM_AUDIT_RATE)
     )
-    config.add_detector("det_1", DEFAULT)
-    config.add_detector("det_2", EDGE_ANSWERS_WITH_ESCALATION)
-    config.add_detector("det_3", NO_CLOUD)
+    config.add_detector(DET_1, DEFAULT)
+    config.add_detector(DET_2, EDGE_ANSWERS_WITH_ESCALATION)
+    config.add_detector(DET_3, NO_CLOUD)
 
     payload = config.to_payload()
 
@@ -272,8 +277,8 @@ def test_model_dump_shape_for_edge_endpoint_config():
 def test_edge_endpoint_config_from_payload_round_trip():
     """Round-trips edge endpoint config through payload helpers."""
     config = EdgeEndpointConfig()
-    config.add_detector("det_1", DEFAULT)
-    config.add_detector("det_2", NO_CLOUD)
+    config.add_detector(DET_1, DEFAULT)
+    config.add_detector(DET_2, NO_CLOUD)
 
     payload = config.to_payload()
     reconstructed = EdgeEndpointConfig.from_payload(payload)
@@ -286,14 +291,14 @@ def test_edge_endpoint_config_from_payload_accepts_literal_payload():
     payload = {
         "global_config": {"refresh_rate": REFRESH_RATE_SECONDS},
         "edge_inference_configs": {"default": {"enabled": True}},
-        "detectors": [{"detector_id": "det_1", "edge_inference_config": "default"}],
+        "detectors": [{"detector_id": DET_1, "edge_inference_config": "default"}],
     }
 
     config = EdgeEndpointConfig.from_payload(payload)
 
     assert config.global_config.refresh_rate == REFRESH_RATE_SECONDS
     assert config.edge_inference_configs["default"].name == "default"
-    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+    assert [detector.detector_id for detector in config.detectors] == [DET_1]
 
 
 def test_inference_config_validation_errors():
@@ -318,7 +323,7 @@ def test_edge_get_config_parses_response():
     payload = {
         "global_config": {"refresh_rate": REFRESH_RATE_SECONDS},
         "edge_inference_configs": {"default": {"enabled": True}},
-        "detectors": [{"detector_id": "det_1", "edge_inference_config": "default"}],
+        "detectors": [{"detector_id": DET_1, "edge_inference_config": "default"}],
     }
 
     mock_response = Mock()
@@ -333,4 +338,4 @@ def test_edge_get_config_parses_response():
     assert isinstance(config, EdgeEndpointConfig)
     assert config.global_config.refresh_rate == REFRESH_RATE_SECONDS
     assert config.edge_inference_configs["default"].name == "default"
-    assert [d.detector_id for d in config.detectors] == ["det_1"]
+    assert [d.detector_id for d in config.detectors] == [DET_1]