wehub-resource-sync
3e779be6f3
CI / lint (push) Failing after 13m4s
CI / test (3.11, ubuntu-latest) (push) Failing after 2m4s
CI / test (3.13, ubuntu-latest) (push) Successful in 13m30s
CI / test (3.14, ubuntu-latest) (push) Successful in 17m21s
CI / test (3.12, ubuntu-latest) (push) Successful in 17m55s
CI / discover-apps-ps (push) Successful in 1m56s
CI / test (3.9, ubuntu-latest) (push) Successful in 13m17s
CI / test (3.10, ubuntu-latest) (push) Successful in 26m21s
CI / audit (push) Successful in 13m38s
Deploy site / deploy (push) Has been cancelled
CI / test (3.14, ubuntu-24.04-arm) (push) Has been cancelled
294 lines
10 KiB
Python
294 lines
10 KiB
Python
# SPDX-License-Identifier: MIT
|
|
"""Tests for winpodx.core.pod — focused on Bug B's RDP recovery helper."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from unittest.mock import MagicMock
|
|
|
|
from winpodx.core.config import Config
|
|
from winpodx.core.pod import recover_rdp_if_needed
|
|
|
|
|
|
def _cfg(backend: str = "podman") -> Config:
|
|
cfg = Config()
|
|
cfg.pod.backend = backend
|
|
cfg.rdp.ip = "127.0.0.1"
|
|
cfg.rdp.port = 3390
|
|
cfg.pod.vnc_port = 8007
|
|
cfg.pod.container_name = "winpodx-windows"
|
|
return cfg
|
|
|
|
|
|
# --- recover_rdp_if_needed: short-circuit paths ---
|
|
|
|
|
|
def test_recover_rdp_no_op_when_rdp_already_alive(monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", lambda *a, **k: True)
|
|
fake_run = MagicMock()
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", fake_run)
|
|
assert recover_rdp_if_needed(_cfg()) is True
|
|
fake_run.assert_not_called()
|
|
|
|
|
|
def test_recover_rdp_returns_false_when_vnc_also_dead(monkeypatch):
|
|
# Both probes return False -> RDP dead and the VNC TCP-accept
|
|
# fallback says "container is dead too" -> recovery skips.
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", lambda *a, **k: False)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", lambda *a, **k: False)
|
|
fake_run = MagicMock()
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", fake_run)
|
|
assert recover_rdp_if_needed(_cfg()) is False
|
|
fake_run.assert_not_called()
|
|
|
|
|
|
def test_recover_rdp_skips_manual_backend(monkeypatch):
|
|
fake_run = MagicMock()
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", fake_run)
|
|
# Returns True for manual so callers don't block.
|
|
assert recover_rdp_if_needed(_cfg(backend="manual")) is True
|
|
assert recover_rdp_if_needed(_cfg(backend="manual")) is True
|
|
fake_run.assert_not_called()
|
|
|
|
|
|
def test_recover_rdp_rejects_bad_container_name(monkeypatch):
|
|
cfg = _cfg()
|
|
cfg.pod.container_name = "../escape; rm -rf /"
|
|
|
|
# VNC alive, RDP dead -- would normally fire recovery, but bad name blocks.
|
|
# check_rdp_port is the RDP-handshake probe (False = RDP dead);
|
|
# check_tcp_port handles the VNC liveness check (True = container alive).
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", lambda *a, **k: False)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", lambda *a, **k: True)
|
|
fake_run = MagicMock()
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", fake_run)
|
|
assert recover_rdp_if_needed(cfg) is False
|
|
fake_run.assert_not_called()
|
|
|
|
|
|
# --- recover_rdp_if_needed: full restart path ---
|
|
|
|
|
|
def _make_probe(rdp_results: list[bool], vnc_alive: bool = True):
|
|
"""Probe sequence pair: ``rdp_probe`` returns the next bool from
|
|
rdp_results on each call (used to monkeypatch ``check_rdp_port``,
|
|
the X.224-handshake flavor); ``tcp_probe`` returns ``vnc_alive``
|
|
for the VNC port (used to monkeypatch ``check_tcp_port``, the
|
|
plain TCP-accept flavor that ``recover_rdp_if_needed`` uses
|
|
specifically for the VNC liveness probe). Both are returned so a
|
|
single call site can wire both monkeypatches:
|
|
|
|
rdp_probe, tcp_probe = _make_probe([False, True])
|
|
monkeypatch.setattr(\".../check_rdp_port\", rdp_probe)
|
|
monkeypatch.setattr(\".../check_tcp_port\", tcp_probe)
|
|
"""
|
|
rdp_iter = iter(rdp_results)
|
|
|
|
def rdp_probe(ip, port, timeout=5.0):
|
|
try:
|
|
return next(rdp_iter)
|
|
except StopIteration:
|
|
return False
|
|
|
|
def tcp_probe(ip, port, timeout=5.0):
|
|
if port == 8007: # VNC port from _cfg()
|
|
return vnc_alive
|
|
return False
|
|
|
|
return rdp_probe, tcp_probe
|
|
|
|
|
|
def test_recover_rdp_fires_termservice_when_vnc_alive(monkeypatch):
|
|
cfg = _cfg()
|
|
rdp_probe, tcp_probe = _make_probe([False, True])
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", rdp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", tcp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.time.sleep", lambda _x: None)
|
|
|
|
captured_cmds = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
captured_cmds.append(cmd)
|
|
return subprocess.CompletedProcess(cmd, 0, "", "")
|
|
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", fake_run)
|
|
|
|
assert recover_rdp_if_needed(cfg) is True
|
|
assert len(captured_cmds) == 1
|
|
cmd = captured_cmds[0]
|
|
# v0.1.9.5: recovery now restarts the *container* (the TermService
|
|
# path was on the broken podman exec channel and never worked). The
|
|
# restart subcommand is what the test should verify.
|
|
assert cmd[0] == "podman"
|
|
assert "restart" in cmd
|
|
assert cfg.pod.container_name in cmd
|
|
|
|
|
|
def test_recover_rdp_returns_false_after_max_attempts(monkeypatch):
|
|
cfg = _cfg()
|
|
rdp_probe, tcp_probe = _make_probe([False, False, False, False]) # always dead
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", rdp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", tcp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.time.sleep", lambda _x: None)
|
|
|
|
monkeypatch.setattr(
|
|
"winpodx.core.pod.health.subprocess.run",
|
|
lambda cmd, **kw: subprocess.CompletedProcess(cmd, 0, "", ""),
|
|
)
|
|
|
|
assert recover_rdp_if_needed(cfg, max_attempts=3) is False
|
|
|
|
|
|
def test_recover_rdp_tolerates_subprocess_errors(monkeypatch):
|
|
"""A failed/timed-out exec must not raise; recovery returns False on the fallthrough."""
|
|
cfg = _cfg()
|
|
rdp_probe, tcp_probe = _make_probe([False, False])
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", rdp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", tcp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.time.sleep", lambda _x: None)
|
|
|
|
def boom(cmd, **kw):
|
|
raise subprocess.TimeoutExpired(cmd, kw.get("timeout", 0))
|
|
|
|
monkeypatch.setattr("winpodx.core.pod.health.subprocess.run", boom)
|
|
assert recover_rdp_if_needed(cfg, max_attempts=1) is False
|
|
|
|
|
|
def test_recover_rdp_succeeds_even_when_exec_returns_nonzero(monkeypatch):
|
|
"""sc.exe failure actions can revive TermService independently of the explicit
|
|
Restart-Service call -- accept rc!=0 as long as the post-recovery probe is alive."""
|
|
cfg = _cfg()
|
|
rdp_probe, tcp_probe = _make_probe([False, True])
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_rdp_port", rdp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.check_tcp_port", tcp_probe)
|
|
monkeypatch.setattr("winpodx.core.pod.health.time.sleep", lambda _x: None)
|
|
monkeypatch.setattr(
|
|
"winpodx.core.pod.health.subprocess.run",
|
|
lambda cmd, **kw: subprocess.CompletedProcess(cmd, 1, "", "service stuck"),
|
|
)
|
|
assert recover_rdp_if_needed(cfg) is True
|
|
|
|
|
|
# --- v0.2.0.5: pod wait-ready ---
|
|
|
|
|
|
class TestWaitReady:
|
|
"""Multi-phase wait gate: container -> RDP port -> activation."""
|
|
|
|
def _patch_cfg(self, monkeypatch):
|
|
from winpodx.core.config import Config
|
|
|
|
cfg = Config()
|
|
cfg.pod.backend = "podman"
|
|
cfg.pod.container_name = "winpodx-windows"
|
|
cfg.rdp.ip = "127.0.0.1"
|
|
cfg.rdp.port = 3389
|
|
monkeypatch.setattr("winpodx.core.config.Config.load", classmethod(lambda cls: cfg))
|
|
return cfg
|
|
|
|
def test_rejects_unsupported_backend(self, monkeypatch, capsys):
|
|
import pytest
|
|
|
|
from winpodx.cli.pod import _wait_ready
|
|
from winpodx.core.config import Config
|
|
|
|
cfg = Config()
|
|
cfg.pod.backend = "manual"
|
|
monkeypatch.setattr("winpodx.core.config.Config.load", classmethod(lambda cls: cfg))
|
|
|
|
with pytest.raises(SystemExit) as excinfo:
|
|
_wait_ready(timeout=10, show_logs=False)
|
|
assert excinfo.value.code == 2
|
|
assert "wait-ready not supported" in capsys.readouterr().out
|
|
|
|
def test_happy_path_traverses_three_phases(self, monkeypatch, capsys):
|
|
from unittest.mock import MagicMock
|
|
|
|
from winpodx.cli.pod import _wait_ready
|
|
from winpodx.core.pod import PodState
|
|
|
|
self._patch_cfg(monkeypatch)
|
|
# Phase 1: container running immediately.
|
|
monkeypatch.setattr(
|
|
"winpodx.core.pod.pod_status",
|
|
lambda cfg: MagicMock(state=PodState.RUNNING),
|
|
)
|
|
# Phase 2: RDP port immediately reachable.
|
|
monkeypatch.setattr(
|
|
"winpodx.core.pod.check_rdp_port",
|
|
lambda ip, port, timeout=1.0: True,
|
|
)
|
|
# Phase 3: activation probe succeeds.
|
|
monkeypatch.setattr(
|
|
"winpodx.core.provisioner.wait_for_windows_responsive",
|
|
lambda cfg, timeout=180: True,
|
|
)
|
|
# Phase 4 (OEM reboot pass): pretend the marker was never
|
|
# written so the helper short-circuits to True via its
|
|
# appear-grace fallback without sleeping.
|
|
monkeypatch.setattr(
|
|
"winpodx.cli.pod._wait_for_oem_reboot",
|
|
lambda cfg, timeout: True,
|
|
)
|
|
|
|
_wait_ready(timeout=30, show_logs=False)
|
|
out = capsys.readouterr().out
|
|
assert "[1/4]" in out
|
|
assert "[2/4]" in out
|
|
assert "[3/4]" in out
|
|
assert "[4/4]" in out
|
|
assert "Windows ready" in out
|
|
assert "OEM reboot pass complete" in out
|
|
|
|
|
|
# -- pod recreate --keep-iso (storage wipe that preserves the cached ISO) ----
|
|
|
|
|
|
def _populate_storage(d):
|
|
(d / "data.img").write_text("disk")
|
|
(d / "win11x64.iso").write_text("iso")
|
|
(d / "windows.ver").write_text("11")
|
|
(d / "drivers").mkdir()
|
|
(d / "drivers" / "x.inf").write_text("drv")
|
|
|
|
|
|
def test_wipe_pod_storage_keep_iso_preserves_iso(tmp_path):
|
|
from winpodx.cli.pod import _wipe_pod_storage
|
|
|
|
_populate_storage(tmp_path)
|
|
cfg = Config()
|
|
cfg.pod.backend = "podman"
|
|
cfg.pod.storage_path = str(tmp_path)
|
|
|
|
_wipe_pod_storage(cfg, keep_iso=True)
|
|
|
|
names = {p.name for p in tmp_path.iterdir()}
|
|
assert names == {"win11x64.iso"} # only the ISO survives
|
|
|
|
|
|
def test_wipe_pod_storage_full_removes_everything(tmp_path):
|
|
from winpodx.cli.pod import _wipe_pod_storage
|
|
|
|
_populate_storage(tmp_path)
|
|
cfg = Config()
|
|
cfg.pod.backend = "podman"
|
|
cfg.pod.storage_path = str(tmp_path)
|
|
|
|
_wipe_pod_storage(cfg, keep_iso=False)
|
|
|
|
assert list(tmp_path.iterdir()) == [] # dir kept, contents gone
|
|
|
|
|
|
def test_wipe_pod_storage_keep_iso_warns_when_no_iso(tmp_path, capsys):
|
|
from winpodx.cli.pod import _wipe_pod_storage
|
|
|
|
(tmp_path / "data.img").write_text("disk") # no ISO present
|
|
cfg = Config()
|
|
cfg.pod.backend = "podman"
|
|
cfg.pod.storage_path = str(tmp_path)
|
|
|
|
_wipe_pod_storage(cfg, keep_iso=True)
|
|
|
|
assert list(tmp_path.iterdir()) == []
|
|
assert "no cached ISO" in capsys.readouterr().out
|