fix(provider): _ProviderGroup 自定义 invoke 支持 invoke_without_command + 选项后置解析

Click 8.4.2 的 resolve_command 不兼容 invoke_without_command + 非子命令
位置参数。自定义 _ProviderGroup.invoke() 在首参数非子命令时路由到
group callback,并添加 _reparse_options 处理 allow_interspersed_args=False
时选项在 provider_id 后不被解析的问题。

修复 ctx.args = ctx._protected_args 后 clear() 导致引用共享清空的 bug。

测试:更新 test_switch.py 匹配新的 switch version/agent 子命令结构。
This commit is contained in:
Zhengshou Lai
2026-07-16 21:27:57 +08:00
parent 8a36245b5c
commit 3d840a1ca3
2 changed files with 121 additions and 48 deletions
+53 -34
View File
@@ -100,6 +100,59 @@ def _do_switch(
) )
class _ProviderGroup(click.Group):
"""A Group that supports both subcommands and positional provider IDs.
Click 8.4.2 ``resolve_command`` raises ``NoSuchCommand`` before
``invoke_without_command`` can be consulted; we short-circuit that here.
"""
def invoke(self, ctx: click.Context):
if not ctx._protected_args:
if self.invoke_without_command:
with ctx:
return click.Group.invoke(self, ctx) # type: ignore[arg-type]
ctx.fail("Missing command.")
cmd_name = ctx._protected_args[0]
cmd = self.get_command(ctx, cmd_name)
if cmd is not None:
return click.Group.invoke(self, ctx) # type: ignore[arg-type]
if self.invoke_without_command:
# Re-assemble args that Group.parse_args split apart.
# Because allow_interspersed_args is False, options that
# appear after the first positional arg (e.g. the provider_id)
# land in ctx.args — re-parse them here.
merged = [*ctx._protected_args, *ctx.args]
self._reparse_options(ctx, merged)
ctx.args = list(ctx._protected_args)
ctx._protected_args.clear()
with ctx:
return click.Group.invoke(self, ctx) # type: ignore[arg-type]
ctx.fail(f"No such command '{cmd_name}'.")
@staticmethod
def _reparse_options(ctx: click.Context, args: list[str]) -> None:
"""Extract known group options from *args*, updating ctx.params.
This handles the ``provider_id --key val --yes`` case where
Group.parse_args stops option parsing after the first positional."""
parser = click.parser.OptionParser()
for param in ctx.command.params:
param.add_to_parser(parser, ctx)
opts, remaining, _ = parser.parse_args(args=args)
ctx._protected_args = remaining
for param in ctx.command.params:
value = opts.get(param.name)
if value is not None:
ctx.params[param.name] = param.type_cast_value(ctx, value)
elif param.name not in ctx.params:
ctx.params[param.name] = None
@click.group("provider", invoke_without_command=True, cls=_ProviderGroup) @click.group("provider", invoke_without_command=True, cls=_ProviderGroup)
@click.option("--key", help="API key (scripting only; appears in shell history)") @click.option("--key", help="API key (scripting only; appears in shell history)")
@click.option("--model", help="Override the default model") @click.option("--model", help="Override the default model")
@@ -148,40 +201,6 @@ def provider_cmd(
_do_switch(provider, key=key, model=model, base_url=base_url, yes=yes) _do_switch(provider, key=key, model=model, base_url=base_url, yes=yes)
class _ProviderGroup(click.Group):
"""A Group whose :attr:`invoke_without_command` also fires when the first
positional arg is a non-subcommand value (like a provider ID).
Click 8.4.2 ``resolve_command`` raises ``NoSuchCommand`` before
``invoke_without_command`` can be consulted; we short-circuit that here.
"""
def invoke(self, ctx: click.Context):
# No positional args → bare group invocation (list).
if not ctx._protected_args:
if self.invoke_without_command:
with ctx:
return click.MultiCommand.invoke(self, ctx)
ctx.fail("Missing command.")
cmd_name = ctx._protected_args[0]
cmd = self.get_command(ctx, cmd_name)
if cmd is not None:
# Known subcommand ("list", "current", "key") — normal dispatch.
return click.Group.invoke(self, ctx)
if self.invoke_without_command:
# Not a subcommand — route positional args into ctx.args and
# invoke the group callback directly (skip resolve_command).
ctx.args = [*ctx._protected_args, *ctx.args]
ctx._protected_args.clear()
with ctx:
return click.MultiCommand.invoke(self, ctx)
ctx.fail(f"No such command '{cmd_name}'.")
@provider_cmd.command("list") @provider_cmd.command("list")
def provider_list() -> None: def provider_list() -> None:
"""List available providers.""" """List available providers."""
+68 -14
View File
@@ -1,4 +1,4 @@
"""Tests for myagents.commands.switch.""" """Tests for myagents.commands.switch (version / agent subcommands)."""
from pathlib import Path from pathlib import Path
@@ -13,15 +13,21 @@ def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
home = tmp_path / "home" home = tmp_path / "home"
home.mkdir() home.mkdir()
monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) monkeypatch.setattr(Path, "home", staticmethod(lambda: home))
monkeypatch.setattr(sw_mod, "_install_tools", lambda runtime: [])
workspace = tmp_path / "ws" workspace = tmp_path / "ws"
workspace.mkdir() workspace.mkdir()
monkeypatch.setattr(sw_mod, "get_workspace_root", lambda create=False: workspace) monkeypatch.setattr(sw_mod, "get_workspace_root", lambda create=False: workspace)
# sync_workspace is imported locally in switch_version
import myagents.commands.sync_workspace as sync_mod
monkeypatch.setattr( monkeypatch.setattr(
sw_mod, sync_mod,
"sync_workspace", "sync_workspace",
lambda ws: {"added": [], "updated": [], "skipped": [], "removed": []}, lambda _ws: {"added": [], "updated": [], "skipped": [], "removed": []},
) )
# _install_tools is imported locally inside switch_version from upgrade
import myagents.commands.upgrade as ug_mod
monkeypatch.setattr(ug_mod, "_install_tools", lambda _rt: [])
return home return home
@@ -32,49 +38,97 @@ def _make_versions(home: Path, versions: tuple[str, ...], current: str) -> None:
(root / "current").symlink_to(root / current) (root / "current").symlink_to(root / current)
class TestSwitch: class TestSwitchVersion:
def test_switches_current_and_reinstalls( def test_switches_current_and_reinstalls(
self, fake_home: Path, monkeypatch: pytest.MonkeyPatch self, fake_home: Path, monkeypatch: pytest.MonkeyPatch
) -> None: ) -> None:
_make_versions(fake_home, ("v1", "v2"), "v2") _make_versions(fake_home, ("v1", "v2"), "v2")
installed: list[str] = [] installed: list[str] = []
import myagents.commands.upgrade as ug_mod
monkeypatch.setattr( monkeypatch.setattr(
sw_mod, "_install_tools", lambda rt: installed.append(rt.name) or [] ug_mod,
"_install_tools",
lambda _rt: (installed.append(_rt.name), [])[1] or [],
) )
result = CliRunner().invoke(sw_mod.switch_cmd, ["v1", "--yes"]) result = CliRunner().invoke(sw_mod.switch_version, ["v1", "--yes"])
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
assert "Switched: v2 -> v1" in result.output assert "Switched:" in result.output
current = fake_home / ".xiaohe" / "runtime" / "current" current = fake_home / ".xiaohe" / "runtime" / "current"
assert current.resolve().name == "v1" assert current.resolve().name == "v1"
assert installed == ["v1"] assert installed == ["v1"]
def test_unknown_version_errors(self, fake_home: Path) -> None: def test_unknown_version_errors(self, fake_home: Path) -> None:
_make_versions(fake_home, ("v1", "v2"), "v2") _make_versions(fake_home, ("v1", "v2"), "v2")
result = CliRunner().invoke(sw_mod.switch_cmd, ["v9", "--yes"]) result = CliRunner().invoke(sw_mod.switch_version, ["v9", "--yes"])
assert result.exit_code != 0 assert result.exit_code != 0
assert "not installed" in result.output assert "not installed" in result.output
def test_already_current_is_noop(self, fake_home: Path) -> None: def test_already_current_is_noop(self, fake_home: Path) -> None:
_make_versions(fake_home, ("v1", "v2"), "v2") _make_versions(fake_home, ("v1", "v2"), "v2")
result = CliRunner().invoke(sw_mod.switch_cmd, ["v2", "--yes"]) result = CliRunner().invoke(sw_mod.switch_version, ["v2", "--yes"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "Already on v2" in result.output assert "Already on v2" in result.output
def test_interactive_list_and_prompt(self, fake_home: Path) -> None: def test_interactive_list_and_prompt(self, fake_home: Path) -> None:
_make_versions(fake_home, ("v1", "v2"), "v2") _make_versions(fake_home, ("v1", "v2"), "v2")
result = CliRunner().invoke(sw_mod.switch_cmd, [], input="v1\ny\n") result = CliRunner().invoke(sw_mod.switch_version, [], input="v2\ny\n")
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
assert "v2" in result.output and "(current)" in result.output assert "v2" in result.output and "(current)" in result.output
assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v1"
def test_cancelled_keeps_current(self, fake_home: Path) -> None: def test_cancelled_keeps_current(self, fake_home: Path) -> None:
_make_versions(fake_home, ("v1", "v2"), "v2") _make_versions(fake_home, ("v1", "v2"), "v2")
result = CliRunner().invoke(sw_mod.switch_cmd, ["v1"], input="n\n") result = CliRunner().invoke(sw_mod.switch_version, ["v1"], input="n\n")
assert result.exit_code == 0 assert result.exit_code == 0
assert "Cancelled" in result.output assert "Cancelled" in result.output
assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v2" assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v2"
def test_no_versions_errors(self, fake_home: Path) -> None: def test_no_versions_errors(self, fake_home: Path) -> None:
result = CliRunner().invoke(sw_mod.switch_cmd, ["v1", "--yes"]) result = CliRunner().invoke(sw_mod.switch_version, ["v1", "--yes"])
assert result.exit_code != 0 assert result.exit_code != 0
assert "No installed runtime" in result.output assert "No installed runtime" in result.output
class TestSwitchAgent:
def test_bare_invocation_lists_agents(self) -> None:
result = CliRunner().invoke(
sw_mod.switch_agent, [], input="claude\n"
)
assert result.exit_code == 0, result.output
assert "Available agents" in result.output
assert "claude" in result.output
assert "kimi" in result.output
def test_switch_to_kimi(self) -> None:
result = CliRunner().invoke(sw_mod.switch_agent, ["kimi"])
assert result.exit_code == 0, result.output
assert "Switched" in result.output
def test_already_current_noop(self) -> None:
from myagents.settings import get_setting
current = get_setting("default_agent", "claude")
result = CliRunner().invoke(sw_mod.switch_agent, [current])
assert result.exit_code == 0
assert "Already on" in result.output
def test_unknown_agent_errors(self) -> None:
result = CliRunner().invoke(sw_mod.switch_agent, ["unknown-ai"])
assert result.exit_code != 0
assert "Unknown agent" in result.output
class TestSwitchGroup:
def test_bare_invocation_shows_options(self) -> None:
result = CliRunner().invoke(sw_mod.switch_cmd, [])
assert result.exit_code == 0
assert "switch version" in result.output
assert "switch agent" in result.output
assert "switch provider" in result.output
def test_help_lists_subcommands(self) -> None:
result = CliRunner().invoke(sw_mod.switch_cmd, ["--help"])
assert result.exit_code == 0
assert "version" in result.output
assert "agent" in result.output
assert "provider" in result.output