SyncVito/tests/test_update_watcher.py

"""Führt den echten PowerShell-Wächter mit harmlosen Testprogrammen aus."""

import base64
import json
import os
import shutil
import subprocess
import time
from pathlib import Path

import pytest

from syncvito import updater

pytestmark = pytest.mark.skipif(os.name != "nt", reason="Windows-Prozesslebenszyklus")


def wait_for_file(path: Path, timeout: float = 15) -> None:
    deadline = time.monotonic() + timeout
    while not path.exists():
        if time.monotonic() > deadline:
            pytest.fail(f"Testprozess hat keine Datei geschrieben: {path}")
        time.sleep(0.05)


@pytest.fixture(scope="module")
def helper_exe(tmp_path_factory: pytest.TempPathFactory) -> Path:
    root = tmp_path_factory.mktemp("watcher-helper")
    source = root / "helper.cs"
    source.write_text(
        """
using System;
using System.IO;
using System.Threading;
class Helper {
    static int Main(string[] args) {
        File.WriteAllLines(args[0], args);
        var deadline = DateTime.UtcNow.AddSeconds(30);
        while (!File.Exists(args[1])) {
            if (DateTime.UtcNow > deadline) return 99;
            Thread.Sleep(50);
        }
        return Int32.Parse(args[2]);
    }
}
""",
        encoding="utf-8",
    )
    compiler = Path(os.environ.get("SYSTEMROOT", r"C:\Windows")) / "Microsoft.NET/Framework64/v4.0.30319/csc.exe"
    if not compiler.exists():
        pytest.skip("Windows-.NET-Compiler für Testprogramm nicht vorhanden")
    executable = root / "SyncVitoWatcherTest.exe"
    subprocess.run(
        [str(compiler), "/nologo", "/target:winexe", f"/out:{executable}", str(source)],
        check=True,
        capture_output=True,
        timeout=30,
    )
    return executable


@pytest.mark.parametrize("exit_code", [0, 5])
def test_watcher_waits_for_all_instances_and_real_setup_exit(
    helper_exe: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, exit_code: int
) -> None:
    root = tmp_path / "Ordner mit Leerzeichen und 'Apostroph'"
    root.mkdir()
    installer = root / "SetupTest.exe"
    shutil.copy2(helper_exe, installer)
    unrelated_dir = root / "anderer Installationspfad"
    unrelated_dir.mkdir()
    unrelated_exe = unrelated_dir / helper_exe.name
    shutil.copy2(helper_exe, unrelated_exe)
    children = []
    releases = []

    def start_client(exe: Path, name: str) -> subprocess.Popen:
        ready = root / f"{name}.ready"
        release = root / f"{name}.release"
        releases.append(release)
        process = subprocess.Popen([str(exe), str(ready), str(release), "0"])
        children.append(process)
        wait_for_file(ready)
        return process

    def launch_watcher(executable: str, arguments: list[str]) -> None:
        children.append(subprocess.Popen([executable, *arguments], creationflags=subprocess.CREATE_NO_WINDOW))

    try:
        parent = start_client(helper_exe, "parent")
        extra = start_client(helper_exe, "extra")
        unrelated = start_client(unrelated_exe, "unrelated")
        monkeypatch.setattr(updater, "_start_detached", launch_watcher)
        output = root / "setup.arguments"
        setup_release = root / "setup.release"
        releases.append(setup_release)
        log = root / "bootstrap.log"
        arguments = [str(output), str(setup_release), str(exit_code), f"/DIR={root}", f"/LOG={root / 'setup.log'}"]
        updater._start_installer_after_exit(installer, arguments, [parent.pid], log, helper_exe)
        watcher = children[-1]
        wait_for_file(log)
        assert not output.exists()
        releases[0].touch()
        parent.wait(timeout=10)
        time.sleep(0.5)
        assert not output.exists(), "Weitere Instanz wurde nicht abgewartet"
        releases[1].touch()
        extra.wait(timeout=10)
        wait_for_file(output)
        assert unrelated.poll() is None, "Anderer Installationspfad darf weiterlaufen"
        assert output.read_text(encoding="utf-8-sig").splitlines() == arguments
        assert watcher.poll() is None, "Wächter muss auf Setup warten"
        assert not (root / "update-result.json").exists()
        setup_release.touch()
        assert watcher.wait(timeout=15) == (0 if exit_code == 0 else 1)
        receipt = json.loads((root / "update-result.json").read_text(encoding="utf-8"))
        assert receipt["success"] is (exit_code == 0)
        assert receipt["exit_code"] == exit_code
        assert f"Installer-Exitcode: {exit_code}" in log.read_text(encoding="utf-8-sig")
    finally:
        for release in releases:
            release.touch()
        for child in children:
            try:
                child.wait(timeout=10)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait(timeout=5)


def test_watcher_reports_timeout_without_starting_installer(
    helper_exe: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    ready = tmp_path / "client.ready"
    release = tmp_path / "client.release"
    log = tmp_path / "bootstrap.log"
    client = subprocess.Popen([str(helper_exe), str(ready), str(release), "0"])
    watchers = []

    def launch_with_short_deadline(executable: str, arguments: list[str]) -> None:
        script = base64.b64decode(arguments[-1]).decode("utf-16-le")
        script = script.replace("AddMinutes(2)", "AddMilliseconds(400)")
        arguments[-1] = base64.b64encode(script.encode("utf-16-le")).decode("ascii")
        watchers.append(subprocess.Popen([executable, *arguments], creationflags=subprocess.CREATE_NO_WINDOW))

    try:
        wait_for_file(ready)
        monkeypatch.setattr(updater, "_start_detached", launch_with_short_deadline)
        updater._start_installer_after_exit(tmp_path / "not-started.exe", ["/VERYSILENT"], [], log, helper_exe)
        assert watchers[0].wait(timeout=15) == 1
        receipt = json.loads((tmp_path / "update-result.json").read_text(encoding="utf-8"))
        assert not receipt["success"]
        assert str(client.pid) in receipt["message"]
        assert "Beenden" in receipt["message"]
        assert "Starte verifizierten Installer" not in log.read_text(encoding="utf-8-sig")
        assert client.poll() is None
    finally:
        release.touch()
        client.wait(timeout=10)
        for watcher in watchers:
            watcher.wait(timeout=10)