kernalix7--winpodx/tests/test_provisioner.py

# SPDX-License-Identifier: MIT
"""Tests for auto-provisioning engine."""

from __future__ import annotations

import pytest

from winpodx.core.provisioner import ProvisionError


def test_provision_error():
    err = ProvisionError("test error")
    assert str(err) == "test error"
    assert isinstance(err, Exception)


def test_ensure_config_creates_default(tmp_path, monkeypatch):
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))

    from winpodx.core.provisioner import _ensure_config

    cfg = _ensure_config()

    assert cfg.rdp.user == "WPX-User"
    assert cfg.rdp.ip == "127.0.0.1"
    assert cfg.rdp.password
    assert (tmp_path / "winpodx" / "winpodx.toml").exists()

    from winpodx.core.config import Config

    loaded = Config.load()
    assert loaded.rdp.password == cfg.rdp.password


# Rotation tests moved to tests/test_rotation/test_rotation.py (Sprint 1 Step 2).

# --- v0.1.9.4: runtime applies via FreeRDP RemoteApp (windows_exec.run_in_windows) ---
#
# The v0.1.9.0-v0.1.9.3 versions of these tests mocked podman-exec subprocess
# calls — but podman exec can't reach the Windows VM inside the dockur Linux
# container, so the helpers never actually applied anything (they just logged
# warnings). v0.1.9.4 routes them through windows_exec.run_in_windows, which
# launches PowerShell as a FreeRDP RemoteApp. These tests mock that helper.


def _mock_run_in_windows(monkeypatch, *, rc: int = 0, stdout: str = "", stderr: str = ""):
    """Return a list that captures every (description, payload) call."""
    from winpodx.core.windows_exec import WindowsExecResult

    captured: list[tuple[str, str]] = []

    def fake(cfg, payload, *, timeout=60, description="windows-exec"):
        captured.append((description, payload))
        return WindowsExecResult(rc=rc, stdout=stdout, stderr=stderr)

    import winpodx.core.provisioner as prov

    monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", fake)
    # Make sure the lazy-imported reference inside provisioner picks up the patched fn.
    monkeypatch.setattr(prov, "_apply_max_sessions", prov._apply_max_sessions)
    return captured


