fix: route plugin CLI arguments independently

This commit is contained in:
jinli.yl 2026-08-21 20:48:30 +08:00
parent 34615bffb6
commit 31f46649ff
6 changed files with 93 additions and 35 deletions

View file

@ -1,10 +1,19 @@
"""Config"""
from .config_parser import deep_merge_config, expand_env_vars, parse_args, resolve_app_config
from .config_parser import (
deep_merge_config,
expand_env_vars,
parse_action,
parse_args,
parse_kwargs,
resolve_app_config,
)
__all__ = [
"deep_merge_config",
"expand_env_vars",
"parse_action",
"parse_args",
"parse_kwargs",
"resolve_app_config",
]

View file

@ -226,8 +226,29 @@ def _strip_arg_dashes(arg: str) -> str:
return arg
def parse_args(*args) -> tuple[str, dict]:
"""Parse CLI args: first arg is action, rest are key=value pairs.
def parse_action(arg: str) -> str:
"""Parse and validate one top-level CLI action."""
action = _strip_arg_dashes(arg)
if "=" in action:
raise ValueError(f"First argument must be action, got: {arg}")
return action
def parse_kwargs(*args: str) -> dict:
"""Parse application-style ``key=value`` CLI arguments."""
kvs: list[str] = []
for raw in args:
arg = _strip_arg_dashes(raw)
if "=" in arg:
kvs.append(arg)
else:
raise ValueError(f"Invalid argument format (expected key=value): {raw}")
return parse_dot_notation(kvs) if kvs else {}
def parse_args(*args: str) -> tuple[str, dict]:
"""Parse an application CLI action followed by ``key=value`` arguments.
Usage: reme app config=paw.yaml service.name=test
Returns: (action, parsed_kv_dict)
@ -235,20 +256,7 @@ def parse_args(*args) -> tuple[str, dict]:
if not args:
raise ValueError("No arguments provided")
first = _strip_arg_dashes(args[0])
if "=" in first:
raise ValueError(f"First argument must be action, got: {args[0]}")
kvs: list[str] = []
for raw in args[1:]:
arg = _strip_arg_dashes(raw)
if "=" in arg:
kvs.append(arg)
else:
raise ValueError(f"Invalid argument format (expected key=value): {raw}")
parsed = parse_dot_notation(kvs) if kvs else {}
return first, parsed
return parse_action(args[0]), parse_kwargs(*args[1:])
def resolve_app_config(*, log_config: bool = True, **kwargs) -> dict:

View file

