Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(wizard): add Channel Common, API Server menus and field constraint validation

Browse Source 
- Add [H] Channel Common menu to configure send_progress, send_tool_hints,
  send_max_retries, and transcription_provider
- Add [I] API Server menu to configure host, port, timeout
- Add real-time Pydantic field constraint validation (ge/gt/le/lt/min_length/max_length)
  with constraint hints shown in field display (e.g. "Send Max Retries (0-10)")
- Add _pause() to View Configuration Summary to prevent immediate screen clear
- Fix _format_value dict branch to handle BaseModel instances without crashing
This commit is contained in:
chengyongru
2026-04-18 21:56:10 +08:00
committed by Xubin Ren
parent 58110afb88
commit ebb5179cab
2 changed files with 550 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/cli/onboard.py
+101 -9
View File
@@ -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),
tests/agent/test_onboard_logic.py
+449
View File
@@ -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