zhou
de368ea810
CI / Lint, Typecheck & Test (push) Successful in 1m11s
1. 删除 13 个已有 YAML 配置的 cli .py 入口脚本, 统一通过 pf 调用 2. gittool.yaml 用 CLEAN_EXCLUDES 数组变量配置 git clean 的 -e 参数, 保留 .venv/.tox/node_modules/.idea 等目录避免误删 3. run_cli 执行前打印调用信息: [gittool] 执行: c 4. 更新 pyproject.toml 移除 13 个冗余 entry points, 仅保留 pf 5. 清理测试文件中的 TestMain 类 (测 _ops 模块的测试保留)
98 lines
4.2 KiB
Python
98 lines
4.2 KiB
Python
"""Tests for cli.sshcopyid module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from pyflowx.cli._ops import system
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# ssh_copy_id
|
|
# ---------------------------------------------------------------------- #
|
|
class TestSshCopyId:
|
|
"""Test ssh_copy_id function."""
|
|
|
|
def test_ssh_copy_id_pub_key_not_exists(self, tmp_path: Path) -> None:
|
|
"""Should handle nonexistent public key."""
|
|
with patch.object(Path, "expanduser", return_value=tmp_path / "nonexistent.pub"), pytest.raises(SystemExit):
|
|
system.ssh_copy_id("localhost", "user", "password")
|
|
|
|
def test_ssh_copy_id_sshpass_not_found(self, tmp_path: Path) -> None:
|
|
"""Should handle sshpass not found."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch(
|
|
"subprocess.run", side_effect=FileNotFoundError
|
|
), pytest.raises(SystemExit):
|
|
system.ssh_copy_id("localhost", "user", "password")
|
|
|
|
def test_ssh_copy_id_timeout(self, tmp_path: Path) -> None:
|
|
"""Should handle SSH timeout."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch(
|
|
"subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 30)
|
|
), pytest.raises(SystemExit):
|
|
system.ssh_copy_id("localhost", "user", "password")
|
|
|
|
def test_ssh_copy_id_process_error(self, tmp_path: Path) -> None:
|
|
"""Should handle SSH process error."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch(
|
|
"subprocess.run", side_effect=subprocess.CalledProcessError(1, "cmd")
|
|
), pytest.raises(SystemExit):
|
|
system.ssh_copy_id("localhost", "user", "password")
|
|
|
|
def test_ssh_copy_id_success(self, tmp_path: Path) -> None:
|
|
"""Should deploy SSH key successfully."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
system.ssh_copy_id("localhost", "user", "password")
|
|
assert mock_run.called
|
|
|
|
def test_ssh_copy_id_with_custom_port(self, tmp_path: Path) -> None:
|
|
"""Should handle custom port."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
system.ssh_copy_id("localhost", "user", "password", port=2222)
|
|
# Verify port is used
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "2222" in call_args
|
|
|
|
def test_ssh_copy_id_with_custom_keypath(self, tmp_path: Path) -> None:
|
|
"""Should handle custom keypath."""
|
|
custom_key = tmp_path / "custom.pub"
|
|
custom_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=custom_key), patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
system.ssh_copy_id("localhost", "user", "password", keypath=str(custom_key))
|
|
assert mock_run.called
|
|
|
|
def test_ssh_copy_id_with_custom_timeout(self, tmp_path: Path) -> None:
|
|
"""Should handle custom timeout."""
|
|
pub_key = tmp_path / "id_rsa.pub"
|
|
pub_key.write_text("ssh-rsa AAAAB3...")
|
|
|
|
with patch.object(Path, "expanduser", return_value=pub_key), patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
system.ssh_copy_id("localhost", "user", "password", timeout=60)
|
|
# Verify timeout is used in ConnectTimeout option
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "ConnectTimeout=60" in call_args
|