Simone
3eb29b404f
- Introduced a new test to verify behavior when SSH transport connects but is not ready, ensuring it returns False after the deadline. - Added comprehensive tests for the `deploy-linux` template, covering various scenarios including missing public keys, VM directory existence, SHA256 mismatches, and failures during ISO creation and VM conversion. - Implemented tests for SSH-related functions, including connection retries and streaming command execution, ensuring proper handling of exit statuses and progress indications. - Enhanced existing tests to cover edge cases and ensure robustness in the deployment process.
481 lines
16 KiB
Python
481 lines
16 KiB
Python
"""Tests for SshTransport (paramiko mocked).
|
|
|
|
Covers AGENTS.md error #12 indirectly: with paramiko there is no native
|
|
``ssh-keygen`` invocation, so the PowerShell stderr-handling pitfall does
|
|
not apply. The default permissive host-key policy mirrors the
|
|
``StrictHostKeyChecking=no`` / ``UserKnownHostsFile=NUL`` defaults of
|
|
``scripts/_Transport.psm1``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import types
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError
|
|
from ci_orchestrator.transport.ssh import SshTransport
|
|
|
|
|
|
class _FakeChannel:
|
|
def __init__(self, rc: int) -> None:
|
|
self._rc = rc
|
|
|
|
def recv_exit_status(self) -> int:
|
|
return self._rc
|
|
|
|
|
|
class _FakeStream:
|
|
def __init__(self, data: bytes, rc: int = 0) -> None:
|
|
self._data = data
|
|
self.channel = _FakeChannel(rc)
|
|
|
|
def read(self) -> bytes:
|
|
return self._data
|
|
|
|
|
|
class _FakeSFTP:
|
|
def __init__(self) -> None:
|
|
self.puts: list[tuple[str, str]] = []
|
|
self.gets: list[tuple[str, str]] = []
|
|
|
|
def __enter__(self) -> _FakeSFTP:
|
|
return self
|
|
|
|
def __exit__(self, *_e: object) -> None:
|
|
pass
|
|
|
|
def put(self, local: str, remote: str) -> None:
|
|
self.puts.append((local, remote))
|
|
|
|
def get(self, remote: str, local: str) -> None:
|
|
self.gets.append((remote, local))
|
|
|
|
|
|
class _FakeSSHClient:
|
|
def __init__(self) -> None:
|
|
self.connect_kwargs: dict[str, Any] = {}
|
|
self.policy: Any = None
|
|
self.next_stdout = b""
|
|
self.next_stderr = b""
|
|
self.next_rc = 0
|
|
self.commands: list[str] = []
|
|
self.sftp = _FakeSFTP()
|
|
self.closed = False
|
|
|
|
def set_missing_host_key_policy(self, policy: Any) -> None:
|
|
self.policy = policy
|
|
|
|
def load_host_keys(self, _path: str) -> None:
|
|
pass
|
|
|
|
def connect(self, **kwargs: Any) -> None:
|
|
self.connect_kwargs = kwargs
|
|
|
|
def exec_command(self, command: str, timeout: float | None = None) -> Any:
|
|
self.commands.append(command)
|
|
return (
|
|
None,
|
|
_FakeStream(self.next_stdout, self.next_rc),
|
|
_FakeStream(self.next_stderr, self.next_rc),
|
|
)
|
|
|
|
def open_sftp(self) -> _FakeSFTP:
|
|
return self.sftp
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def _install_fake_paramiko(monkeypatch: pytest.MonkeyPatch, client: _FakeSSHClient) -> Any:
|
|
fake = types.ModuleType("paramiko")
|
|
fake.SSHClient = lambda: client # type: ignore[attr-defined]
|
|
fake.AutoAddPolicy = lambda: "auto-add" # type: ignore[attr-defined]
|
|
monkeypatch.setitem(sys.modules, "paramiko", fake)
|
|
return fake
|
|
|
|
|
|
def test_run_returns_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
client.next_stdout = b"hello\n"
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", username="ci_build", key_path="/tmp/key")
|
|
result = t.run("echo hello")
|
|
assert result.ok
|
|
assert result.stdout == "hello\n"
|
|
assert client.commands == ["echo hello"]
|
|
|
|
|
|
def test_run_raises_on_nonzero(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
client.next_rc = 2
|
|
client.next_stderr = b"oops"
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportCommandError):
|
|
t.run("bad")
|
|
|
|
|
|
def test_close_is_idempotent_and_swallows(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _ExplodingClient(_FakeSSHClient):
|
|
def close(self) -> None: # type: ignore[override]
|
|
raise RuntimeError("already closed")
|
|
|
|
client = _ExplodingClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
t._connect() # type: ignore[attr-defined]
|
|
t.close()
|
|
t.close() # idempotent on already-closed (no client)
|
|
|
|
|
|
def test_context_manager_connects_and_closes(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
with SshTransport("10.0.0.2", key_path="/tmp/key") as t:
|
|
assert isinstance(t, SshTransport)
|
|
assert client.closed
|
|
|
|
|
|
def test_known_hosts_file_is_loaded(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Any
|
|
) -> None:
|
|
kh = tmp_path / "known_hosts"
|
|
kh.write_text("")
|
|
loaded: list[str] = []
|
|
|
|
class _RecordingClient(_FakeSSHClient):
|
|
def load_host_keys(self, p: str) -> None: # type: ignore[override]
|
|
loaded.append(p)
|
|
|
|
client = _RecordingClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key", known_hosts=str(kh))
|
|
t.run("ls")
|
|
assert loaded == [str(kh)]
|
|
|
|
|
|
def test_is_ready_returns_true_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
assert t.is_ready() is True
|
|
|
|
|
|
def test_is_ready_returns_false_on_connect_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
class _BoomClient(_FakeSSHClient):
|
|
def connect(self, **_kw: Any) -> None: # type: ignore[override]
|
|
raise RuntimeError("net down")
|
|
|
|
client = _BoomClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
assert t.is_ready() is False
|
|
|
|
|
|
def test_copy_wraps_sftp_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
|
|
class _BadSFTP(_FakeSFTP):
|
|
def put(self, _l: str, _r: str) -> None: # type: ignore[override]
|
|
raise RuntimeError("sftp boom")
|
|
|
|
client = _FakeSSHClient()
|
|
client.sftp = _BadSFTP()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportConnectError):
|
|
t.copy("/x", "/y")
|
|
|
|
|
|
def test_fetch_wraps_sftp_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _BadSFTP(_FakeSFTP):
|
|
def get(self, _r: str, _l: str) -> None: # type: ignore[override]
|
|
raise RuntimeError("sftp boom")
|
|
|
|
client = _FakeSSHClient()
|
|
client.sftp = _BadSFTP()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportConnectError):
|
|
t.fetch("/y", "/x")
|
|
|
|
|
|
def test_default_policy_is_auto_add(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Mirrors permissive `StrictHostKeyChecking=no` default for ephemeral CI VMs."""
|
|
client = _FakeSSHClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
SshTransport("10.0.0.2", key_path="/tmp/key").run("true")
|
|
assert client.policy == "auto-add"
|
|
|
|
|
|
def test_copy_and_fetch(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
t.copy("local.bin", "/remote/local.bin")
|
|
t.fetch("/remote/out.bin", "out.bin")
|
|
assert client.sftp.puts == [("local.bin", "/remote/local.bin")]
|
|
assert client.sftp.gets == [("/remote/out.bin", "out.bin")]
|
|
|
|
|
|
def test_is_ready_returns_false_on_connect_failure(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
fake = types.ModuleType("paramiko")
|
|
|
|
class _Boom:
|
|
def set_missing_host_key_policy(self, _p: object) -> None:
|
|
pass
|
|
|
|
def connect(self, **_kw: Any) -> None:
|
|
raise OSError("refused")
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
fake.SSHClient = lambda: _Boom() # type: ignore[attr-defined]
|
|
fake.AutoAddPolicy = lambda: object() # type: ignore[attr-defined]
|
|
monkeypatch.setitem(sys.modules, "paramiko", fake)
|
|
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
assert t.is_ready() is False
|
|
with pytest.raises(TransportConnectError):
|
|
t.run("x")
|
|
|
|
|
|
# ── run_streaming helpers ─────────────────────────────────────────────────────
|
|
|
|
class _FakeStreamingChannel:
|
|
"""Simulates a paramiko Channel for run_streaming tests."""
|
|
|
|
def __init__(
|
|
self,
|
|
stdout_chunks: list[bytes],
|
|
stderr_chunks: list[bytes],
|
|
rc: int = 0,
|
|
) -> None:
|
|
self._stdout = list(stdout_chunks)
|
|
self._stderr = list(stderr_chunks)
|
|
self._rc = rc
|
|
self.closed = False
|
|
self._timeout_set: float | None = None
|
|
|
|
def settimeout(self, t: float) -> None:
|
|
self._timeout_set = t
|
|
|
|
def exec_command(self, _cmd: str) -> None:
|
|
pass
|
|
|
|
def recv_ready(self) -> bool:
|
|
return bool(self._stdout)
|
|
|
|
def recv(self, _n: int) -> bytes:
|
|
return self._stdout.pop(0) if self._stdout else b""
|
|
|
|
def recv_stderr_ready(self) -> bool:
|
|
return bool(self._stderr)
|
|
|
|
def recv_stderr(self, _n: int) -> bytes:
|
|
return self._stderr.pop(0) if self._stderr else b""
|
|
|
|
def exit_status_ready(self) -> bool:
|
|
return not self._stdout and not self._stderr
|
|
|
|
def recv_exit_status(self) -> int:
|
|
return self._rc
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class _FakeTransport:
|
|
def __init__(self, chan: _FakeStreamingChannel) -> None:
|
|
self._chan = chan
|
|
|
|
def open_session(self) -> _FakeStreamingChannel:
|
|
return self._chan
|
|
|
|
|
|
class _FakeSSHClientWithTransport(_FakeSSHClient):
|
|
def __init__(self, chan: _FakeStreamingChannel) -> None:
|
|
super().__init__()
|
|
self._chan = chan
|
|
|
|
def get_transport(self) -> _FakeTransport:
|
|
return _FakeTransport(self._chan)
|
|
|
|
|
|
# ── run_streaming tests ───────────────────────────────────────────────────────
|
|
|
|
def test_run_streaming_returns_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
chan = _FakeStreamingChannel([b"hello\n", b"world\n"], [])
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
result = t.run_streaming("echo hello")
|
|
assert result.ok
|
|
assert result.stdout == "hello\nworld\n"
|
|
assert result.stderr == ""
|
|
assert chan.closed
|
|
|
|
|
|
def test_run_streaming_captures_stderr(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
chan = _FakeStreamingChannel([], [b"err line\n"])
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
result = t.run_streaming("cmd", check=False)
|
|
assert result.stderr == "err line\n"
|
|
|
|
|
|
def test_run_streaming_raises_on_nonzero(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
chan = _FakeStreamingChannel([b"out"], [], rc=1)
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportCommandError):
|
|
t.run_streaming("bad")
|
|
|
|
|
|
def test_run_streaming_check_false_no_raise(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
chan = _FakeStreamingChannel([], [], rc=2)
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
result = t.run_streaming("bad", check=False)
|
|
assert result.returncode == 2
|
|
|
|
|
|
def test_run_streaming_sets_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
chan = _FakeStreamingChannel([], [])
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
t.run_streaming("cmd", timeout=30.0)
|
|
assert chan._timeout_set == 30.0
|
|
|
|
|
|
def test_run_streaming_exec_command_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _BoomChannel(_FakeStreamingChannel):
|
|
def exec_command(self, _cmd: str) -> None:
|
|
raise OSError("channel error")
|
|
|
|
chan = _BoomChannel([], [])
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportConnectError, match="SSH exec_command failed"):
|
|
t.run_streaming("cmd")
|
|
|
|
|
|
def test_run_streaming_recv_raises_wraps_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _BoomRecvChannel(_FakeStreamingChannel):
|
|
def recv_ready(self) -> bool:
|
|
return True
|
|
|
|
def recv(self, _n: int) -> bytes:
|
|
raise OSError("recv broke")
|
|
|
|
chan = _BoomRecvChannel([], [])
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportConnectError, match="SSH stream failed"):
|
|
t.run_streaming("cmd")
|
|
|
|
|
|
# ── run() exec_command exception ─────────────────────────────────────────────
|
|
|
|
def test_run_exec_command_exception_wraps(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _BoomExecClient(_FakeSSHClient):
|
|
def exec_command(self, _cmd: str, timeout: float | None = None) -> Any:
|
|
raise OSError("exec broke")
|
|
|
|
client = _BoomExecClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
with pytest.raises(TransportConnectError, match="SSH exec_command failed"):
|
|
t.run("cmd")
|
|
|
|
|
|
def test_run_streaming_idle_iteration_sleeps(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Channel reports no data for one iteration before producing output.
|
|
|
|
This exercises the ``if not progressed: time.sleep(0.05)`` branch.
|
|
"""
|
|
slept: list[float] = []
|
|
monkeypatch.setattr("ci_orchestrator.transport.ssh.time.sleep", slept.append)
|
|
|
|
class _LazyChannel(_FakeStreamingChannel):
|
|
def __init__(self) -> None:
|
|
super().__init__([b"data"], [], rc=0)
|
|
self._tick = 0
|
|
|
|
def recv_ready(self) -> bool:
|
|
# No data on first tick; data available from second tick onward.
|
|
return self._tick > 0 and bool(self._stdout)
|
|
|
|
def exit_status_ready(self) -> bool:
|
|
self._tick += 1
|
|
return self._tick > 1 and not self._stdout and not self._stderr
|
|
|
|
chan = _LazyChannel()
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
result = t.run_streaming("cmd")
|
|
assert result.stdout == "data"
|
|
assert slept # sleep was called at least once for the idle iteration
|
|
|
|
|
|
# ── _connect reuse ────────────────────────────────────────────────────────────
|
|
|
|
def test_connect_reuses_existing_client(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = _FakeSSHClient()
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
t.run("first")
|
|
t.run("second")
|
|
# connect() called only once despite two run() calls.
|
|
assert client.commands == ["first", "second"]
|
|
# Only one connect attempt.
|
|
assert client.connect_kwargs["hostname"] == "10.0.0.2"
|
|
|
|
|
|
# ── run_streaming: progressed=True but exit not yet ready → loop back ─────────
|
|
|
|
|
|
def test_run_streaming_progressed_no_sleep_loops_back(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Data received (progressed=True) but exit_status_ready()=False on first check.
|
|
|
|
Covers the ``if not progressed: time.sleep(0.05)`` branch when progressed
|
|
is True — no sleep occurs, loop continues (branch 170->150).
|
|
"""
|
|
slept: list[float] = []
|
|
monkeypatch.setattr("ci_orchestrator.transport.ssh.time.sleep", slept.append)
|
|
|
|
class _ProgressedNoBreakChannel(_FakeStreamingChannel):
|
|
def __init__(self) -> None:
|
|
super().__init__([b"hello"], [], rc=0)
|
|
self._ready_count = 0
|
|
|
|
def exit_status_ready(self) -> bool:
|
|
# First call: stdout already drained but signal exit not yet ready.
|
|
# Subsequent calls: ready → break.
|
|
self._ready_count += 1
|
|
return self._ready_count > 1
|
|
|
|
chan = _ProgressedNoBreakChannel()
|
|
client = _FakeSSHClientWithTransport(chan)
|
|
_install_fake_paramiko(monkeypatch, client)
|
|
t = SshTransport("10.0.0.2", key_path="/tmp/key")
|
|
result = t.run_streaming("cmd")
|
|
assert result.stdout == "hello"
|
|
# sleep must NOT have been called (progressed=True skips sleep branch)
|
|
assert not slept
|