pyflowx/tests/test_shell.py

"""px.sh 命令执行封装的测试。"""

from __future__ import annotations

import subprocess
from typing import Any

import pyflowx as px
from pyflowx.shell import sh


class TestShListInput:
    """list[str] 输入（直接执行，非 shell）。"""

    def test_sh_list_success(self) -> None:
        """list[str] 命令成功执行，返回 CompletedProcess。"""
        result = sh(["true"])
        assert result is not None
        assert result.returncode == 0

    def test_sh_list_success_with_capture(self) -> None:
        """capture=True 时捕获 stdout。"""
        result = sh(["echo", "hello"], capture=True, label="echo")
        assert result is not None
        assert result.stdout == "hello\n"

    def test_sh_list_no_capture_stdout_is_none(self) -> None:
        """capture=False 时 stdout 为 None（输出到终端）。"""
        result = sh(["true"])
        assert result is not None
        assert result.stdout is None

    def test_sh_list_failure_returns_none(self, capsys: Any) -> None:
        """list[str] 命令失败（非零返回码）返回 None + 打印中文错误。"""
        result = sh(["false"], label="测试命令")
        assert result is None
        captured = capsys.readouterr()
        assert "测试命令失败" in captured.out
        assert "returncode=" in captured.out

    def test_sh_list_file_not_found_returns_none(self, capsys: Any) -> None:
        """list[str] 命令不存在返回 None + 打印中文错误。"""
        result = sh(["this_command_does_not_exist_xyz"], label="缺失命令")
        assert result is None
        captured = capsys.readouterr()
        assert "缺失命令失败" in captured.out
        assert "命令未找到" in captured.out


class TestShStrInput:
    """str 输入（shell=True）。"""

    def test_sh_str_success(self) -> None:
        """str 命令成功执行（shell=True）。"""
        result = sh("true")
        assert result is not None
        assert result.returncode == 0

    def test_sh_str_with_pipe(self) -> None:
        """str 支持 shell 管道语法。"""
        result = sh("echo hello | tr a-z A-Z", capture=True, label="管道")
        assert result is not None
        assert result.stdout == "HELLO\n"

    def test_sh_str_with_redirect(self, tmp_path: Any) -> None:
        """str 支持 shell 重定向语法。"""
        out_file = tmp_path / "out.txt"
        sh(f"echo content > {out_file}")
        assert out_file.read_text() == "content\n"

    def test_sh_str_failure_returns_none(self, capsys: Any) -> None:
        """str 命令失败返回 None + 打印中文错误。"""
        result = sh("false", label="shell 命令")
        assert result is None
        captured = capsys.readouterr()
        assert "shell 命令失败" in captured.out

    def test_sh_str_capture_stderr(self) -> None:
        """capture=True 时捕获 stderr。"""
        result = sh("echo err >&2", capture=True, label="stderr")
        assert result is not None
        assert result.stderr == "err\n"


class TestShMocked:
    """用 monkeypatch mock subprocess.run 验证参数传递。"""

    def test_sh_list_passes_shell_false(self, monkeypatch: Any) -> None:
        """list[str] 输入时 shell=False。"""
        captured: dict[str, Any] = {}

        def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
            captured["cmd"] = cmd
            captured["shell"] = kwargs.get("shell")
            captured["capture_output"] = kwargs.get("capture_output")
            captured["text"] = kwargs.get("text")
            captured["check"] = kwargs.get("check")
            return subprocess.CompletedProcess(cmd, 0)

        monkeypatch.setattr(subprocess, "run", fake_run)
        sh(["ls", "-la"])
        assert captured["cmd"] == ["ls", "-la"]
        assert captured["shell"] is False
        assert captured["capture_output"] is False
        assert captured["text"] is True
        assert captured["check"] is True

    def test_sh_str_passes_shell_true(self, monkeypatch: Any) -> None:
        """str 输入时 shell=True。"""
        captured: dict[str, Any] = {}

        def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
            captured["cmd"] = cmd
            captured["shell"] = kwargs.get("shell")
            return subprocess.CompletedProcess(cmd, 0)

        monkeypatch.setattr(subprocess, "run", fake_run)
        sh("ls -la | grep foo")
        assert captured["cmd"] == "ls -la | grep foo"
        assert captured["shell"] is True

    def test_sh_capture_passes_capture_output_true(self, monkeypatch: Any) -> None:
        """capture=True 时 capture_output=True。"""
        captured: dict[str, Any] = {}

        def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
            captured["capture_output"] = kwargs.get("capture_output")
            return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")

        monkeypatch.setattr(subprocess, "run", fake_run)
        sh(["echo"], capture=True)
        assert captured["capture_output"] is True

    def test_sh_called_process_error_returns_none(self, monkeypatch: Any, capsys: Any) -> None:
        """CalledProcessError 被捕获，打印中文错误并返回 None。"""
        def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
            raise subprocess.CalledProcessError(returncode=1, cmd=cmd)

        monkeypatch.setattr(subprocess, "run", fake_run)
        result = sh(["failing"], label="自定义标签")
        assert result is None
        captured = capsys.readouterr()
        assert "自定义标签失败" in captured.out
        assert "returncode=1" in captured.out

    def test_sh_file_not_found_error_returns_none(self, monkeypatch: Any, capsys: Any) -> None:
        """FileNotFoundError 被捕获，打印中文错误并返回 None。"""
        def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
            raise FileNotFoundError("[Errno 2] No such file")

        monkeypatch.setattr(subprocess, "run", fake_run)
        result = sh(["missing"], label="缺失工具")
        assert result is None
        captured = capsys.readouterr()
        assert "缺失工具失败" in captured.out
        assert "命令未找到" in captured.out


class TestShExported:
    """验证 px.sh 顶层导出。"""

    def test_sh_is_exported_from_px(self) -> None:
        """px.sh 等价于 pyflowx.shell.sh。"""
        assert px.sh is sh