# SPDX-License-Identifier: MIT """Tests for winpodx.core.info — the 5-section snapshot consumed by CLI + GUI.""" from __future__ import annotations import subprocess from winpodx.core.config import Config from winpodx.core.info import ( _bundled_oem_version, _bundled_rdprrap_version, _read_os_release, _read_text_file, gather_info, ) def _cfg() -> Config: cfg = Config() cfg.pod.backend = "podman" cfg.rdp.ip = "127.0.0.1" cfg.rdp.port = 3390 cfg.pod.vnc_port = 8007 cfg.pod.container_name = "winpodx-windows" return cfg # --- _read_text_file --- def test_read_text_file_returns_none_when_missing(tmp_path): assert _read_text_file(tmp_path / "no.txt") is None def test_read_text_file_returns_none_when_oversized(tmp_path): p = tmp_path / "big.txt" p.write_bytes(b"x" * 10_000) assert _read_text_file(p, max_bytes=4096) is None def test_read_text_file_strips_trailing_whitespace(tmp_path): p = tmp_path / "v.txt" p.write_text("0.1.9\n\n", encoding="utf-8") assert _read_text_file(p) == "0.1.9" # --- OS release parsing --- def test_read_os_release_parses_quoted(tmp_path, monkeypatch): fake = tmp_path / "os-release" fake.write_text( 'ID=opensuse-tumbleweed\nVERSION="20260101"\nNAME="openSUSE Tumbleweed"\n', encoding="utf-8", ) monkeypatch.setattr( "winpodx.core.info._read_text_file", lambda p, max_bytes=8192: fake.read_text() ) osr = _read_os_release() assert osr["ID"] == "opensuse-tumbleweed" assert osr["VERSION"] == "20260101" def test_read_os_release_returns_empty_when_missing(monkeypatch): monkeypatch.setattr("winpodx.core.info._read_text_file", lambda *a, **k: None) assert _read_os_release() == {} # --- bundled version pin readers --- def test_bundled_oem_version_matches_install_bat_marker(monkeypatch): """The bundled OEM version parsed from install.bat must track the ``set WINPODX_OEM_VERSION=`` marker so guest-sync delivers guest-script changes (e.g. the agent keep-alive task) to already-provisioned pods. Force the install.bat fallback by nulling the .txt stamp reader. """ monkeypatch.setattr("winpodx.core.info._read_text_file", lambda *a, **k: None) assert _bundled_oem_version() == "28" def test_bundled_oem_version_falls_back_to_unknown(monkeypatch, tmp_path): """When neither the .txt stamp NOR the install.bat fallback is available, the helper must return ``(unknown)`` rather than raising or returning ''. The implementation reads: 1. ``config/oem/oem_version.txt`` via _read_text_file 2. ``config/oem/install.bat`` via path.open (streaming, not _read_text_file) Both sources need to be redirected to a nonexistent path for the test to actually exercise the fallback path. Patching only _read_text_file leaves install.bat reachable on a dev checkout and the helper returns the real version. """ monkeypatch.setattr("winpodx.core.info._read_text_file", lambda *a, **k: None) monkeypatch.setattr("winpodx.core.info.bundle_dir", lambda: tmp_path) assert _bundled_oem_version() == "(unknown)" def test_bundled_rdprrap_version_keeps_first_line_only(monkeypatch, tmp_path): pin = tmp_path / "rdprrap_version.txt" pin.write_text("0.1.3\nsha256:deadbeef...\n", encoding="utf-8") monkeypatch.setattr( "winpodx.core.info._read_text_file", lambda p, max_bytes=128: pin.read_text() if "rdprrap" in str(p) else None, ) assert _bundled_rdprrap_version() == "0.1.3" # --- gather_info: end-to-end shape --- def _stub_probes(monkeypatch, *, rdp_alive: bool, vnc_alive: bool): """Override the network probes deterministically. Two probes are patched because info.py uses ``check_rdp_port`` for the RDP port (X.224 handshake) and ``check_tcp_port`` for the VNC port (plain TCP accept -- VNC speaks RFB, not RDP, so the handshake probe would always fail; see #214 fix). """ def rdp_probe(ip, port, timeout=5.0): return rdp_alive def vnc_probe(ip, port, timeout=5.0): return vnc_alive monkeypatch.setattr("winpodx.core.info.check_rdp_port", rdp_probe) monkeypatch.setattr("winpodx.core.info.check_tcp_port", vnc_probe) def _stub_pod_status(monkeypatch, state_value: str): from winpodx.core.pod import PodState, PodStatus state = PodState(state_value) if state_value in (s.value for s in PodState) else PodState.ERROR monkeypatch.setattr("winpodx.core.info.pod_status", lambda cfg: PodStatus(state=state)) def _stub_uptime(monkeypatch, value: str = ""): """Stub _container_uptime directly so we don't accidentally monkeypatch subprocess globally (display.scaling also reads from subprocess.run for kreadconfig6 etc.).""" monkeypatch.setattr("winpodx.core.info._container_uptime", lambda cfg: value) def _stub_display(monkeypatch): monkeypatch.setattr( "winpodx.core.info._display_section", lambda: { "session_type": "wayland", "desktop_environment": "kde", "wayland_freerdp": "True", "raw_scale": "1.00", "rdp_scale": "100%", }, ) def _stub_dependencies(monkeypatch): monkeypatch.setattr( "winpodx.core.info._dependencies_section", lambda: {"python3": {"found": "true", "path": "/usr/bin/python3"}}, ) def _stub_active_sessions(monkeypatch, count: int = 0): monkeypatch.setattr("winpodx.core.info._active_session_count", lambda: count) def _stub_all(monkeypatch, *, rdp=True, vnc=True, state="running", uptime="", sessions=0): """Compose all the stubs gather_info needs. Each test customizes via kwargs.""" _stub_probes(monkeypatch, rdp_alive=rdp, vnc_alive=vnc) _stub_pod_status(monkeypatch, state) _stub_uptime(monkeypatch, uptime) _stub_display(monkeypatch) _stub_dependencies(monkeypatch) _stub_active_sessions(monkeypatch, sessions) def test_gather_info_returns_all_five_sections(monkeypatch): cfg = _cfg() _stub_all(monkeypatch, uptime="2026-04-25T00:00:00Z") info = gather_info(cfg) assert set(info.keys()) == {"system", "display", "dependencies", "pod", "config"} def test_gather_info_pod_section_when_running(monkeypatch): cfg = _cfg() _stub_all(monkeypatch, uptime="2026-04-25T00:00:00Z") info = gather_info(cfg) pod = info["pod"] assert pod["state"] == "running" assert pod["rdp_reachable"] is True assert pod["vnc_reachable"] is True assert pod["uptime"] == "2026-04-25T00:00:00Z" def test_gather_info_pod_section_when_rdp_dead_vnc_alive(monkeypatch): """Bug B's exact scenario — info should reflect the asymmetry honestly.""" cfg = _cfg() _stub_all(monkeypatch, rdp=False, vnc=True, state="starting", uptime="2026-04-25T00:00:00Z") info = gather_info(cfg) assert info["pod"]["rdp_reachable"] is False assert info["pod"]["vnc_reachable"] is True def test_gather_info_pod_section_when_pod_down(monkeypatch): cfg = _cfg() _stub_all(monkeypatch, rdp=False, vnc=False, state="stopped", uptime="") info = gather_info(cfg) assert info["pod"]["uptime"] == "" assert info["pod"]["rdp_reachable"] is False assert info["pod"]["vnc_reachable"] is False def test_gather_info_pod_section_libvirt_uptime_empty(monkeypatch): cfg = _cfg() cfg.pod.backend = "libvirt" _stub_all(monkeypatch) # _container_uptime stubbed to "" anyway, libvirt path also returns "" info = gather_info(cfg) assert info["pod"]["uptime"] == "" def test_gather_info_config_section_includes_budget_when_oversubscribed(monkeypatch): cfg = _cfg() cfg.pod.max_sessions = 50 cfg.pod.ram_gb = 4 _stub_all(monkeypatch) info = gather_info(cfg) assert info["config"]["budget_warning"] assert "ram_gb" in info["config"]["budget_warning"] def test_gather_info_config_section_silent_on_default(monkeypatch): cfg = _cfg() # default 10 sessions / 4 GB _stub_all(monkeypatch) info = gather_info(cfg) assert info["config"]["budget_warning"] == "" def test_gather_info_active_sessions_count_propagates(monkeypatch): cfg = _cfg() _stub_all(monkeypatch, sessions=3) info = gather_info(cfg) assert info["pod"]["active_sessions"] == 3 def test_active_session_count_returns_zero_on_helper_failure(monkeypatch): """The real _active_session_count must swallow exceptions and return 0.""" from winpodx.core.info import _active_session_count def boom(): raise RuntimeError("simulated process tracking failure") monkeypatch.setattr("winpodx.core.process.list_active_sessions", boom) assert _active_session_count() == 0 def test_container_uptime_returns_empty_on_subprocess_timeout(monkeypatch): """The real _container_uptime must swallow TimeoutExpired.""" from winpodx.core.info import _container_uptime def boom(cmd, **kwargs): raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout", 0)) monkeypatch.setattr("winpodx.core.info.subprocess.run", boom) cfg = _cfg() assert _container_uptime(cfg) == ""