From c106ee58d614e0edd05e3fb3bb2ff76df1676cfa Mon Sep 17 00:00:00 2001 From: muhamadrifansyah Date: Sun, 22 Feb 2026 11:23:07 +0700 Subject: [PATCH 1/2] improve(config): add early validation with contextual error messages for ModeSystemConfig --- src/runtime/config.py | 110 ++++++++++++++- tests/runtime/test_config.py | 259 +++++++++++++++++++++++++++++++++++ 2 files changed, 363 insertions(+), 6 deletions(-) diff --git a/src/runtime/config.py b/src/runtime/config.py index f986218fa..b4f29e052 100644 --- a/src/runtime/config.py +++ b/src/runtime/config.py @@ -88,6 +88,45 @@ def validate_config_schema(raw_config: dict) -> None: raise +def _validate_required_field( + config_dict: dict, + field_name: str, + context: str = "", + field_type: str = "field", +) -> Any: + """ + Validate that a required field exists in a configuration dictionary. + + Parameters + ---------- + config_dict : dict + The configuration dictionary to check. + field_name : str + The name of the required field. + context : str + Additional context for error message (e.g., mode name, rule index). + field_type : str + Type of field for error message (e.g., 'field', 'property'). + + Returns + ------- + Any + The field value if it exists. + + Raises + ------ + ValueError + If the required field is missing with detailed error context. + """ + if field_name not in config_dict: + context_msg = f" in {context}" if context else "" + raise ValueError( + f"Required {field_type} '{field_name}' is missing{context_msg}. " + f"Please ensure this field is defined in your configuration." + ) + return config_dict[field_name] + + @dataclass class RuntimeConfig: """ @@ -564,10 +603,15 @@ def load_mode_config( load_unitree(g_ut_eth) + # Validate and extract default_mode early to provide clear error messages + default_mode = _validate_required_field( + raw_config, "default_mode", context="global configuration", field_type="field" + ) + mode_system_config = ModeSystemConfig( version=config_version, name=raw_config.get("name", "mode_system"), - default_mode=raw_config["default_mode"], + default_mode=default_mode, config_name=config_name, allow_manual_switching=raw_config.get("allow_manual_switching", True), mode_memory_enabled=raw_config.get("mode_memory_enabled", True), @@ -585,12 +629,20 @@ def load_mode_config( ) for mode_name, mode_data in raw_config.get("modes", {}).items(): + # Validate required mode fields early + system_prompt_base = _validate_required_field( + mode_data, + "system_prompt_base", + context=f"mode '{mode_name}'", + field_type="field", + ) + mode_config = ModeConfig( version=mode_data.get("version", "1.0.1"), name=mode_name, display_name=mode_data.get("display_name", mode_name), description=mode_data.get("description", ""), - system_prompt_base=mode_data["system_prompt_base"], + system_prompt_base=system_prompt_base, hertz=mode_data.get("hertz", 1.0), lifecycle_hooks=parse_lifecycle_hooks( mode_data.get("lifecycle_hooks", []), api_key=g_api_key @@ -610,11 +662,31 @@ def load_mode_config( mode_system_config.modes[mode_name] = mode_config - for rule_data in raw_config.get("transition_rules", []): + for rule_idx, rule_data in enumerate(raw_config.get("transition_rules", [])): + # Validate required transition rule fields early + from_mode = _validate_required_field( + rule_data, + "from_mode", + context=f"transition rule at index {rule_idx}", + field_type="field", + ) + to_mode = _validate_required_field( + rule_data, + "to_mode", + context=f"transition rule at index {rule_idx}", + field_type="field", + ) + transition_type_str = _validate_required_field( + rule_data, + "transition_type", + context=f"transition rule at index {rule_idx}", + field_type="field", + ) + rule = TransitionRule( - from_mode=rule_data["from_mode"], - to_mode=rule_data["to_mode"], - transition_type=TransitionType(rule_data["transition_type"]), + from_mode=from_mode, + to_mode=to_mode, + transition_type=TransitionType(transition_type_str), trigger_keywords=rule_data.get("trigger_keywords", []), priority=rule_data.get("priority", 1), cooldown_seconds=rule_data.get("cooldown_seconds", 0.0), @@ -623,6 +695,32 @@ def load_mode_config( ) mode_system_config.transition_rules.append(rule) + # Validate that default_mode exists in the loaded modes + if default_mode not in mode_system_config.modes: + available_modes = ", ".join(mode_system_config.modes.keys()) + raise ValueError( + f"Default mode '{default_mode}' not found in available modes. " + f"Available modes: {available_modes if available_modes else 'none'}. " + f"Please ensure the default_mode matches one of the defined modes in your configuration." + ) + + # Validate that modes referenced in transition rules exist + for rule_idx, rule in enumerate(mode_system_config.transition_rules): + if rule.from_mode not in mode_system_config.modes: + available_modes = ", ".join(mode_system_config.modes.keys()) + raise ValueError( + f"Transition rule at index {rule_idx} references unknown 'from_mode' '{rule.from_mode}'. " + f"Available modes: {available_modes}. " + f"Please ensure all transition rules reference valid mode names." + ) + if rule.to_mode not in mode_system_config.modes: + available_modes = ", ".join(mode_system_config.modes.keys()) + raise ValueError( + f"Transition rule at index {rule_idx} references unknown 'to_mode' '{rule.to_mode}'. " + f"Available modes: {available_modes}. " + f"Please ensure all transition rules reference valid mode names." + ) + return mode_system_config diff --git a/tests/runtime/test_config.py b/tests/runtime/test_config.py index 02892f36f..41d5dacd9 100644 --- a/tests/runtime/test_config.py +++ b/tests/runtime/test_config.py @@ -14,6 +14,7 @@ TransitionType, _load_mode_components, _load_schema, + _validate_required_field, load_mode_config, mode_config_to_dict, validate_config_schema, @@ -468,6 +469,212 @@ def test_load_mode_config_invalid_version(self): finally: os.unlink(temp_file) + def test_load_mode_config_missing_default_mode(self): + """Test load_mode_config with missing default_mode field.""" + config_data = { + "version": "v1.0.3", + "name": "test_system", + "api_key": "test_key", + "system_governance": "Test governance", + "cortex_llm": {"type": "test_llm"}, + "modes": { + "default": { + "display_name": "Default", + "description": "Default mode", + "system_prompt_base": "Test prompt", + } + }, + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json5", delete=False) as f: + import json5 + + json5.dump(config_data, f) + temp_file = f.name + + try: + with patch("runtime.config.os.path.join") as mock_join: + mock_join.return_value = temp_file + + with pytest.raises(ValueError) as exc_info: + load_mode_config("missing_default_mode_test") + + error_msg = str(exc_info.value) + assert "default_mode" in error_msg + assert "is missing" in error_msg + + finally: + os.unlink(temp_file) + + def test_load_mode_config_missing_system_prompt_base(self): + """Test load_mode_config with missing system_prompt_base in mode.""" + config_data = { + "version": "v1.0.3", + "name": "test_system", + "default_mode": "default", + "api_key": "test_key", + "system_governance": "Test governance", + "cortex_llm": {"type": "test_llm"}, + "modes": { + "default": { + "display_name": "Default", + "description": "Default mode", + # Missing required system_prompt_base + } + }, + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json5", delete=False) as f: + import json5 + + json5.dump(config_data, f) + temp_file = f.name + + try: + with patch("runtime.config.os.path.join") as mock_join: + mock_join.return_value = temp_file + + with pytest.raises(ValueError) as exc_info: + load_mode_config("missing_system_prompt_test") + + error_msg = str(exc_info.value) + assert "system_prompt_base" in error_msg + assert "mode 'default'" in error_msg + assert "is missing" in error_msg + + finally: + os.unlink(temp_file) + + def test_load_mode_config_default_mode_not_in_modes(self): + """Test load_mode_config when default_mode doesn't match any defined mode.""" + config_data = { + "version": "v1.0.3", + "name": "test_system", + "default_mode": "nonexistent", + "api_key": "test_key", + "system_governance": "Test governance", + "cortex_llm": {"type": "test_llm"}, + "modes": { + "default": { + "display_name": "Default", + "description": "Default mode", + "system_prompt_base": "Test prompt", + } + }, + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json5", delete=False) as f: + import json5 + + json5.dump(config_data, f) + temp_file = f.name + + try: + with patch("runtime.config.os.path.join") as mock_join: + mock_join.return_value = temp_file + + with pytest.raises(ValueError) as exc_info: + load_mode_config("default_mode_not_in_modes_test") + + error_msg = str(exc_info.value) + assert "Default mode 'nonexistent' not found" in error_msg + assert "Available modes:" in error_msg + assert "default" in error_msg + + finally: + os.unlink(temp_file) + + def test_load_mode_config_transition_rule_invalid_from_mode(self): + """Test load_mode_config with transition rule referencing unknown from_mode.""" + config_data = { + "version": "v1.0.3", + "name": "test_system", + "default_mode": "default", + "api_key": "test_key", + "system_governance": "Test governance", + "cortex_llm": {"type": "test_llm"}, + "modes": { + "default": { + "display_name": "Default", + "description": "Default mode", + "system_prompt_base": "Test prompt", + } + }, + "transition_rules": [ + { + "from_mode": "unknown", + "to_mode": "default", + "transition_type": "manual", + } + ], + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json5", delete=False) as f: + import json5 + + json5.dump(config_data, f) + temp_file = f.name + + try: + with patch("runtime.config.os.path.join") as mock_join: + mock_join.return_value = temp_file + + with pytest.raises(ValueError) as exc_info: + load_mode_config("invalid_from_mode_test") + + error_msg = str(exc_info.value) + assert "unknown 'from_mode' 'unknown'" in error_msg + assert "transition rule at index 0" in error_msg + + finally: + os.unlink(temp_file) + + def test_load_mode_config_transition_rule_missing_from_mode(self): + """Test load_mode_config with transition rule missing from_mode field.""" + config_data = { + "version": "v1.0.3", + "name": "test_system", + "default_mode": "default", + "api_key": "test_key", + "system_governance": "Test governance", + "cortex_llm": {"type": "test_llm"}, + "modes": { + "default": { + "display_name": "Default", + "description": "Default mode", + "system_prompt_base": "Test prompt", + } + }, + "transition_rules": [ + { + "to_mode": "default", + "transition_type": "manual", + # Missing from_mode + } + ], + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json5", delete=False) as f: + import json5 + + json5.dump(config_data, f) + temp_file = f.name + + try: + with patch("runtime.config.os.path.join") as mock_join: + mock_join.return_value = temp_file + + with pytest.raises(ValueError) as exc_info: + load_mode_config("missing_from_mode_test") + + error_msg = str(exc_info.value) + assert "from_mode" in error_msg + assert "is missing" in error_msg + assert "transition rule at index 0" in error_msg + + finally: + os.unlink(temp_file) + class TestModeConfigToDict: """Test cases for mode_config_to_dict function.""" @@ -738,3 +945,55 @@ def test_validate_config_error_message_at_root(self, caplog): with pytest.raises(ValidationError): validate_config_schema(config) # type: ignore assert "root" in caplog.text or "Schema validation failed" in caplog.text + + +class TestValidateRequiredField: + """Test cases for _validate_required_field helper function.""" + + def test_validate_required_field_exists(self): + """Test that function returns field value when it exists.""" + config = {"name": "test_mode", "value": 42} + result = _validate_required_field(config, "name") + assert result == "test_mode" + + def test_validate_required_field_missing_no_context(self): + """Test clear error message when required field is missing.""" + config = {"other_field": "value"} + with pytest.raises(ValueError) as exc_info: + _validate_required_field(config, "system_prompt_base") + error_msg = str(exc_info.value) + assert "system_prompt_base" in error_msg + assert "is missing" in error_msg + assert "Please ensure this field is defined" in error_msg + + def test_validate_required_field_missing_with_mode_context(self): + """Test error message includes mode context for better diagnostics.""" + config = {"other_field": "value"} + with pytest.raises(ValueError) as exc_info: + _validate_required_field( + config, "system_prompt_base", context="mode 'advanced'" + ) + error_msg = str(exc_info.value) + assert "system_prompt_base" in error_msg + assert "in mode 'advanced'" in error_msg + + def test_validate_required_field_missing_with_rule_context(self): + """Test error message includes rule index context.""" + config = {} + with pytest.raises(ValueError) as exc_info: + _validate_required_field( + config, "from_mode", context="transition rule at index 2" + ) + error_msg = str(exc_info.value) + assert "from_mode" in error_msg + assert "transition rule at index 2" in error_msg + + def test_validate_required_field_with_custom_field_type(self): + """Test error message with custom field type label.""" + config = {} + with pytest.raises(ValueError) as exc_info: + _validate_required_field( + config, "cortex_llm", field_type="required property" + ) + error_msg = str(exc_info.value) + assert "required property 'cortex_llm'" in error_msg From 731cefca78901ff179b845856225942fd949f626 Mon Sep 17 00:00:00 2001 From: muhamadrifansyah Date: Sun, 22 Feb 2026 19:32:53 +0700 Subject: [PATCH 2/2] fix(runtime): guard optional json5 and zenoh imports to allow pytest collection without optional dependencies --- src/runtime/config.py | 7 ++++++- src/runtime/manager.py | 17 +++++++++++++---- src/zenoh_msgs/session.py | 15 ++++++++++++--- 3 files changed, 31 insertions(+), 8 deletions(-) diff --git a/src/runtime/config.py b/src/runtime/config.py index b4f29e052..913fe413b 100644 --- a/src/runtime/config.py +++ b/src/runtime/config.py @@ -6,7 +6,10 @@ from pathlib import Path from typing import Any, Dict, List, Optional -import json5 +try: + import json5 +except ImportError: + json5 = None # type: ignore from jsonschema import ValidationError, validate from actions import load_action @@ -582,6 +585,8 @@ def load_mode_config( with open(config_path, "r") as f: try: + if json5 is None: + raise ImportError("json5 is required to load configuration files") raw_config = json5.load(f) except Exception as e: raise ValueError( diff --git a/src/runtime/manager.py b/src/runtime/manager.py index e7ae4ff38..bc1922a3c 100644 --- a/src/runtime/manager.py +++ b/src/runtime/manager.py @@ -6,8 +6,15 @@ from dataclasses import dataclass, field from typing import Callable, Dict, List, Optional -import json5 -import zenoh +try: + import json5 +except ImportError: + json5 = None # type: ignore + +try: + import zenoh +except ImportError: + zenoh = None # type: ignore from runtime.config import ( LifecycleHookType, @@ -146,6 +153,8 @@ def _create_runtime_config_file(self): temp_file = runtime_config_path + ".tmp" with open(temp_file, "w") as f: + if json5 is None: + raise ImportError("json5 is required to write configuration files") json5.dump(runtime_config, f, indent=2) os.rename(temp_file, runtime_config_path) @@ -715,7 +724,7 @@ async def process_tick( return None - def _zenoh_mode_status_request(self, data: zenoh.Sample): + def _zenoh_mode_status_request(self, data: "zenoh.Sample"): """ Process incoming mode status requests via Zenoh. @@ -764,7 +773,7 @@ def _zenoh_mode_status_request(self, data: zenoh.Sample): mode_status_response.serialize() ) - def _zenoh_context_update(self, data: zenoh.Sample): + def _zenoh_context_update(self, data: "zenoh.Sample"): """ Process incoming context update messages via Zenoh. diff --git a/src/zenoh_msgs/session.py b/src/zenoh_msgs/session.py index 32cc2aa79..124823b84 100644 --- a/src/zenoh_msgs/session.py +++ b/src/zenoh_msgs/session.py @@ -1,11 +1,20 @@ +from __future__ import annotations + import logging +from typing import TYPE_CHECKING + +try: + import zenoh +except ImportError: + zenoh = None # type: ignore -import zenoh +if TYPE_CHECKING: + from zenoh import Config, Session logging.basicConfig(level=logging.INFO) -def create_zenoh_config(network_discovery: bool = True) -> zenoh.Config: +def create_zenoh_config(network_discovery: bool = True) -> "zenoh.Config": """ Create a Zenoh configuration for a client connecting to a local server. @@ -27,7 +36,7 @@ def create_zenoh_config(network_discovery: bool = True) -> zenoh.Config: return config -def open_zenoh_session() -> zenoh.Session: +def open_zenoh_session() -> "zenoh.Session": """ Open a Zenoh session with a local connection first, then fall back to network discovery.