From 6c51e3555f2f1c0269882dde61dcfb39955ab6af Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 17 Oct 2026 03:35:40 +0800 Subject: [PATCH] fix(cli): validate state records and type managed process status --- app/cli.py | 50 +++++++++++++++++++++++----------------- docs/cli.md | 2 +- tests/test_cli_state.py | 51 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 22 deletions(-) create mode 100644 tests/test_cli_state.py diff --git a/app/cli.py b/app/cli.py index 1e52733ee..fc5cf97c0 100644 --- a/app/cli.py +++ b/app/cli.py @@ -8,7 +8,7 @@ import sys import time from collections import deque from pathlib import Path -from typing import Any, Dict, Iterable, Optional, get_args, get_origin +from typing import Any, Dict, Iterable, Literal, Optional, Union, get_args, get_origin from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen @@ -57,7 +57,8 @@ PREPARED_UPDATE_STATE = PREPARED_UPDATE_ROOT / "state.json" def _repo_root() -> Path: """返回后端仓库根目录。""" - return get_runtime_setting("ROOT_PATH") + root: Path = get_runtime_setting("ROOT_PATH") + return root def _read_json_file(path: Path) -> Optional[Dict[str, Any]]: @@ -65,7 +66,8 @@ def _read_json_file(path: Path) -> Optional[Dict[str, Any]]: if not path.exists(): return None try: - return json.loads(path.read_text(encoding="utf-8", errors="replace")) + payload = json.loads(path.read_text(encoding="utf-8", errors="replace")) + return payload if isinstance(payload, dict) else None except (OSError, json.JSONDecodeError): return None @@ -143,7 +145,8 @@ def _frontend_base_url(runtime: Optional[Dict[str, Any]] = None) -> str: def _runtime_api_token(runtime: Optional[Dict[str, Any]] = None) -> str: """从运行记录或当前配置读取本地 API 凭据。""" runtime = runtime or _backend_runtime() or {} - return runtime.get("api_token") or get_runtime_setting("API_TOKEN") + api_token: str = runtime.get("api_token") or get_runtime_setting("API_TOKEN") + return api_token def _http_request( @@ -575,8 +578,9 @@ def _ensure_frontend_not_running_alone(timeout: int) -> None: _stop_frontend_service(timeout=timeout, force=True) -def _managed_backend_status() -> tuple[ - str, Optional[Dict[str, Any]], Optional[psutil.Process], Optional[Dict[str, Any]] +def _managed_backend_status() -> Union[ + tuple[Literal["running", "starting"], Optional[Dict[str, Any]], psutil.Process, Optional[Dict[str, Any]]], + tuple[Literal["stopped", "running-unmanaged"], None, None, Optional[Dict[str, Any]]], ]: """汇总受管后端的进程身份和健康状态。""" runtime = _backend_runtime() @@ -596,8 +600,9 @@ def _managed_backend_status() -> tuple[ return "stopped", None, None, None -def _managed_frontend_status() -> tuple[ - str, Optional[Dict[str, Any]], Optional[psutil.Process], Optional[Dict[str, Any]] +def _managed_frontend_status() -> Union[ + tuple[Literal["running", "starting"], Optional[Dict[str, Any]], psutil.Process, Optional[Dict[str, Any]]], + tuple[Literal["stopped", "running-unmanaged"], None, None, Optional[Dict[str, Any]]], ]: """汇总受管前端的进程身份和健康状态。""" runtime = _frontend_runtime() @@ -650,7 +655,8 @@ def _annotation_name(annotation: Any) -> str: origin = get_origin(annotation) if origin is None: if hasattr(annotation, "__name__"): - return annotation.__name__ + name: str = annotation.__name__ + return name return str(annotation).replace("typing.", "") args = [arg for arg in get_args(annotation) if arg is not type(None)] @@ -867,7 +873,7 @@ def _spawn_process( cwd: Path, log_file: Optional[Path], env: Optional[Dict[str, str]] = None, -) -> subprocess.Popen: +) -> subprocess.Popen[bytes]: """启动受管子进程并将输出写入指定日志。""" kwargs: Dict[str, Any] = { "cwd": str(cwd), @@ -883,14 +889,14 @@ def _spawn_process( else: kwargs["stdout"] = subprocess.DEVNULL kwargs["stderr"] = subprocess.DEVNULL - if os.name == "nt": + if sys.platform == "win32": kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS else: kwargs["start_new_session"] = True return subprocess.Popen(command, **kwargs) -def _spawn_backend_process(*, safe: bool = False) -> subprocess.Popen: +def _spawn_backend_process(*, safe: bool = False) -> subprocess.Popen[bytes]: """构造后端启动命令并创建受管进程。""" backend_env = { **os.environ, @@ -936,7 +942,7 @@ def _ensure_frontend_runtime() -> None: raise click.ClickException("前端运行依赖未安装，请重新执行 `moviepilot install frontend` 或 `moviepilot setup`") -def _spawn_frontend_process(backend_port: int) -> subprocess.Popen: +def _spawn_frontend_process(backend_port: int) -> subprocess.Popen[bytes]: """构造前端启动命令并创建受管进程。""" _ensure_frontend_runtime() node_bin = _frontend_node_binary() @@ -1289,8 +1295,10 @@ def apply_prepared_update() -> None: @cli.command(context_settings=CONTEXT_SETTINGS) def status() -> None: """查看本地 MoviePilot 前后端服务状态""" - backend_state, backend_runtime, backend_process, backend_health = _managed_backend_status() - frontend_state, frontend_runtime, frontend_process, frontend_health = _managed_frontend_status() + backend_status = _managed_backend_status() + backend_state, backend_runtime, _, backend_health = backend_status + frontend_status = _managed_frontend_status() + frontend_state, frontend_runtime, _, frontend_health = frontend_status if backend_state == "stopped" and frontend_state == "stopped": click.echo("MoviePilot 未运行") @@ -1300,9 +1308,9 @@ def status() -> None: return click.echo("Backend:") - if backend_state == "stopped": + if backend_status[0] == "stopped": click.echo(" stopped") - elif backend_state == "running-unmanaged": + elif backend_status[0] == "running-unmanaged": data = (backend_health or {}).get("data") or {} click.echo(" running (unmanaged)") click.echo(f" URL: {_backend_base_url()}") @@ -1310,19 +1318,19 @@ def status() -> None: else: data = (backend_health or {}).get("data") or {} click.echo(f" {'running' if backend_state == 'running' else 'starting'}") - click.echo(f" PID: {backend_process.pid}") + click.echo(f" PID: {backend_status[2].pid}") click.echo(f" URL: {_backend_base_url(backend_runtime)}") click.echo(f" Version: {data.get('BACKEND_VERSION', get_app_version())}") click.echo(f" App Log: {BACKEND_APP_LOG_FILE}") click.echo(f" Stdout Log: {BACKEND_STDIO_LOG_FILE}") click.echo("Frontend:") - if frontend_state == "stopped": + if frontend_status[0] == "stopped": click.echo(" stopped") installed_frontend = _installed_frontend_version() if installed_frontend: click.echo(f" Installed Version: {installed_frontend}") - elif frontend_state == "running-unmanaged": + elif frontend_status[0] == "running-unmanaged": frontend_version = ( ((frontend_health or {}).get("version") if isinstance(frontend_health, dict) else None) or _installed_frontend_version() @@ -1338,7 +1346,7 @@ def status() -> None: or "unknown" ) click.echo(f" {'running' if frontend_state == 'running' else 'starting'}") - click.echo(f" PID: {frontend_process.pid}") + click.echo(f" PID: {frontend_status[2].pid}") click.echo(f" URL: {_frontend_base_url(frontend_runtime)}") click.echo(f" Version: {frontend_version}") click.echo(f" Stdout Log: {FRONTEND_STDIO_LOG_FILE}") diff --git a/docs/cli.md b/docs/cli.md index 2cf6e7520..d949d8f25 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -409,7 +409,7 @@ moviepilot agent --new-session 帮我总结当前系统配置有什么明显问 ## 服务管理命令 -`moviepilot start/stop/restart/status` 统一管理前后端。 +`moviepilot start/stop/restart/status` 统一管理前后端。本地状态文件必须是 JSON 对象；无法解析或包含数组、字符串等非对象值时，按无有效状态记录处理，再按既有流程探测服务。 启动、停止、重启与状态： diff --git a/tests/test_cli_state.py b/tests/test_cli_state.py new file mode 100644 index 000000000..fd2914b50 --- /dev/null +++ b/tests/test_cli_state.py @@ -0,0 +1,51 @@ +"""CLI 状态文件格式和受管进程展示的回归。""" + +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest +from click.testing import CliRunner + +from app import cli as cli_module + + +@pytest.mark.parametrize("payload", [None, [], [1], "invalid", 3, True]) +def test_non_object_state_file_is_ignored(tmp_path: Path, payload: object) -> None: + """状态文件即使是合法 JSON，也必须是对象才能交给进程及更新状态消费者。""" + path = tmp_path / "state.json" + path.write_text(json.dumps(payload), encoding="utf-8") + assert cli_module._read_json_file(path) is None + + +def test_object_state_file_preserves_contents(tmp_path: Path) -> None: + """有效状态文件保留原有字段及嵌套内容。""" + path = tmp_path / "state.json" + payload = {"pid": 123, "targets": ["resources"], "metadata": {"ok": True}} + path.write_text(json.dumps(payload), encoding="utf-8") + assert cli_module._read_json_file(path) == payload + + +@pytest.mark.parametrize("backend_state", ["running", "starting", "running-unmanaged", "stopped"]) +@pytest.mark.parametrize("frontend_state", ["running", "starting", "running-unmanaged", "stopped"]) +def test_status_only_reads_managed_process_ids( + monkeypatch: pytest.MonkeyPatch, backend_state: str, frontend_state: str +) -> None: + """受管服务展示 PID；停止或外部服务没有进程对象时仍可显示状态。""" + def service_status(state: str, pid: int) -> tuple: + """返回与服务探测器相同的状态及进程关联。""" + if state in {"running", "starting"}: + return state, {"pid": pid}, SimpleNamespace(pid=pid), {} + return state, None, None, {} + + monkeypatch.setattr(cli_module, "_managed_backend_status", lambda: service_status(backend_state, 1001)) + monkeypatch.setattr(cli_module, "_managed_frontend_status", lambda: service_status(frontend_state, 2002)) + monkeypatch.setattr(cli_module, "_installed_frontend_version", lambda: "test") + monkeypatch.setattr(cli_module, "_backend_base_url", lambda *_args: "http://localhost:3001") + monkeypatch.setattr(cli_module, "_frontend_base_url", lambda *_args: "http://localhost:3000") + + result = CliRunner().invoke(cli_module.cli, ["status"]) + + assert result.exit_code == 0, result.output + assert ("PID: 1001" in result.output) == (backend_state in {"running", "starting"}) + assert ("PID: 2002" in result.output) == (frontend_state in {"running", "starting"})