Public Access
emil
ef3b882e5b
Post-merge / detect-type (push) Successful in 13s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 28s
Post-merge / vikunja (push) Successful in 44s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / release (push) Successful in 1m7s
Post-merge / publish (push) Successful in 44s
Post-merge / badges (push) Successful in 1m6s
97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
"""Unit tests for devx.utils.ssh."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from devx.utils.ssh import docker_exec_on_vm, ssh_exec, wait_for_ssh
|
|
|
|
|
|
class TestSshExec:
|
|
@patch("devx.utils.ssh.subprocess.run")
|
|
def test_success(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
|
|
result = ssh_exec("10.0.0.1", "uname -a")
|
|
assert result.returncode == 0
|
|
mock_run.assert_called_once()
|
|
|
|
@patch("devx.utils.ssh.subprocess.run")
|
|
def test_failure_with_check(self, mock_run: MagicMock) -> None:
|
|
mock_result = MagicMock(returncode=1, stdout="", stderr="error")
|
|
mock_result.check_returncode.side_effect = subprocess.CalledProcessError(1, "ssh")
|
|
mock_run.return_value = mock_result
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
ssh_exec("10.0.0.1", "false")
|
|
|
|
@patch("devx.utils.ssh.subprocess.run")
|
|
def test_failure_without_check(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
|
|
result = ssh_exec("10.0.0.1", "false", check=False)
|
|
assert result.returncode == 1
|
|
|
|
@patch("devx.utils.ssh.subprocess.run")
|
|
def test_custom_user(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
|
ssh_exec("10.0.0.1", "whoami", user="root")
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "root@10.0.0.1" in cmd
|
|
|
|
|
|
class TestDockerExecOnVm:
|
|
@patch("devx.utils.ssh.ssh_exec")
|
|
def test_simple_command(self, mock_ssh: MagicMock) -> None:
|
|
mock_ssh.return_value = MagicMock(stdout="output\n")
|
|
result = docker_exec_on_vm("10.0.0.1", "mycontainer", "ls /")
|
|
assert result == "output"
|
|
mock_ssh.assert_called_once_with("10.0.0.1", "docker exec mycontainer ls /", user="deploy", timeout=30)
|
|
|
|
@patch("devx.utils.ssh.ssh_exec")
|
|
def test_psql_mode(self, mock_ssh: MagicMock) -> None:
|
|
mock_ssh.return_value = MagicMock(stdout="result\n")
|
|
result = docker_exec_on_vm("10.0.0.1", "db", "SELECT 1", db_user="postgres", db_name="mydb")
|
|
assert result == "result"
|
|
call_args = mock_ssh.call_args[0][1]
|
|
assert "psql -U postgres -d mydb" in call_args
|
|
assert "SELECT 1" in call_args
|
|
|
|
@patch("devx.utils.ssh.ssh_exec")
|
|
def test_psql_escapes_single_quotes(self, mock_ssh: MagicMock) -> None:
|
|
mock_ssh.return_value = MagicMock(stdout="\n")
|
|
docker_exec_on_vm("10.0.0.1", "db", "SELECT 'it''s ok'", db_user="pg", db_name="db")
|
|
call_args = mock_ssh.call_args[0][1]
|
|
assert "'\"'\"'" in call_args
|
|
|
|
|
|
class TestWaitForSsh:
|
|
@patch("devx.utils.ssh.socket.create_connection")
|
|
def test_immediate_success(self, mock_conn: MagicMock) -> None:
|
|
mock_conn.return_value.__enter__ = MagicMock()
|
|
mock_conn.return_value.__exit__ = MagicMock(return_value=False)
|
|
wait_for_ssh("10.0.0.1")
|
|
mock_conn.assert_called_once()
|
|
|
|
@patch("devx.utils.ssh.socket.create_connection")
|
|
@patch("devx.utils.ssh.time.sleep")
|
|
def test_retries_until_success(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None:
|
|
# Fail twice, then succeed
|
|
mock_conn.side_effect = [
|
|
OSError("refused"),
|
|
OSError("refused"),
|
|
MagicMock(),
|
|
]
|
|
mock_conn.return_value.__enter__ = MagicMock()
|
|
mock_conn.return_value.__exit__ = MagicMock(return_value=False)
|
|
wait_for_ssh("10.0.0.1", max_attempts=5)
|
|
assert mock_conn.call_count == 3
|
|
|
|
@patch("devx.utils.ssh.socket.create_connection")
|
|
@patch("devx.utils.ssh.time.sleep")
|
|
def test_timeout_after_max_attempts(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None:
|
|
mock_conn.side_effect = OSError("refused")
|
|
with pytest.raises(RuntimeError, match="SSH not available"):
|
|
wait_for_ssh("10.0.0.1", max_attempts=3)
|
|
assert mock_conn.call_count == 3
|