zhou
d93da0d8b4
CI / Lint, Typecheck & Test (push) Has been cancelled
将 src/pyflowx/cli/_ops/ 整体迁移至 src/pyflowx/ops/, 与 cli/ 平级 (工具函数非 CLI 专属, 可被 YAML 任务编排通用引用). 分类保持不变: dev (git/pip/bump/autofmt), files (date/level/back/zip), media (pdf/screenshot), system (ls/pack/ssh). 同步更新 15 个引用文件 (yaml_loader + 14 个测试) 的 import 路径, README 模块结构表与 test_registry docstring.
205 lines
8.1 KiB
Python
205 lines
8.1 KiB
Python
"""Tests for cli.piptool module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from pyflowx.ops import dev
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _get_installed_packages
|
|
# ---------------------------------------------------------------------- #
|
|
class TestGetInstalledPackages:
|
|
"""Test _get_installed_packages function."""
|
|
|
|
def test_get_installed_packages_success(self) -> None:
|
|
"""Should get installed packages."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="numpy==1.0.0\npandas==2.0.0\n", returncode=0)
|
|
result = dev._get_installed_packages()
|
|
assert "numpy" in result
|
|
assert "pandas" in result
|
|
|
|
def test_get_installed_packages_empty(self) -> None:
|
|
"""Should handle empty output."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="", returncode=0)
|
|
result = dev._get_installed_packages()
|
|
assert result == []
|
|
|
|
def test_get_installed_packages_error(self) -> None:
|
|
"""Should handle subprocess error."""
|
|
with patch("subprocess.run", side_effect=subprocess.SubprocessError):
|
|
result = dev._get_installed_packages()
|
|
assert result == []
|
|
|
|
def test_get_installed_packages_oserror(self) -> None:
|
|
"""Should handle OSError."""
|
|
with patch("subprocess.run", side_effect=OSError):
|
|
result = dev._get_installed_packages()
|
|
assert result == []
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _expand_wildcard_packages
|
|
# ---------------------------------------------------------------------- #
|
|
class TestExpandWildcardPackages:
|
|
"""Test _expand_wildcard_packages function."""
|
|
|
|
def test_expand_wildcard_no_pattern(self) -> None:
|
|
"""Should return package name when no wildcard."""
|
|
result = dev._expand_wildcard_packages("numpy")
|
|
assert result == ["numpy"]
|
|
|
|
def test_expand_wildcard_with_star(self) -> None:
|
|
"""Should expand wildcard with star."""
|
|
with patch.object(dev, "_get_installed_packages", return_value=["numpy", "numpy-core", "pandas"]):
|
|
result = dev._expand_wildcard_packages("numpy*")
|
|
assert "numpy" in result
|
|
assert "numpy-core" in result
|
|
|
|
def test_expand_wildcard_with_question(self) -> None:
|
|
"""Should expand wildcard with question mark."""
|
|
with patch.object(dev, "_get_installed_packages", return_value=["numpy", "numba"]):
|
|
result = dev._expand_wildcard_packages("num??")
|
|
assert len(result) > 0
|
|
|
|
def test_expand_wildcard_no_match(self) -> None:
|
|
"""Should return empty list when no match."""
|
|
with patch.object(dev, "_get_installed_packages", return_value=["pandas", "scipy"]):
|
|
result = dev._expand_wildcard_packages("numpy*")
|
|
assert result == []
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _filter_protected_packages
|
|
# ---------------------------------------------------------------------- #
|
|
class TestFilterProtectedPackages:
|
|
"""Test _filter_protected_packages function."""
|
|
|
|
def test_filter_protected_packages_normal(self) -> None:
|
|
"""Should filter protected packages."""
|
|
result = dev._filter_protected_packages(["numpy", "pandas", "pyflowx"])
|
|
assert "numpy" in result
|
|
assert "pandas" in result
|
|
assert "pyflowx" not in result
|
|
|
|
def test_filter_protected_packages_all_protected(self) -> None:
|
|
"""Should filter all protected packages."""
|
|
result = dev._filter_protected_packages(["pyflowx", "bitool"])
|
|
assert result == []
|
|
|
|
def test_filter_protected_packages_case_insensitive(self) -> None:
|
|
"""Should filter case insensitive."""
|
|
result = dev._filter_protected_packages(["PyFlowX", "BITOOL"])
|
|
assert result == []
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# pip_uninstall
|
|
# ---------------------------------------------------------------------- #
|
|
class TestPipUninstall:
|
|
"""Test pip_uninstall function."""
|
|
|
|
def test_pip_uninstall_single_package(self) -> None:
|
|
"""Should uninstall single package."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_uninstall(["numpy"])
|
|
assert mock_run.called
|
|
|
|
def test_pip_uninstall_multiple_packages(self) -> None:
|
|
"""Should uninstall multiple packages."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_uninstall(["numpy", "pandas", "scipy"])
|
|
# Should call pip uninstall
|
|
assert mock_run.called
|
|
|
|
def test_pip_uninstall_with_wildcard(self) -> None:
|
|
"""Should handle wildcard in package name."""
|
|
with patch.object(dev, "_expand_wildcard_packages", return_value=["numpy", "numpy-core"]), patch(
|
|
"subprocess.run"
|
|
) as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_uninstall(["numpy*"])
|
|
assert mock_run.called
|
|
|
|
def test_pip_uninstall_empty_packages(self) -> None:
|
|
"""Should handle empty packages list."""
|
|
with patch.object(dev, "_expand_wildcard_packages", return_value=[]):
|
|
dev.pip_uninstall(["nonexistent*"])
|
|
# Should not call subprocess.run
|
|
|
|
def test_pip_uninstall_all_protected(self) -> None:
|
|
"""Should handle all protected packages."""
|
|
dev.pip_uninstall(["pyflowx"])
|
|
# Should not call subprocess.run
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# pip_reinstall
|
|
# ---------------------------------------------------------------------- #
|
|
class TestPipReinstall:
|
|
"""Test pip_reinstall function."""
|
|
|
|
def test_pip_reinstall_single_package(self) -> None:
|
|
"""Should reinstall single package."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_reinstall(["numpy"])
|
|
# Should call pip uninstall and pip install
|
|
assert mock_run.call_count == 2
|
|
|
|
def test_pip_reinstall_offline(self) -> None:
|
|
"""Should reinstall packages offline."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_reinstall(["numpy"], offline=True)
|
|
# Should call pip install with offline flags
|
|
assert mock_run.called
|
|
|
|
def test_pip_reinstall_all_protected(self) -> None:
|
|
"""Should handle all protected packages."""
|
|
dev.pip_reinstall(["pyflowx"])
|
|
# Should not call subprocess.run
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# pip_download
|
|
# ---------------------------------------------------------------------- #
|
|
class TestPipDownload:
|
|
"""Test pip_download function."""
|
|
|
|
def test_pip_download_single_package(self) -> None:
|
|
"""Should download single package."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_download(["numpy"])
|
|
assert mock_run.called
|
|
|
|
def test_pip_download_offline(self) -> None:
|
|
"""Should download packages offline."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
dev.pip_download(["numpy"], offline=True)
|
|
# Should call pip download with offline flags
|
|
assert mock_run.called
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# pip_freeze
|
|
# ---------------------------------------------------------------------- #
|
|
class TestPipFreeze:
|
|
"""Test pip_freeze function."""
|
|
|
|
def test_pip_freeze(self, tmp_path: Path) -> None:
|
|
"""Should freeze dependencies."""
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(stdout="numpy==1.0.0\npandas==2.0.0", returncode=0)
|
|
dev.pip_freeze()
|
|
assert mock_run.called
|