def test_apply_max_sessions_skips_manual_backend(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    captured = _mock_run_in_windows(monkeypatch)
    provisioner._apply_max_sessions(cfg)
    assert captured == []


def test_apply_max_sessions_runs_via_windows_exec(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.max_sessions = 25
    captured = _mock_run_in_windows(monkeypatch, rc=0, stdout="max_sessions: 10 -> 25")
    provisioner._apply_max_sessions(cfg)
    assert len(captured) == 1
    description, payload = captured[0]
    assert description == "apply-max-sessions"
    assert "MaxInstanceCount" in payload
    assert "$desired = 25" in payload
    assert "fSingleSessionPerUser" in payload
    # v0.1.9.5: Restart-Service intentionally removed — restarting the
    # TermService that hosts our own RDP session kills the session
    # before the wrapper can write its result file.
    assert "Restart-Service" not in payload


def test_apply_max_sessions_raises_on_nonzero_rc(monkeypatch):
    """v0.1.9.4: helpers no longer silently swallow non-zero rc."""

    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    _mock_run_in_windows(monkeypatch, rc=2, stderr="permission denied")
    with pytest.raises(RuntimeError, match="rc=2"):
        provisioner._apply_max_sessions(cfg)


def test_apply_max_sessions_propagates_channel_error(monkeypatch):

    from winpodx.core import provisioner
    from winpodx.core.config import Config
    from winpodx.core.windows_exec import WindowsExecError

    cfg = Config()

    def fake(*a, **k):
        raise WindowsExecError("FreeRDP not found")

    monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", fake)
    with pytest.raises(WindowsExecError, match="FreeRDP not found"):
        provisioner._apply_max_sessions(cfg)


def test_apply_rdp_timeouts_skips_manual(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    captured = _mock_run_in_windows(monkeypatch)
    provisioner._apply_rdp_timeouts(cfg)
    assert captured == []


def test_apply_rdp_timeouts_payload_contains_all_keys(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    captured = _mock_run_in_windows(monkeypatch, rc=0, stdout="rdp_timeouts applied")
    provisioner._apply_rdp_timeouts(cfg)
    assert len(captured) == 1
    description, payload = captured[0]
    assert description == "apply-rdp-timeouts"
    for token in (
        "MaxIdleTime",
        "MaxDisconnectionTime",
        "MaxConnectionTime",
        "KeepAliveEnable",
        "KeepAliveInterval",
        "KeepAliveTimeout",
        "RDP-Tcp",
        "Terminal Services",
    ):
        assert token in payload, f"missing {token!r} in payload"


def test_apply_agent_keepalive_skips_manual(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    captured = _mock_run_in_windows(monkeypatch)
    provisioner._apply_agent_keepalive(cfg)
    assert captured == []


def test_apply_agent_keepalive_payload_registers_task(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    captured = _mock_run_in_windows(
        monkeypatch, rc=0, stdout="agent_keepalive: WinpodxAgentKeepAlive registered for X"
    )
    provisioner._apply_agent_keepalive(cfg)
    assert len(captured) == 1
    description, payload = captured[0]
    assert description == "apply-agent-keepalive"
    # Registers the keep-alive scheduled task...
    assert "WinpodxAgentKeepAlive" in payload
    assert "Register-ScheduledTask" in payload
    # ...with BOTH an AtLogOn trigger and a 1-minute repetition.
    assert "New-ScheduledTaskTrigger -AtLogOn" in payload
    assert "New-TimeSpan -Minutes 1" in payload
    # Interactive user principal -- NOT SYSTEM / S4U -- so the agent's
    # /exec keeps the user's HKCU + Start Menu context for discovery /
    # reverse-open. Regression guard: an S4U / SYSTEM flip here would
    # silently change the discovery context.
    assert "-LogonType Interactive" in payload
    assert "RunLevel Limited" in payload
    assert "SYSTEM" not in payload
    assert "-LogonType S4U" not in payload
    # Launches windowless via the hidden-launcher wrapper (no console flash).
    assert "hidden-launcher.vbs" in payload
    # Staged from vbs_launchers, copied to the persistent C:\winpodx run dir.
    assert "C:\\winpodx\\agent-keepalive.ps1" in payload


def test_apply_agent_keepalive_raises_on_nonzero_rc(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    _mock_run_in_windows(monkeypatch, rc=2, stderr="schtasks denied")
    with pytest.raises(RuntimeError, match="rc=2"):
        provisioner._apply_agent_keepalive(cfg)


def test_apply_guest_share_skips_manual(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    captured = _mock_run_in_windows(monkeypatch)
    provisioner._apply_guest_share(cfg)
    assert captured == []


def test_apply_guest_share_payload_creates_share(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config
    from winpodx.core.guest_disk import GUEST_SMB_SHARE

    cfg = Config()
    cfg.rdp.user = "WPX-User"
    captured = _mock_run_in_windows(monkeypatch, rc=0, stdout="guest-share ok: winpodx-c -> C:\\")
    provisioner._apply_guest_share(cfg)
    assert len(captured) == 1
    description, payload = captured[0]
    assert description == "apply-guest-share"
    # Shares C:\ as the named share, granting the winpodx user.
    assert "New-SmbShare" in payload
    assert GUEST_SMB_SHARE in payload
    assert "-Path 'C:\\'" in payload
    assert "WPX-User" in payload
    # Service + local-account network auth + firewall are asserted.
    assert "LanmanServer" in payload
    assert "LocalAccountTokenFilterPolicy" in payload
    assert "File and Printer Sharing" in payload


def test_apply_guest_share_raises_on_nonzero_rc(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    _mock_run_in_windows(monkeypatch, rc=1, stderr="access denied")
    with pytest.raises(RuntimeError, match="rc=1"):
        provisioner._apply_guest_share(cfg)


def test_apply_oem_runtime_fixes_skips_manual(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    captured = _mock_run_in_windows(monkeypatch)
    provisioner._apply_oem_runtime_fixes(cfg)
    assert captured == []


def test_apply_oem_runtime_fixes_payload_contains_nic_and_termservice(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    captured = _mock_run_in_windows(monkeypatch, rc=0, stdout="oem v7 baseline applied")
    provisioner._apply_oem_runtime_fixes(cfg)
    assert len(captured) == 1
    description, payload = captured[0]
    assert description == "apply-oem"
    assert "Set-NetAdapterPowerManagement" in payload
    assert "AllowComputerToTurnOffDevice" in payload
    assert "sc.exe failure TermService" in payload
    assert "restart/5000/restart/5000/restart/5000" in payload


def test_ensure_ready_does_not_auto_apply_runtime_fixes(monkeypatch):
    """v0.2.2 (post-rollback Sprint 3): ensure_ready no longer auto-fires
    the 4 apply functions. install.bat applied them at first boot;
    upgrades use the explicit ``winpodx pod apply-fixes`` path.

    Regression: if anyone re-introduces the auto-apply, the next user app
    launch fires 4 FreeRDP RemoteApp PowerShell windows in sequence — the
    "PS창 깜빡깜빡" symptom kernalix7 reported on 2026-04-30."""
    from winpodx.core import provisioner
    from winpodx.core.config import Config
    from winpodx.core.pod import PodState, PodStatus

    cfg = Config()
    cfg.pod.backend = "podman"

    monkeypatch.setattr(provisioner, "_check_rotation_pending", lambda: None)
    monkeypatch.setattr(provisioner, "_auto_rotate_password", lambda c: c)
    monkeypatch.setattr(provisioner, "_ensure_config", lambda: cfg)
    monkeypatch.setattr(provisioner, "pod_status", lambda c: PodStatus(state=PodState.RUNNING))
    monkeypatch.setattr(provisioner, "check_rdp_port", lambda *a, **k: True)

    calls = {"max_sessions": 0, "rdp_timeouts": 0, "oem_runtime_fixes": 0, "multi_session": 0}

    def make_recorder(name):
        def f(c):
            calls[name] += 1

        return f

    monkeypatch.setattr(provisioner, "_apply_max_sessions", make_recorder("max_sessions"))
    monkeypatch.setattr(provisioner, "_apply_rdp_timeouts", make_recorder("rdp_timeouts"))
    monkeypatch.setattr(provisioner, "_apply_oem_runtime_fixes", make_recorder("oem_runtime_fixes"))
    monkeypatch.setattr(provisioner, "_apply_multi_session", make_recorder("multi_session"))

    result = provisioner.ensure_ready(cfg, timeout=1)
    assert result is cfg
    # ZERO calls — install.bat did the work. apply_windows_runtime_fixes
    # (manual / GUI button) is the explicit retry path.
    assert calls == {
        "max_sessions": 0,
        "rdp_timeouts": 0,
        "oem_runtime_fixes": 0,
        "multi_session": 0,
    }


def test_ensure_ready_skips_apply_when_pod_not_running(monkeypatch):
    """When pod isn't running, the early-apply branch is skipped (later branch handles)."""
    from winpodx.core import provisioner
    from winpodx.core.config import Config
    from winpodx.core.pod import PodState, PodStatus

    cfg = Config()
    cfg.pod.backend = "podman"

    monkeypatch.setattr(provisioner, "_check_rotation_pending", lambda: None)
    monkeypatch.setattr(provisioner, "_auto_rotate_password", lambda c: c)
    monkeypatch.setattr(provisioner, "_ensure_config", lambda: cfg)
    monkeypatch.setattr(provisioner, "pod_status", lambda c: PodStatus(state=PodState.STOPPED))
    monkeypatch.setattr(provisioner, "check_rdp_port", lambda *a, **k: True)
    early_calls = {"n": 0}

    def recorder(c):
        early_calls["n"] += 1

    monkeypatch.setattr(provisioner, "_apply_oem_runtime_fixes", recorder)
    monkeypatch.setattr(provisioner, "_apply_max_sessions", recorder)
    monkeypatch.setattr(provisioner, "_apply_rdp_timeouts", recorder)
    monkeypatch.setattr(provisioner, "_apply_multi_session", recorder)

    provisioner.ensure_ready(cfg, timeout=1)
    # Stopped pod -> the early-branch `pod_status==RUNNING` guard prevents
    # the apply calls from firing on the early return path.
    assert early_calls["n"] == 0


# --- v0.1.9.3: apply_windows_runtime_fixes public API ---


def test_apply_windows_runtime_fixes_skips_manual():
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()
    cfg.pod.backend = "manual"
    result = provisioner.apply_windows_runtime_fixes(cfg)
    assert "backend" in result
    assert "skipped" in result["backend"]


def test_apply_windows_runtime_fixes_returns_per_helper_status(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config
    from winpodx.core.windows_exec import WindowsExecResult

    cfg = Config()

    def fake(cfg_inner, payload, *, timeout=60, description="windows-exec"):
        return WindowsExecResult(rc=0, stdout="ok", stderr="")

    monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", fake)
    result = provisioner.apply_windows_runtime_fixes(cfg)
    assert set(result.keys()) == {
        "max_sessions",
        "rdp_timeouts",
        "oem_runtime_fixes",
        "multi_session",
        "vbs_launchers",
        "agent_keepalive",
        "guest_share",
    }
    for v in result.values():
        assert v == "ok"


def test_apply_windows_runtime_fixes_records_individual_failures(monkeypatch):
    from winpodx.core import provisioner
    from winpodx.core.config import Config

    cfg = Config()

    def fake_max_sessions(c):
        raise RuntimeError("boom max_sessions")

    monkeypatch.setattr(provisioner, "_apply_max_sessions", fake_max_sessions)
    monkeypatch.setattr(provisioner, "_apply_rdp_timeouts", lambda c: None)
    monkeypatch.setattr(provisioner, "_apply_oem_runtime_fixes", lambda c: None)
    result = provisioner.apply_windows_runtime_fixes(cfg)
    assert result["max_sessions"].startswith("failed: ")
    assert result["rdp_timeouts"] == "ok"
    assert result["oem_runtime_fixes"] == "ok"


# --- v0.2.0.6: wait_for_windows_responsive retry loop ---


class TestWaitForWindowsResponsiveRetries:
    """v0.2.2 (post-rollback Sprint 4): readiness probe is HTTP /health, NOT
    FreeRDP RemoteApp ping. The FreeRDP probe was the source of the
    "PowerShell 창 폭주" symptom kernalix7 reported on 2026-04-30 — every
    3-second tick fired one PS-window flash for the entire timeout.

    The HTTP path is invisible (loopback, no UI) and the agent.ps1 listener
    only binds AFTER install.bat finishes, making /health an unambiguous
    "Windows is ready" signal."""

    def _cfg(self):
        from winpodx.core.config import Config

        cfg = Config()
        cfg.pod.backend = "podman"
        cfg.rdp.ip = "127.0.0.1"
        cfg.rdp.port = 3389
        cfg.rdp.password = "abc123"
        return cfg

    def test_returns_true_when_health_eventually_responds(self, monkeypatch):
        """First N /health probes return unavailable; eventually one
        responds available and the helper must return True instead of
        bailing on probe #1."""
        from winpodx.core.provisioner import wait_for_windows_responsive
        from winpodx.core.transport.base import HealthStatus

        cfg = self._cfg()
        monkeypatch.setattr(
            "winpodx.core.provisioner.check_rdp_port",
            lambda ip, port, timeout=1.0: True,
        )
        # Compress the inter-probe sleep so the test is fast.
        monkeypatch.setattr("winpodx.core.provisioner.time.sleep", lambda _: None)

        attempts: list[int] = []

        def fake_health(self):
            attempts.append(len(attempts))
            if len(attempts) < 4:
                return HealthStatus(available=False, detail="agent still booting")
            return HealthStatus(available=True, version="0.2.2-rev1")

        monkeypatch.setattr("winpodx.core.transport.agent.AgentTransport.health", fake_health)
        assert wait_for_windows_responsive(cfg, timeout=60) is True
        assert len(attempts) >= 4, "must keep polling past first unavailable"

    def test_returns_true_when_rdp_up_but_agent_never_responds(self, monkeypatch):
        """If RDP is up but /health never responds within the agent
        budget, the helper now returns ``True`` (Windows is responsive
        — host code can fall back to FreeRDP RemoteApp). Pre-fix this
        returned ``False`` after polling /health for the full timeout,
        which deadlocked install.sh's wait-ready phase 3 for 60 minutes
        on pods where the agent didn't come up cleanly. Polling must
        still happen — we don't return on the first failure — but the
        eventual outcome is True with a logged warning, not False."""
        from winpodx.core.provisioner import wait_for_windows_responsive
        from winpodx.core.transport.base import HealthStatus

        cfg = self._cfg()
        monkeypatch.setattr(
            "winpodx.core.provisioner.check_rdp_port",
            lambda ip, port, timeout=1.0: True,
        )
        # Virtual clock so the test doesn't actually wait.
        clock = {"t": 0.0}
        monkeypatch.setattr("winpodx.core.provisioner.time.monotonic", lambda: clock["t"])
        monkeypatch.setattr(
            "winpodx.core.provisioner.time.sleep",
            lambda s: clock.update(t=clock["t"] + s),
        )

        attempts: list[int] = []

        def fake_health(self):
            attempts.append(0)
            # Each probe consumes ~2s of virtual time (HEALTH_TIMEOUT budget).
            clock["t"] += 2
            return HealthStatus(available=False, detail="connection refused")

        monkeypatch.setattr("winpodx.core.transport.agent.AgentTransport.health", fake_health)

        result = wait_for_windows_responsive(cfg, timeout=120)
        assert result is True
        assert len(attempts) >= 2, "must retry rather than bail on first failure"