test: update IP pool test to use Windows VMX and clarify test purpose

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.
+    # Build was invoked with the pool IP via the Windows branch.
-    assert calls["linux_build"][0]["ip_address"] == "192.168.79.201"
+    assert calls["windows_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
 
 
 def test_destroy_clone_releases_pool_ip(tmp_path: Path) -> None:

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"