Files
pyflowx/tests/cli/test_system_run.py
T
zhou 1f7127357e
CI / Lint, Typecheck & Test (push) Has been cancelled
refactor: 将 cli/system、cli/llm、cli/dev 脚本迁移为 YAML 配置 + register_fn 模式
删除 cli/system (clearscreen/taskkill/which)、cli/llm (msdownload/sglang)、cli/dev (dockercmd/envdev) 三个目录; 新增 7 个 YAML 配置; 新增 ops/llm.py 模块; 扩展 ops/system.py 和 ops/dev.py; pf.py 添加工具别名; 删除 examples/ 目录
2026-07-05 21:12:41 +08:00

144 lines
6.5 KiB
Python

"""Tests for ops.system 清屏/进程终止/命令查找函数."""
from __future__ import annotations
import subprocess
from unittest.mock import MagicMock
import pytest
from pyflowx.conditions import Constants
from pyflowx.ops.system import clear_screen_run, taskkill_run, which_run
# ---------------------------------------------------------------------- #
# clear_screen_run
# ---------------------------------------------------------------------- #
class TestClearScreenRun:
"""``clear_screen_run`` 函数测试."""
def test_windows_uses_cls(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Windows 上应调用 cls."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
ran_cmds: list[list[str]] = []
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
clear_screen_run()
assert ran_cmds == [["cls"]]
def test_non_windows_uses_clear(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""非 Windows 上应调用 clear."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
ran_cmds: list[list[str]] = []
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
clear_screen_run()
assert ran_cmds == [["clear"]]
# ---------------------------------------------------------------------- #
# taskkill_run
# ---------------------------------------------------------------------- #
class TestTaskkillRun:
"""``taskkill_run`` 函数测试."""
def test_windows_uses_taskkill(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Windows 上应使用 taskkill /f /im 并加通配符."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
ran_cmds: list[list[str]] = []
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
taskkill_run(["chrome.exe", "python.exe"])
assert ran_cmds == [
["taskkill", "/f", "/im", "chrome.exe*"],
["taskkill", "/f", "/im", "python.exe*"],
]
def test_non_windows_uses_pkill(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""非 Windows 上应使用 pkill -f 并加通配符."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
ran_cmds: list[list[str]] = []
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
taskkill_run(["chrome"])
assert ran_cmds == [["pkill", "-f", "chrome*"]]
def test_empty_list_does_nothing(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""空列表不应调用 subprocess."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
called: list[list[str]] = []
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd) or MagicMock())
taskkill_run([])
assert called == []
def test_prints_progress(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""应打印每个进程的终止信息."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
monkeypatch.setattr(subprocess, "run", lambda *_, **__: MagicMock())
taskkill_run(["chrome.exe"])
captured = capsys.readouterr()
assert "chrome.exe" in captured.out
# ---------------------------------------------------------------------- #
# which_run
# ---------------------------------------------------------------------- #
class TestWhichRun:
"""``which_run`` 函数测试."""
def test_windows_uses_where(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Windows 上应调用 where."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
ran_cmds: list[list[str]] = []
mock_result = MagicMock(returncode=0, stdout="C:\\Windows\\python.exe\n")
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or mock_result)
which_run(["python"])
assert ran_cmds == [["where", "python"]]
def test_non_windows_uses_which(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""非 Windows 上应调用 which."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
ran_cmds: list[list[str]] = []
mock_result = MagicMock(returncode=0, stdout="/usr/bin/python\n")
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or mock_result)
which_run(["python"])
assert ran_cmds == [["which", "python"]]
def test_found_prints_path(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""找到命令时应打印 <cmd> -> <path>."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
mock_result = MagicMock(returncode=0, stdout="/usr/bin/python\n")
monkeypatch.setattr(subprocess, "run", lambda *_, **__: mock_result)
which_run(["python"])
captured = capsys.readouterr()
assert "python" in captured.out
assert "/usr/bin/python" in captured.out
def test_not_found_prints_message(
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""未找到命令时应打印未找到."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
mock_result = MagicMock(returncode=1, stdout="")
monkeypatch.setattr(subprocess, "run", lambda *_, **__: mock_result)
which_run(["nonexistent_cmd"])
captured = capsys.readouterr()
assert "未找到" in captured.out
def test_multiple_commands(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""多个命令应分别查找."""
monkeypatch.setattr(Constants, "IS_WINDOWS", False)
ran_cmds: list[list[str]] = []
mock_result = MagicMock(returncode=0, stdout="/usr/bin/python\n")
monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or mock_result)
which_run(["python", "pip"])
assert ran_cmds == [["which", "python"], ["which", "pip"]]
def test_windows_multiline_takes_first(
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""Windows where 返回多行时应取第一行."""
monkeypatch.setattr(Constants, "IS_WINDOWS", True)
mock_result = MagicMock(returncode=0, stdout="C:\\first\\python.exe\nC:\\second\\python.exe\n")
monkeypatch.setattr(subprocess, "run", lambda *_, **__: mock_result)
which_run(["python"])
captured = capsys.readouterr()
assert "C:\\first\\python.exe" in captured.out
assert "C:\\second" not in captured.out