kernalix7--winpodx/tests/test_windows_exec.py

# SPDX-License-Identifier: MIT
"""Tests for winpodx.core.windows_exec — FreeRDP RemoteApp PowerShell channel."""

from __future__ import annotations

import json
import subprocess
from unittest.mock import MagicMock

import pytest

from winpodx.core.config import Config
from winpodx.core.windows_exec import WindowsExecError, WindowsExecResult, run_in_windows


def _cfg(password: str = "secret") -> Config:
    cfg = Config()
    cfg.pod.backend = "podman"
    cfg.rdp.ip = "127.0.0.1"
    cfg.rdp.port = 3390
    cfg.rdp.user = "User"
    cfg.rdp.password = password
    return cfg


def _patch_data_dir(monkeypatch, tmp_path):
    """Force windows_exec to use tmp_path so the test can predict the result file."""
    fake_home = tmp_path / "home"
    fake_home.mkdir()
    fake_data = fake_home / ".local" / "share" / "winpodx"
    fake_data.mkdir(parents=True)
    monkeypatch.setattr("winpodx.core.windows_exec.data_dir", lambda: fake_data)
    monkeypatch.setattr("winpodx.core.windows_exec.Path.home", staticmethod(lambda: fake_home))
    return fake_home, fake_data


def test_run_in_windows_raises_when_freerdp_missing(monkeypatch, tmp_path):
    _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr("winpodx.core.windows_exec.find_freerdp", lambda: None)
    with pytest.raises(WindowsExecError, match="FreeRDP not found"):
        run_in_windows(_cfg(), "Write-Output 'hello'")


def test_run_in_windows_raises_when_password_missing(monkeypatch, tmp_path):
    _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )
    cfg = _cfg(password="")
    with pytest.raises(WindowsExecError, match="password not set"):
        run_in_windows(cfg, "Write-Output 'hello'")


def test_run_in_windows_raises_on_freerdp_timeout(monkeypatch, tmp_path):
    _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    def boom(cmd, **kw):
        raise subprocess.TimeoutExpired(cmd, kw.get("timeout", 0))

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", boom)
    with pytest.raises(WindowsExecError, match="timed out"):
        run_in_windows(_cfg(), "Write-Output 'hello'", timeout=1)


def test_run_in_windows_raises_when_no_result_file(monkeypatch, tmp_path):
    """FreeRDP exits but the wrapper PS never wrote a result — likely auth/connect failure."""
    _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    def fake_run(cmd, **kw):
        # FreeRDP "succeeded" (rc=0) but produced no result file.
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = "auth failure: NTLM"
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    with pytest.raises(WindowsExecError, match="No result file"):
        run_in_windows(_cfg(), "Write-Output 'hello'")


def test_run_in_windows_returns_result_when_wrapper_writes_json(monkeypatch, tmp_path):
    """Happy path — simulate the wrapper writing the JSON result file."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)
    result_path = work_dir / "winpodx-exec-result.json"

    def fake_run(cmd, **kw):
        # Simulate the in-guest wrapper writing the result file before exiting.
        result_path.write_text(
            json.dumps({"rc": 0, "stdout": "registry updated", "stderr": ""}),
            encoding="utf-8",
        )
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    result = run_in_windows(_cfg(), "Write-Output 'registry updated'")
    assert isinstance(result, WindowsExecResult)
    assert result.rc == 0
    assert result.ok is True
    assert "registry updated" in result.stdout
    # Result file cleaned up afterwards.
    assert not result_path.exists()


def test_run_in_windows_propagates_nonzero_rc_from_wrapper(monkeypatch, tmp_path):
    """If the user payload returned non-zero, surface it in WindowsExecResult.rc."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)
    result_path = work_dir / "winpodx-exec-result.json"

    def fake_run(cmd, **kw):
        result_path.write_text(
            json.dumps({"rc": 7, "stdout": "", "stderr": "boom"}), encoding="utf-8"
        )
        m = MagicMock()
        m.returncode = 7
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    result = run_in_windows(_cfg(), "exit 7")
    assert result.rc == 7
    assert result.ok is False
    assert result.stderr == "boom"


