92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from sshctl.operations.cleanup import _is_socket_in_use, cleanup_control_sockets
|
||
|
|
|
||
|
|
|
||
|
|
def test_cleanup_no_control_dir(ssh_dir: Path) -> None:
|
||
|
|
result = cleanup_control_sockets(ssh_dir)
|
||
|
|
assert result == {"total": 0, "removed": 0, "active": 0}
|
||
|
|
|
||
|
|
|
||
|
|
def test_cleanup_empty(ssh_dir: Path) -> None:
|
||
|
|
control = ssh_dir / "control.d"
|
||
|
|
control.mkdir()
|
||
|
|
result = cleanup_control_sockets(ssh_dir)
|
||
|
|
assert result == {"total": 0, "removed": 0, "active": 0}
|
||
|
|
|
||
|
|
|
||
|
|
def test_cleanup_removes_stale(ssh_dir: Path) -> None:
|
||
|
|
control = ssh_dir / "control.d"
|
||
|
|
control.mkdir()
|
||
|
|
(control / "abc.sock").write_text("")
|
||
|
|
(control / "def.sock").write_text("")
|
||
|
|
|
||
|
|
with patch("sshctl.operations.cleanup._is_socket_in_use", return_value=False):
|
||
|
|
result = cleanup_control_sockets(ssh_dir)
|
||
|
|
|
||
|
|
assert result["total"] == 2
|
||
|
|
assert result["removed"] == 2
|
||
|
|
assert result["active"] == 0
|
||
|
|
assert not (control / "abc.sock").exists()
|
||
|
|
assert not (control / "def.sock").exists()
|
||
|
|
|
||
|
|
|
||
|
|
def test_cleanup_dry_run_keeps_sockets(ssh_dir: Path) -> None:
|
||
|
|
control = ssh_dir / "control.d"
|
||
|
|
control.mkdir()
|
||
|
|
sock = control / "abc.sock"
|
||
|
|
sock.write_text("")
|
||
|
|
|
||
|
|
with patch("sshctl.operations.cleanup._is_socket_in_use", return_value=False):
|
||
|
|
result = cleanup_control_sockets(ssh_dir, dry_run=True)
|
||
|
|
|
||
|
|
assert result["total"] == 1
|
||
|
|
assert result["removed"] == 1
|
||
|
|
assert sock.exists()
|
||
|
|
|
||
|
|
|
||
|
|
def test_cleanup_active_not_removed(ssh_dir: Path) -> None:
|
||
|
|
control = ssh_dir / "control.d"
|
||
|
|
control.mkdir()
|
||
|
|
(control / "abc.sock").write_text("")
|
||
|
|
(control / "def.sock").write_text("")
|
||
|
|
|
||
|
|
def fake_in_use(sock: Path) -> bool:
|
||
|
|
return sock.name == "abc.sock"
|
||
|
|
|
||
|
|
with patch("sshctl.operations.cleanup._is_socket_in_use", side_effect=fake_in_use):
|
||
|
|
result = cleanup_control_sockets(ssh_dir)
|
||
|
|
|
||
|
|
assert result["total"] == 2
|
||
|
|
assert result["active"] == 1
|
||
|
|
assert result["removed"] == 1
|
||
|
|
assert (control / "abc.sock").exists()
|
||
|
|
assert not (control / "def.sock").exists()
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_socket_in_use_with_ssh() -> None:
|
||
|
|
with patch(
|
||
|
|
"sshctl.operations.cleanup.subprocess.run",
|
||
|
|
) as mock_run:
|
||
|
|
mock_run.return_value.stdout = " 123 ssh user cwd DIR /tmp/socket"
|
||
|
|
assert _is_socket_in_use(Path("/tmp/x.sock")) is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_socket_in_use_no_ssh() -> None:
|
||
|
|
with patch(
|
||
|
|
"sshctl.operations.cleanup.subprocess.run",
|
||
|
|
) as mock_run:
|
||
|
|
mock_run.return_value.stdout = " 123 vim user cwd DIR /tmp"
|
||
|
|
assert _is_socket_in_use(Path("/tmp/x.sock")) is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_socket_in_use_error() -> None:
|
||
|
|
with patch(
|
||
|
|
"sshctl.operations.cleanup.subprocess.run",
|
||
|
|
side_effect=FileNotFoundError,
|
||
|
|
):
|
||
|
|
assert _is_socket_in_use(Path("/tmp/x.sock")) is False
|