@ -1,11 +1,13 @@
"""ReMe memory management application entry point."""
import asyncio
from collections.abc import Sequence
from dataclasses import dataclass
import sys
from .application import Application
from .components.service.cli_service import prepare_start_config, should_precheck_start
from .config import parse_args, resolve_app_config
from .config import parse_action, parse_kwargs, resolve_app_config
from .enumeration import ComponentEnum
from .plugin import resolve_plugin_runtime
from .utils import cli_find_reme, load_env, precheck_start, running_app_config
@ -17,6 +19,21 @@ class ReMe(Application):
"""ReMe memory management application."""
@dataclass(frozen=True)
class CliInvocation:
"""A top-level CLI action with arguments in that action's own syntax."""
action: str
arguments: tuple[str, ...]
def parse_cli_invocation(argv: Sequence[str]) -> CliInvocation:
"""Parse only the grammar shared by every CLI command family."""
if not argv:
raise ValueError("No arguments provided")
return CliInvocation(action=parse_action(argv[0]), arguments=tuple(argv[1:]))
async def call_server(action: str, **kwargs):
"""Call the running server with a client matching its *actual* service config.
@ -63,26 +80,39 @@ async def call_server(action: str, **kwargs):
print()
def main():
def _run_plugin_command(argv: Sequence[str]) -> None:
"""Run local package management without initializing the application."""
from .plugin_cli import plugin_cli
status = plugin_cli(argv)
if status:
raise SystemExit(status)
def _start_application(kwargs: dict, environment: dict) -> None:
"""Resolve startup configuration and run the application."""
kwargs = prepare_start_config(kwargs)
kwargs["environment"] = environment
if should_precheck_start(kwargs) and not precheck_start(kwargs.get("service")):
return
ReMe(**kwargs).run_app()
def main() -> None:
"""Parse CLI arguments and launch the appropriate mode."""
if len(sys.argv) > 1 and sys.argv[1] == "plugins":
invocation = parse_cli_invocation(sys.argv[1:])
action = invocation.action
if action == "plugins":
# Package management is local-only and must not load application config,
# environment files, or a running service.
from .plugin_cli import plugin_cli
status = plugin_cli(sys.argv[2:])
if status:
raise SystemExit(status)
_run_plugin_command(invocation.arguments)
return
environment = load_env()
action, kwargs = parse_args(*sys.argv[1:])
kwargs = parse_kwargs(*invocation.arguments)
if action == "start":
kwargs = prepare_start_config(kwargs)
kwargs["environment"] = environment
if should_precheck_start(kwargs) and not precheck_start(kwargs.get("service")):
return
ReMe(**kwargs).run_app()
_start_application(kwargs, environment)
elif action == "find_reme":
cli_find_reme()
else:

View file

@ -135,6 +135,14 @@ def test_parse_args_rejects_non_key_value_extra_argument():
parse_args("search", "hello")
def test_parse_args_separates_action_and_application_kwargs():
"""The shared action grammar is independent from application key/value parsing."""
action, kwargs = parse_args("--search", "--query=hello", "limit=3")
assert action == "search"
assert kwargs == {"query": "hello", "limit": 3}
@pytest.mark.parametrize("item", ["=1", ".a=1", "a.=1", "a..b=1"])
def test_parse_dot_notation_rejects_empty_key_segments(item):
"""Dot notation keys cannot contain empty path segments."""

View file

@ -5,6 +5,8 @@
from pathlib import Path
from types import SimpleNamespace
import pytest
from reme import plugin_cli as plugin_cli_module
from reme import reme as reme_module
from reme.components import R
@ -217,9 +219,10 @@ def test_plugin_command_errors_are_clean(monkeypatch, capsys):
assert "Plugin 'missing' is not installed" in capsys.readouterr().err
def test_main_routes_plugins_before_loading_environment(monkeypatch):
@pytest.mark.parametrize("action", ["plugins", "--plugins"])
def test_main_routes_plugins_before_loading_environment(monkeypatch, action):
events = []
monkeypatch.setattr("sys.argv", ["reme", "plugins", "list"])
monkeypatch.setattr("sys.argv", ["reme", action, "list"])
monkeypatch.setattr(reme_module, "load_env", lambda: events.append("load_env"))
monkeypatch.setattr(plugin_cli_module, "plugin_cli", lambda argv: events.append(list(argv)) or 0)

View file

@ -98,8 +98,8 @@ def test_main_loads_env_before_calling_server(monkeypatch):
events = []
main_globals = reme_module.main.__globals__
monkeypatch.setattr("sys.argv", ["reme", "shell", "cmd=pwd"])
monkeypatch.setitem(main_globals, "load_env", lambda: events.append("load_env"))
monkeypatch.setitem(main_globals, "parse_args", lambda *_args: ("shell", {"cmd": "pwd"}))
async def fake_call_server(action, **kwargs):
events.append(("call_server", action, kwargs))
@ -127,8 +127,8 @@ def test_main_saves_loaded_environment_in_start_config(monkeypatch):
observed["ran"] = True
main_globals = reme_module.main.__globals__
monkeypatch.setattr("sys.argv", ["reme", "start"])
monkeypatch.setitem(main_globals, "load_env", lambda: {"TOOL_ENV": "configured"})
monkeypatch.setitem(main_globals, "parse_args", lambda *_args: ("start", {}))
monkeypatch.setitem(main_globals, "prepare_start_config", lambda _kwargs: {"service": {"backend": "cli"}})
monkeypatch.setitem(main_globals, "ReMe", FakeReMe)