Public Access
emilandDevin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2b032da0cd
Extract common, reusable utilities from the infra and grm repositories into the devx package so they can be shared across all oblachno-oss projects. New devx.utils modules (from infra): - ssh: SSH exec, docker exec on VM, wait_for_ssh (pure-Python socket) - crypto: shell-safe secret/password generation - vault: Ansible vault encrypt/decrypt helpers - network: HTTP connectivity check with tenacity retry - confirm: typed confirmation validation for destructive ops New devx.utils modules (from grm): - json_registry: file-locked JSON registry for local state - step_tracker: multi-step operation tracking with reports - logging: XDG-compliant logging configuration New devx.ci modules (from infra): - validate_deploy_ref: validate git tag for deployments - record_deployed_tag: record deployed tag to Gitea repo variable All modules have 100% test coverage and pass ruff, pyright, bandit, and actionlint checks. Closes DEVX-124 Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
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
|