def test_run_in_windows_command_uses_freerdp_app_remoteapp(monkeypatch, tmp_path):
    """The cmd passed to subprocess.run must look like a RemoteApp /app:program: launch."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    captured: list[list[str]] = []
    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)

    def fake_run(cmd, **kw):
        captured.append(list(cmd))
        # Write a successful result so the function returns cleanly.
        (work_dir / "apply-X-result.json").write_text(
            json.dumps({"rc": 0, "stdout": "", "stderr": ""}), encoding="utf-8"
        )
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    run_in_windows(_cfg(), "Write-Output 'hi'", description="apply-X")

    assert len(captured) == 1
    cmd = captured[0]
    assert cmd[0] == "/usr/bin/xfreerdp"
    # Standard winpodx RDP flags must be there.
    joined = " ".join(cmd)
    assert "/v:127.0.0.1:3390" in joined
    assert "/u:User" in joined
    assert "/p:secret" in joined
    assert "+home-drive" in joined
    assert "/sec:tls" in joined
    assert "/cert:ignore" in joined
    # The /app:program: payload should target powershell.exe with -File pointing
    # at a tsclient UNC path.
    app_arg = next((a for a in cmd if a.startswith("/app:program:")), "")
    assert "powershell.exe" in app_arg
    assert "tsclient" in app_arg
    assert "-WindowStyle Hidden" in app_arg
    assert "-NoProfile" in app_arg
    assert "-File" in app_arg


def test_run_in_windows_supports_flatpak_freerdp_with_spaces(monkeypatch, tmp_path):
    """flatpak-style binary 'flatpak run com.freerdp.FreeRDP' must split correctly."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp",
        lambda: ("flatpak run com.freerdp.FreeRDP", "flatpak"),
    )

    captured: list[list[str]] = []
    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)

    def fake_run(cmd, **kw):
        captured.append(list(cmd))
        (work_dir / "winpodx-exec-result.json").write_text(
            json.dumps({"rc": 0, "stdout": "", "stderr": ""}), encoding="utf-8"
        )
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    run_in_windows(_cfg(), "Write-Output 'flatpak'")
    assert captured[0][:3] == ["flatpak", "run", "com.freerdp.FreeRDP"]


def test_run_in_windows_handles_unparseable_result_file(monkeypatch, tmp_path):
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)
    result_path = work_dir / "winpodx-exec-result.json"

    def fake_run(cmd, **kw):
        result_path.write_text("not valid json", encoding="utf-8")
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    with pytest.raises(WindowsExecError, match="unparseable"):
        run_in_windows(_cfg(), "Write-Output 'hi'")


# --- v0.2.0: streaming progress callback -----------------------------------


def test_run_in_windows_streams_progress_lines(monkeypatch, tmp_path):
    """Streaming path: payload writes lines to the progress file; host tails them."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)
    progress_path = work_dir / "stream-test-progress.log"
    result_path = work_dir / "stream-test-result.json"

    received: list[str] = []

    # Simulate FreeRDP running for ~1s, writing progress lines mid-run, then
    # completing normally with a result file.
    def fake_popen(cmd, **kw):
        # Pre-populate result so the post-loop drain succeeds.
        progress_path.write_text("Stage A\nStage B\n", encoding="utf-8")

        m = MagicMock()
        m.stdout = MagicMock()
        m.stderr = MagicMock()
        # poll() returns None twice (still running), then 0 (done).
        m._poll_count = 0

        def _poll():
            m._poll_count += 1
            if m._poll_count <= 2:
                # While "running", write more progress between polls.
                if m._poll_count == 2:
                    progress_path.write_text("Stage A\nStage B\nStage C\n", encoding="utf-8")
                return None
            return 0

        m.poll.side_effect = _poll
        m.returncode = 0
        m.kill.return_value = None
        m.wait.return_value = 0
        m.communicate.return_value = ("", "")
        # Write the JSON result so the post-popen parse succeeds.
        result_path.write_text(
            json.dumps({"rc": 0, "stdout": "done", "stderr": ""}), encoding="utf-8"
        )
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.Popen", fake_popen)

    # Speed up the polling loop.
    monkeypatch.setattr("winpodx.core.windows_exec.time.sleep", lambda _x: None)

    result = run_in_windows(
        _cfg(),
        "Write-WinpodxProgress 'Stage A'",
        description="stream-test",
        progress_callback=received.append,
    )
    assert result.rc == 0
    assert "Stage A" in received
    assert "Stage B" in received
    assert "Stage C" in received


def test_run_in_windows_progress_file_created_only_when_callback_set(monkeypatch, tmp_path):
    """No progress_callback -> no progress file pre-created (we still cleanup it
    in finally, but the simple path doesn't try to read it)."""
    _, fake_data = _patch_data_dir(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "winpodx.core.windows_exec.find_freerdp", lambda: ("/usr/bin/xfreerdp", "xfreerdp")
    )

    work_dir = fake_data / "windows-exec"
    work_dir.mkdir(parents=True, exist_ok=True)

    def fake_run(cmd, **kw):
        (work_dir / "winpodx-exec-result.json").write_text(
            json.dumps({"rc": 0, "stdout": "", "stderr": ""}), encoding="utf-8"
        )
        m = MagicMock()
        m.returncode = 0
        m.stdout = ""
        m.stderr = ""
        return m

    monkeypatch.setattr("winpodx.core.windows_exec.subprocess.run", fake_run)
    run_in_windows(_cfg(), "Write-Output 'hi'")
    # Without callback, the synchronous subprocess.run path is taken; no
    # progress file should remain (cleanup runs unconditionally).
    progress_files = list(work_dir.glob("*-progress.log"))
    assert progress_files == []