From 1a1560ec3d98106c8419ecf64787e81aeef29198 Mon Sep 17 00:00:00 2001 From: Simone Date: Tue, 26 May 2026 18:51:01 +0200 Subject: [PATCH] test: update IP pool test to use Windows VMX and clarify test purpose --- tests/python/test_commands_job.py | 24 ++-- tests/python/test_ssh_transport.py | 201 +++++++++++++++++++++++++++++ 2 files changed, 216 insertions(+), 9 deletions(-) diff --git a/tests/python/test_commands_job.py b/tests/python/test_commands_job.py index 84cdf94..b606ce0 100644 --- a/tests/python/test_commands_job.py +++ b/tests/python/test_commands_job.py @@ -580,13 +580,22 @@ def test_inject_guestinfo_ip_omits_empty_gateway(tmp_path: Path) -> None: def test_job_with_ip_pool_uses_preassigned_ip( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path + monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - """When ip_pool is configured the job skips _wait_for_ip and uses the pool IP.""" + """When ip_pool is configured the job skips _wait_for_ip and uses the pool IP. + + IP pool is Windows-only (guestinfo injection has no effect on Linux VMware + Tools), so this test uses a Windows VMX. + """ from ci_orchestrator.config import Config, IpPoolConfig from ci_orchestrator.ip_pool import IpSlotPool - backend = _FakeBackend(running_after=1) + # Windows VMX — pool guard requires guest_os != "linux". + win_vmx = tmp_path / "templates" / "WinBuild2025.vmx" + win_vmx.parent.mkdir(parents=True) + win_vmx.write_text('guestOS = "windows10srv-64"\n', encoding="utf-8") + + backend = _FakeBackend(running_after=1, clone_guest_os="windows10srv-64") calls = _patch_common(monkeypatch, backend) pool_addresses = ["192.168.79.201"] @@ -614,7 +623,7 @@ def test_job_with_ip_pool_uses_preassigned_ip( base_artifact = tmp_path / "artifacts" result = CliRunner().invoke( - cli, _common_args(template_vmx, clone_base=base_clone, artifact_base=base_artifact) + cli, _common_args(win_vmx, clone_base=base_clone, artifact_base=base_artifact) ) assert result.exit_code == 0, result.output @@ -629,11 +638,8 @@ def test_job_with_ip_pool_uses_preassigned_ip( released = IpSlotPool(pool_addresses, "255.255.255.0", "", lease_file)._read_leases() assert released["192.168.79.201"] is None - # Build was invoked with the pool IP. - assert calls["linux_build"][0]["ip_address"] == "192.168.79.201" - - # VMX injection confirmed via log output (clone dir removed after job). - assert "pre-assigned" in result.output + # Build was invoked with the pool IP via the Windows branch. + assert calls["windows_build"][0]["ip_address"] == "192.168.79.201" def test_destroy_clone_releases_pool_ip(tmp_path: Path) -> None: diff --git a/tests/python/test_ssh_transport.py b/tests/python/test_ssh_transport.py index 1382efc..e4dc373 100644 --- a/tests/python/test_ssh_transport.py +++ b/tests/python/test_ssh_transport.py @@ -242,3 +242,204 @@ def test_is_ready_returns_false_on_connect_failure(monkeypatch: pytest.MonkeyPat 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"