diff --git a/nanobot/cli/onboard.py b/nanobot/cli/onboard.py index 4e3b6e56..e363566b 100644 --- a/nanobot/cli/onboard.py +++ b/nanobot/cli/onboard.py @@ -264,7 +264,12 @@ def _format_value(value: Any, rich: bool = True, field_name: str = "") -> str: if isinstance(value, list): return ", ".join(str(v) for v in value) if isinstance(value, dict): - return json.dumps(value) + # Handle dicts containing BaseModel instances + parts = [] + for k, v in value.items(): + formatted = _format_value(v, rich=False, field_name=str(k)) + parts.append(f"{k}: {formatted}") + return ", ".join(parts) if parts else ("[dim]not set[/dim]" if rich else "[not set]") return str(value) @@ -279,6 +284,63 @@ def _format_value_for_input(value: Any, field_type: str) -> str: return str(value) +def _validate_field_constraint(value: Any, field_info) -> str | None: + """Validate a value against Pydantic Field constraints. + + Returns an error message string if validation fails, None if valid. + Uses attribute-based detection to handle Pydantic v2 internal types. + """ + if field_info is None or not hasattr(field_info, "metadata"): + return None + + for m in field_info.metadata: + if hasattr(m, "ge") and isinstance(value, (int, float)): + if value < m.ge: + return f"Value must be >= {m.ge}" + if hasattr(m, "gt") and isinstance(value, (int, float)): + if value <= m.gt: + return f"Value must be > {m.gt}" + if hasattr(m, "le") and isinstance(value, (int, float)): + if value > m.le: + return f"Value must be <= {m.le}" + if hasattr(m, "lt") and isinstance(value, (int, float)): + if value >= m.lt: + return f"Value must be < {m.lt}" + if hasattr(m, "min_length") and hasattr(value, "__len__"): + if len(value) < m.min_length: + return f"Length must be >= {m.min_length}" + if hasattr(m, "max_length") and hasattr(value, "__len__"): + if len(value) > m.max_length: + return f"Length must be <= {m.max_length}" + + return None + + +def _get_constraint_hint(field_info) -> str: + """Derive a human-readable constraint hint from field metadata. + + Returns a string like "(0-10)" or "(>= 0)" to append to field display names. + """ + if field_info is None or not hasattr(field_info, "metadata"): + return "" + + ge_val = None + le_val = None + for m in field_info.metadata: + if hasattr(m, "ge"): + ge_val = m.ge + if hasattr(m, "le"): + le_val = m.le + + if ge_val is not None and le_val is not None: + return f" ({ge_val}-{le_val})" + if ge_val is not None: + return f" (>= {ge_val})" + if le_val is not None: + return f" (<= {le_val})" + return "" + + # --- Rich UI Components --- @@ -333,7 +395,7 @@ def _input_bool(display_name: str, current: bool | None) -> bool | None: ).ask() -def _input_text(display_name: str, current: Any, field_type: str) -> Any: +def _input_text(display_name: str, current: Any, field_type: str, field_info=None) -> Any: """Get text input and parse based on field type.""" default = _format_value_for_input(current, field_type) @@ -344,16 +406,28 @@ def _input_text(display_name: str, current: Any, field_type: str) -> Any: if field_type == "int": try: - return int(value) + parsed = int(value) except ValueError: console.print("[yellow]! Invalid number format, value not saved[/yellow]") return None + if field_info: + error = _validate_field_constraint(parsed, field_info) + if error: + console.print(f"[yellow]! {error}, value not saved[/yellow]") + return None + return parsed elif field_type == "float": try: - return float(value) + parsed = float(value) except ValueError: console.print("[yellow]! Invalid number format, value not saved[/yellow]") return None + if field_info: + error = _validate_field_constraint(parsed, field_info) + if error: + console.print(f"[yellow]! {error}, value not saved[/yellow]") + return None + return parsed elif field_type == "list": return [v.strip() for v in value.split(",") if v.strip()] elif field_type == "dict": @@ -367,7 +441,7 @@ def _input_text(display_name: str, current: Any, field_type: str) -> Any: def _input_with_existing( - display_name: str, current: Any, field_type: str + display_name: str, current: Any, field_type: str, field_info=None ) -> Any: """Handle input with 'keep existing' option for non-empty values.""" has_existing = current is not None and current != "" and current != {} and current != [] @@ -381,7 +455,7 @@ def _input_with_existing( if choice == "Keep existing value" or choice is None: return None - return _input_text(display_name, current, field_type) + return _input_text(display_name, current, field_type, field_info=field_info) # --- Pydantic Model Configuration --- @@ -568,7 +642,7 @@ def _configure_pydantic_model( field_name, field_info = fields[field_idx] current_value = getattr(working_model, field_name, None) ftype = _get_field_type_info(field_info) - field_display = _get_field_display_name(field_name, field_info) + field_display = _get_field_display_name(field_name, field_info) + _get_constraint_hint(field_info) # Nested Pydantic model - recurse if ftype.type_name == "model": @@ -610,7 +684,7 @@ def _configure_pydantic_model( if ftype.type_name == "bool": new_value = _input_bool(field_display, current_value) else: - new_value = _input_with_existing(field_display, current_value, ftype.type_name) + new_value = _input_with_existing(field_display, current_value, ftype.type_name, field_info=field_info) if new_value is not None: setattr(working_model, field_name, new_value) @@ -821,18 +895,24 @@ def _configure_channels(config: Config) -> None: _SETTINGS_SECTIONS: dict[str, tuple[str, str, set[str] | None]] = { "Agent Settings": ("Agent Defaults", "Configure default model, temperature, and behavior", None), + "Channel Common": ("Channel Common", "Configure cross-channel behavior: progress, tool hints, retries", None), + "API Server": ("API Server", "Configure OpenAI-compatible API endpoint", None), "Gateway": ("Gateway Settings", "Configure server host, port, and heartbeat", None), "Tools": ("Tools Settings", "Configure web search, shell exec, and other tools", {"mcp_servers"}), } _SETTINGS_GETTER = { "Agent Settings": lambda c: c.agents.defaults, + "Channel Common": lambda c: c.channels, + "API Server": lambda c: c.api, "Gateway": lambda c: c.gateway, "Tools": lambda c: c.tools, } _SETTINGS_SETTER = { "Agent Settings": lambda c, v: setattr(c.agents, "defaults", v), + "Channel Common": lambda c, v: setattr(c, "channels", v), + "API Server": lambda c, v: setattr(c, "api", v), "Gateway": lambda c, v: setattr(c, "gateway", v), "Tools": lambda c, v: setattr(c, "tools", v), } @@ -915,12 +995,20 @@ def _show_summary(config: Config) -> None: # Settings sections for title, model in [ ("Agent Settings", config.agents.defaults), + ("Channel Common", config.channels), + ("API Server", config.api), ("Gateway", config.gateway), ("Tools", config.tools), - ("Channel Common", config.channels), ]: _print_summary_panel(_summarize_model(model), title) + _pause() + + +def _pause() -> None: + """Pause for user acknowledgement before clearing the screen.""" + _get_questionary().text("Press Enter to continue...", default="").ask() + # --- Main Entry Point --- @@ -984,7 +1072,9 @@ def run_onboard(initial_config: Config | None = None) -> OnboardResult: choices=[ "[P] LLM Provider", "[C] Chat Channel", + "[H] Channel Common", "[A] Agent Settings", + "[I] API Server", "[G] Gateway", "[T] Tools", "[V] View Configuration Summary", @@ -1007,7 +1097,9 @@ def run_onboard(initial_config: Config | None = None) -> OnboardResult: _MENU_DISPATCH = { "[P] LLM Provider": lambda: _configure_providers(config), "[C] Chat Channel": lambda: _configure_channels(config), + "[H] Channel Common": lambda: _configure_general_settings(config, "Channel Common"), "[A] Agent Settings": lambda: _configure_general_settings(config, "Agent Settings"), + "[I] API Server": lambda: _configure_general_settings(config, "API Server"), "[G] Gateway": lambda: _configure_general_settings(config, "Gateway"), "[T] Tools": lambda: _configure_general_settings(config, "Tools"), "[V] View Configuration Summary": lambda: _show_summary(config), diff --git a/tests/agent/test_onboard_logic.py b/tests/agent/test_onboard_logic.py index 43999f93..17c1f340 100644 --- a/tests/agent/test_onboard_logic.py +++ b/tests/agent/test_onboard_logic.py @@ -22,6 +22,9 @@ from nanobot.cli.onboard import ( _format_value, _get_field_display_name, _get_field_type_info, + _get_constraint_hint, + _input_text, + _validate_field_constraint, run_onboard, ) from nanobot.config.schema import Config @@ -208,6 +211,7 @@ class TestGetFieldTypeInfo: assert inner is None + class TestGetFieldDisplayName: """Tests for _get_field_display_name human-readable name generation.""" @@ -493,3 +497,448 @@ class TestRunOnboardExitBehavior: assert result.should_save is False assert result.config.model_dump(by_alias=True) == initial_config.model_dump(by_alias=True) + + +class TestValidateFieldConstraint: + """Tests for _validate_field_constraint schema-aware input validation.""" + + def test_returns_none_when_no_constraints(self): + """Fields without constraints should pass validation.""" + from pydantic import BaseModel + + class M(BaseModel): + name: str = "hello" + + field_info = M.model_fields["name"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint("anything", field_info) is None + + def test_rejects_value_below_ge_bound(self): + """Value below ge (>=) bound should return error.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + count: int = Field(default=3, ge=0) + + field_info = M.model_fields["count"] + from nanobot.cli.onboard import _validate_field_constraint + + result = _validate_field_constraint(-1, field_info) + assert result is not None + assert "0" in result + + def test_accepts_value_at_ge_bound(self): + """Value exactly at ge (>=) bound should pass.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + count: int = Field(default=3, ge=0) + + field_info = M.model_fields["count"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint(0, field_info) is None + + def test_rejects_value_above_le_bound(self): + """Value above le (<=) bound should return error.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, le=10) + + field_info = M.model_fields["retries"] + from nanobot.cli.onboard import _validate_field_constraint + + result = _validate_field_constraint(11, field_info) + assert result is not None + assert "10" in result + + def test_accepts_value_at_le_bound(self): + """Value exactly at le (<=) bound should pass.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, le=10) + + field_info = M.model_fields["retries"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint(10, field_info) is None + + def test_combined_ge_and_le_bounds(self): + """Field with both ge and le should validate both.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, ge=0, le=10) + + field_info = M.model_fields["retries"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint(5, field_info) is None + assert _validate_field_constraint(-1, field_info) is not None + assert _validate_field_constraint(11, field_info) is not None + + def test_gt_and_lt_bounds(self): + """Strict inequality bounds (gt, lt) should exclude boundary.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + ratio: float = Field(default=0.5, gt=0.0, lt=1.0) + + field_info = M.model_fields["ratio"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint(0.5, field_info) is None + assert _validate_field_constraint(0.0, field_info) is not None + assert _validate_field_constraint(1.0, field_info) is not None + + def test_min_length_constraint(self): + """min_length should validate string/list length.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + name: str = Field(default="x", min_length=1) + + field_info = M.model_fields["name"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint("a", field_info) is None + assert _validate_field_constraint("", field_info) is not None + + def test_max_length_constraint(self): + """max_length should validate string/list length.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + tag: str = Field(default="x", max_length=5) + + field_info = M.model_fields["tag"] + from nanobot.cli.onboard import _validate_field_constraint + + assert _validate_field_constraint("abc", field_info) is None + assert _validate_field_constraint("abcdef", field_info) is not None + + def test_real_send_max_retries_field(self): + """Validate against the actual ChannelsConfig.send_max_retries field.""" + from nanobot.config.schema import ChannelsConfig + from nanobot.cli.onboard import _validate_field_constraint + + field_info = ChannelsConfig.model_fields["send_max_retries"] + assert _validate_field_constraint(3, field_info) is None + assert _validate_field_constraint(0, field_info) is None + assert _validate_field_constraint(10, field_info) is None + assert _validate_field_constraint(-1, field_info) is not None + assert _validate_field_constraint(11, field_info) is not None + + +class TestGetConstraintHint: + """Tests for _get_constraint_hint field display suffix.""" + + def test_no_constraints_returns_empty(self): + """Fields without constraints should return empty string.""" + from pydantic import BaseModel + + class M(BaseModel): + name: str = "hello" + + field_info = M.model_fields["name"] + assert _get_constraint_hint(field_info) == "" + + def test_ge_le_range(self): + """Field with ge+le should show '(min-max)'.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, ge=0, le=10) + + field_info = M.model_fields["retries"] + hint = _get_constraint_hint(field_info) + assert "0" in hint + assert "10" in hint + + def test_ge_only(self): + """Field with only ge should show '(>= N)'.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + count: int = Field(default=1, ge=0) + + field_info = M.model_fields["count"] + hint = _get_constraint_hint(field_info) + assert "0" in hint + assert ">=" in hint + + def test_le_only(self): + """Field with only le should show '(<= N)'.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + ratio: float = Field(default=1.0, le=100.0) + + field_info = M.model_fields["ratio"] + hint = _get_constraint_hint(field_info) + assert "100" in hint + assert "<=" in hint + + def test_real_send_max_retries_hint(self): + """Actual ChannelsConfig.send_max_retries should show '(0-10)'.""" + from nanobot.config.schema import ChannelsConfig + + field_info = ChannelsConfig.model_fields["send_max_retries"] + hint = _get_constraint_hint(field_info) + assert "0" in hint + assert "10" in hint + + +class TestInputTextWithValidation: + """Tests for _input_text integration with constraint validation.""" + + def test_rejects_out_of_range_int(self, monkeypatch): + """_input_text with field_info should reject values violating ge/le constraints.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, ge=0, le=10) + + field_info = M.model_fields["retries"] + monkeypatch.setattr( + onboard_wizard, + "_get_questionary", + lambda: SimpleNamespace(text=lambda *a, **kw: SimpleNamespace(ask=lambda: "15")), + ) + + result = _input_text("Retries", 3, "int", field_info=field_info) + assert result is None + + def test_accepts_valid_int(self, monkeypatch): + """_input_text with field_info should accept valid constrained values.""" + from pydantic import BaseModel, Field + + class M(BaseModel): + retries: int = Field(default=3, ge=0, le=10) + + field_info = M.model_fields["retries"] + monkeypatch.setattr( + onboard_wizard, + "_get_questionary", + lambda: SimpleNamespace(text=lambda *a, **kw: SimpleNamespace(ask=lambda: "5")), + ) + + result = _input_text("Retries", 3, "int", field_info=field_info) + assert result == 5 + + def test_works_without_field_info(self, monkeypatch): + """_input_text without field_info should work as before (no validation).""" + monkeypatch.setattr( + onboard_wizard, + "_get_questionary", + lambda: SimpleNamespace(text=lambda *a, **kw: SimpleNamespace(ask=lambda: "42")), + ) + + result = _input_text("Count", 0, "int") + assert result == 42 + + +class TestChannelCommonRegistration: + """Tests for Channel Common menu registration.""" + + def test_channel_common_in_settings_sections(self): + """Channel Common should be registered in _SETTINGS_SECTIONS.""" + from nanobot.cli.onboard import _SETTINGS_SECTIONS + + assert "Channel Common" in _SETTINGS_SECTIONS + + def test_channel_common_getter_returns_channels(self): + """Channel Common getter should return config.channels.""" + from nanobot.cli.onboard import _SETTINGS_GETTER + + config = Config() + result = _SETTINGS_GETTER["Channel Common"](config) + assert result is config.channels + + def test_channel_common_setter_writes_channels(self): + """Channel Common setter should update config.channels.""" + from nanobot.cli.onboard import _SETTINGS_SETTER + + config = Config() + original = config.channels + new_channels = original.model_copy(deep=True) + new_channels.send_tool_hints = True + _SETTINGS_SETTER["Channel Common"](config, new_channels) + assert config.channels.send_tool_hints is True + + def test_channel_common_edit_preserves_extras(self): + """Editing Channel Common should not lose per-channel extras.""" + config = Config() + config.channels.feishu = {"enabled": True, "appId": "test123"} + channels = config.channels.model_copy(deep=True) + channels.send_tool_hints = True + config.channels = channels + assert config.channels.send_tool_hints is True + assert config.channels.feishu["appId"] == "test123" + + +class TestApiServerRegistration: + """Tests for API Server menu registration.""" + + def test_api_server_in_settings_sections(self): + """API Server should be registered in _SETTINGS_SECTIONS.""" + from nanobot.cli.onboard import _SETTINGS_SECTIONS + + assert "API Server" in _SETTINGS_SECTIONS + + def test_api_server_getter_returns_api(self): + """API Server getter should return config.api.""" + from nanobot.cli.onboard import _SETTINGS_GETTER + + config = Config() + result = _SETTINGS_GETTER["API Server"](config) + assert result is config.api + + def test_api_server_setter_writes_api(self): + """API Server setter should update config.api.""" + from nanobot.cli.onboard import _SETTINGS_SETTER + + config = Config() + from nanobot.config.schema import ApiConfig + + new_api = ApiConfig(host="0.0.0.0", port=9999) + _SETTINGS_SETTER["API Server"](config, new_api) + assert config.api.host == "0.0.0.0" + assert config.api.port == 9999 + + +class TestMainMenuUpdate: + """Tests for main menu including new Channel Common and API Server items.""" + + def test_main_menu_dispatch_includes_channel_common(self): + """Main menu dispatch should route [H] to Channel Common.""" + from nanobot.cli.onboard import run_onboard + + # We verify by checking the dispatch table is set up correctly + # The menu items are defined inline in run_onboard, so we test + # that _configure_general_settings handles the new sections. + from nanobot.cli.onboard import _SETTINGS_SECTIONS, _SETTINGS_GETTER, _SETTINGS_SETTER + + assert "Channel Common" in _SETTINGS_SECTIONS + assert "Channel Common" in _SETTINGS_GETTER + assert "Channel Common" in _SETTINGS_SETTER + + def test_main_menu_dispatch_includes_api_server(self): + """Main menu dispatch should route [I] to API Server.""" + from nanobot.cli.onboard import _SETTINGS_SECTIONS, _SETTINGS_GETTER, _SETTINGS_SETTER + + assert "API Server" in _SETTINGS_SECTIONS + assert "API Server" in _SETTINGS_GETTER + assert "API Server" in _SETTINGS_SETTER + + def test_run_onboard_channel_common_edit(self, monkeypatch): + """run_onboard should handle [H] Channel Common correctly.""" + initial_config = Config() + + responses = iter([ + "[H] Channel Common", + KeyboardInterrupt(), + "[S] Save and Exit", + ]) + + class FakePrompt: + def __init__(self, response): + self.response = response + + def ask(self): + if isinstance(self.response, BaseException): + raise self.response + return self.response + + def fake_select(*_args, **_kwargs): + return FakePrompt(next(responses)) + + def fake_configure_general_settings(config, section): + if section == "Channel Common": + config.channels.send_tool_hints = True + + monkeypatch.setattr(onboard_wizard, "_show_main_menu_header", lambda: None) + monkeypatch.setattr(onboard_wizard, "questionary", SimpleNamespace(select=fake_select)) + monkeypatch.setattr(onboard_wizard, "_configure_general_settings", fake_configure_general_settings) + + result = run_onboard(initial_config=initial_config) + + assert result.should_save is True + assert result.config.channels.send_tool_hints is True + + def test_run_onboard_api_server_edit(self, monkeypatch): + """run_onboard should handle [I] API Server correctly.""" + initial_config = Config() + + responses = iter([ + "[I] API Server", + KeyboardInterrupt(), + "[S] Save and Exit", + ]) + + class FakePrompt: + def __init__(self, response): + self.response = response + + def ask(self): + if isinstance(self.response, BaseException): + raise self.response + return self.response + + def fake_select(*_args, **_kwargs): + return FakePrompt(next(responses)) + + def fake_configure_general_settings(config, section): + if section == "API Server": + config.api.port = 9999 + + monkeypatch.setattr(onboard_wizard, "_show_main_menu_header", lambda: None) + monkeypatch.setattr(onboard_wizard, "questionary", SimpleNamespace(select=fake_select)) + monkeypatch.setattr(onboard_wizard, "_configure_general_settings", fake_configure_general_settings) + + result = run_onboard(initial_config=initial_config) + + assert result.should_save is True + assert result.config.api.port == 9999 + + def test_view_summary_calls_pause(self, monkeypatch): + """[V] View Summary should pause before returning to main menu.""" + initial_config = Config() + pause_called = {"n": 0} + + responses = iter([ + "[V] View Configuration Summary", + "[S] Save and Exit", + ]) + + class FakePrompt: + def __init__(self, response): + self.response = response + + def ask(self): + if isinstance(self.response, BaseException): + raise self.response + return self.response + + def fake_select(*_args, **_kwargs): + return FakePrompt(next(responses)) + + def fake_pause(): + pause_called["n"] += 1 + + monkeypatch.setattr(onboard_wizard, "_show_main_menu_header", lambda: None) + monkeypatch.setattr(onboard_wizard, "questionary", SimpleNamespace(select=fake_select)) + # _pause is called inside _show_summary, so we patch it there + monkeypatch.setattr(onboard_wizard, "_pause", fake_pause) + # Suppress summary output but still call _pause + monkeypatch.setattr(onboard_wizard, "_print_summary_panel", lambda *a, **kw: None) + monkeypatch.setattr(onboard_wizard, "_get_provider_names", lambda: {}) + monkeypatch.setattr(onboard_wizard, "_get_channel_names", lambda: {}) + + result = run_onboard(initial_config=initial_config) + + assert result.should_save is True + assert pause_called["n"] == 1