"""Verify frozen upgrades and process ownership in disposable Windows installations."""

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

import pytest

ROOT = Path(__file__).resolve().parents[2]
UPGRADE_FILES = (
    "upgrade.ps1",
    "upgrade_processes.ps1",
    "installation_lock.ps1",
    "update_installation.py",
    "windows_tools.ps1",
    "download_config.ps1",
    "bootstrap_python.ps1",
)
pytestmark = pytest.mark.skipif(
    sys.platform != "win32" or not shutil.which("powershell.exe") or not shutil.which("uv"),
    reason="Windows PowerShell and uv required",
)


def powershell(script, *, env=None, cwd=None):
    return subprocess.run(
        ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", str(script)],
        cwd=cwd,
        env={**os.environ, **(env or {})},
        capture_output=True,
        encoding="utf-8",
        timeout=45,
    )


def git(root, *args):
    return (
        subprocess.run(["git", "-C", str(root), *args], capture_output=True, check=True, timeout=15)
        .stdout.decode()
        .strip()
    )


def make_venv(root):
    subprocess.run(
        ["uv", "venv", "--offline", "--python", sys.executable, str(root / ".venv")],
        check=True,
        capture_output=True,
        timeout=30,
    )
    return root / ".venv/Scripts/python.exe"


def test_process_selection_is_scoped_and_includes_descendants(tmp_path):
    root = tmp_path / "project"

    def process(pid, executable, command, name="python.exe", parent=0):
        return dict(
            ProcessId=pid,
            ParentProcessId=parent,
            ExecutablePath=executable,
            CommandLine=command,
            Name=name,
        )

    processes = [
        process(10, str(root / ".venv/Scripts/python.exe"), "python -m artemis ui"),
        process(11, "C:\\Python\\python.exe", "python -m multiprocessing.spawn", parent=10),
        process(12, "C:\\tools\\ffmpeg.exe", "ffmpeg recording", "ffmpeg.exe", parent=11),
        process(13, "C:\\tools\\adb.exe", "adb server", "adb.exe", parent=10),
        process(14, "C:\\apps\\codex.exe", "codex", "codex.exe", parent=10),
        process(
            20,
            str(root.with_name("project-other") / ".venv/Scripts/python.exe"),
            "python -m artemis ui",
        ),
        process(21, "C:\\Python\\python.exe", "python -m artemis ui"),
        process(22, "C:\\Python\\python.exe", f'python "{root}/mcp_server/__main__.py"'),
        process(23, "C:\\Python\\python.exe", f'python "{root}/unrelated.py"'),
        process(24, str(root / ".venv/Scripts/python.exe"), "protected updater"),
        process(25, "C:\\tools\\emulator.exe", "emulator", "emulator.exe", parent=10),
    ]
    fixture = tmp_path / "processes.json"
    fixture.write_text(json.dumps(processes), encoding="utf-8")
    script = tmp_path / "select.ps1"
    script.write_text(
        "[Console]::OutputEncoding = New-Object Text.UTF8Encoding($false)\n"
        f". '{ROOT / 'scripts/upgrade_processes.ps1'}'\n"
        "$snapshot = Get-Content -Raw -LiteralPath $env:PROCESS_FIXTURE | ConvertFrom-Json\n"
        "$ids = @(Get-ArtemisUpgradeProcesses -ProjectRoot $env:PROJECT_ROOT -Processes $snapshot -ProtectedIds @(24) | ForEach-Object { [int]$_.ProcessId } | Sort-Object)\n"
        "ConvertTo-Json -InputObject $ids -Compress\n",
        encoding="ascii",
    )
    result = powershell(script, env={"PROCESS_FIXTURE": str(fixture), "PROJECT_ROOT": str(root)})
    assert result.returncode == 0, result.stdout + result.stderr
    assert json.loads(result.stdout) == [10, 11, 12, 22]


def test_real_project_process_tree_stops_but_unrelated_process_survives(tmp_path):
    import psutil

    root = tmp_path / "project with spaces!"
    root.mkdir()
    python = make_venv(root)
    pid_file = root / "child.pid"
    sleeper = root / "sleeper.py"
    sleeper.write_text(
        "import pathlib, subprocess, sys, time\n"
        "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n"
        "pathlib.Path(sys.argv[1]).write_text(str(child.pid))\n"
        "time.sleep(60)\n",
        encoding="ascii",
    )
    project = subprocess.Popen([str(python), str(sleeper), str(pid_file)])
    unrelated = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
    child_pid = None
    try:
        for _ in range(100):
            if pid_file.exists():
                child_pid = int(pid_file.read_text())
                break
            time.sleep(0.05)
        assert child_pid is not None
        script = tmp_path / "stop.ps1"
        script.write_text(
            "$ErrorActionPreference = 'Stop'\n"
            f". '{ROOT / 'scripts/upgrade_processes.ps1'}'\n"
            "Stop-ArtemisUpgradeProcesses -ProjectRoot $env:PROJECT_ROOT\n",
            encoding="ascii",
        )
        result = powershell(script, env={"PROJECT_ROOT": str(root)})
        assert result.returncode == 0, result.stdout + result.stderr
        project.wait(timeout=5)
        assert unrelated.poll() is None
        assert not psutil.pid_exists(child_pid)
    finally:
        for process in (project, unrelated):
            if process.poll() is None:
                process.kill()
            process.wait(timeout=5)
        if child_pid and psutil.pid_exists(child_pid):
            psutil.Process(child_pid).kill()


@pytest.mark.parametrize("prepare_exit", [0, 7])
@pytest.mark.parametrize("initialized", [True, False])
def test_frozen_upgrade_survives_launcher_replacement_and_reports_setup_failure(
    tmp_path, prepare_exit, initialized
):
    if not shutil.which("git"):
        pytest.skip("Git required")
    author = tmp_path / "author"
    (author / "scripts").mkdir(parents=True)
    for name in UPGRADE_FILES:
        shutil.copyfile(ROOT / "scripts" / name, author / "scripts" / name)
    (author / "config").mkdir()
    shutil.copyfile(ROOT / "config/downloads.json", author / "config/downloads.json")
    (author / "pyproject.toml").write_text(
        '[project]\nname="upgrade-fixture"\nversion="0.1.0"\nrequires-python=">=3.12"\ndependencies=[]\n'
    )
    (author / "uv.lock").write_text('version = 1\nrequires-python = ">=3.12"\n')
    for name in ("update.bat", "update-cn.bat"):
        shutil.copyfile(ROOT / name, author / name)
    (author / ".gitignore").write_text(".env\n.venv/\nprepared.txt\n")
    (author / "scripts/start.ps1").write_text("exit 23\n", encoding="ascii")
    git(author, "init", "-b", "main")
    git(author, "config", "user.name", "Fixture")
    git(author, "config", "user.email", "fixture@example.test")
    git(author, "add", ".")
    git(author, "commit", "-m", "old version")
    installed = tmp_path / "installed 中文 with spaces!"
    git(tmp_path, "clone", "-b", "main", str(author), str(installed))
    if initialized:
        make_venv(installed)
    (installed / ".env").write_text("secret stays local")
    # A running file AND both BAT wrappers are replaced during the merge.
    (author / "scripts/upgrade.ps1").write_text("throw 'new updater must not run in this turn'\n")
    (author / "scripts/upgrade_processes.ps1").write_text(
        "throw 'new process helper must not run'\n"
    )
    for name in ("windows_tools.ps1", "download_config.ps1", "bootstrap_python.ps1"):
        (author / "scripts" / name).write_text("throw 'new bootstrap helper must not run'\n")
    (author / "update.bat").write_text("@echo off\nexit /b 33\n")
    (author / "update-cn.bat").write_text("@echo off\nexit /b 34\n")
    (author / "scripts/start.ps1").write_text(
        "param([switch]$PrepareOnly, [switch]$SkipUpdate, [switch]$NoOpen, [string]$Mirror)\n"
        "if (-not ($PrepareOnly -and $SkipUpdate -and $NoOpen) -or $Mirror -ne 'China' -or $env:ARTEMIS_UPGRADE_CHILD -ne '1') { exit 35 }\n"
        "Set-Content -LiteralPath (Join-Path (Split-Path $PSScriptRoot -Parent) 'prepared.txt') -Value 'new version prepared'\n"
        f"exit {prepare_exit}\n",
        encoding="ascii",
    )
    git(author, "add", ".")
    git(author, "commit", "-m", "new version")
    result = subprocess.run(
        ["cmd.exe", "/d", "/c", "update-cn.bat"],
        cwd=installed,
        env={**os.environ, "UV_OFFLINE": "1", "ARTEMIS_NO_PAUSE": "1"},
        capture_output=True,
        encoding="utf-8",
        timeout=45,
    )
    assert result.returncode == (0 if prepare_exit == 0 else 1), result.stdout + result.stderr
    if initialized:
        assert result.stderr == ""
    else:
        # uv prints normal venv creation progress to stderr.
        assert "Creating virtual environment" in result.stderr
    assert (installed / "prepared.txt").read_text().strip() == "new version prepared"
    assert git(installed, "rev-parse", "HEAD") == git(author, "rev-parse", "HEAD")
    assert (installed / ".env").read_text() == "secret stays local"
    assert ("升级完成" in result.stdout) == (prepare_exit == 0)
    assert "new updater must not run" not in result.stdout
    assert "new bootstrap helper must not run" not in result.stdout
    assert ("首次准备" in result.stdout) == (not initialized)
    assert (installed / ".venv/Scripts/python.exe").exists()


@pytest.mark.parametrize("name", ["update.bat", "update-cn.bat"])
@pytest.mark.parametrize("help_flag", ["-Help", "--help"])
def test_upgrade_help_is_safe_and_wrappers_are_ascii(name, help_flag):
    (ROOT / name).read_bytes().decode("ascii")
    result = subprocess.run(
        ["cmd.exe", "/d", "/c", name, help_flag],
        cwd=ROOT,
        capture_output=True,
        encoding="utf-8",
        timeout=15,
    )
    assert result.returncode == 0, result.stdout + result.stderr
    assert result.stderr == ""
    assert "升级会终止本项目" in result.stdout


def test_installation_lock_blocks_same_checkout_and_recovers_after_owner_crash(tmp_path):
    holder_script = tmp_path / "holder.ps1"
    holder_script.write_text(
        f". '{ROOT / 'scripts/installation_lock.ps1'}'\n"
        "$lock = Enter-ArtemisInstallationLock -ProjectRoot $env:PROJECT_ROOT\n"
        "Set-Content -LiteralPath $env:READY_FILE -Value 'ready'\n"
        "Start-Sleep -Seconds 60\n",
        encoding="ascii",
    )
    ready = tmp_path / "ready.txt"
    root = tmp_path / "project"
    holder = subprocess.Popen(
        ["powershell.exe", "-NoProfile", "-File", str(holder_script)],
        env={**os.environ, "PROJECT_ROOT": str(root), "READY_FILE": str(ready)},
    )
    contender = tmp_path / "contender.ps1"
    contender.write_text(
        f". '{ROOT / 'scripts/installation_lock.ps1'}'\n"
        "try {\n"
        "  $lock = Enter-ArtemisInstallationLock -ProjectRoot $env:PROJECT_ROOT\n"
        "  Exit-ArtemisInstallationLock -Mutex $lock\n"
        "  exit 0\n"
        "} catch { Write-Host 'LOCK_BUSY'; exit 37 }\n",
        encoding="ascii",
    )
    try:
        for _ in range(100):
            if ready.exists():
                break
            time.sleep(0.05)
        assert ready.exists()
        blocked = powershell(contender, env={"PROJECT_ROOT": str(root)})
        assert blocked.returncode == 37, blocked.stdout + blocked.stderr
        assert (
            powershell(contender, env={"PROJECT_ROOT": str(root.with_name("other"))}).returncode
            == 0
        )
        holder.kill()
        holder.wait(timeout=5)
        assert powershell(contender, env={"PROJECT_ROOT": str(root)}).returncode == 0
    finally:
        if holder.poll() is None:
            holder.kill()
        holder.wait(timeout=5)


def test_actual_preparation_build_checks_run_without_launching_ui_or_adb(tmp_path):
    root = tmp_path / "prepared project"
    (root / "scripts").mkdir(parents=True)
    for name in ("start.ps1", "installation_lock.ps1", "windows_tools.ps1"):
        shutil.copyfile(ROOT / "scripts" / name, root / "scripts" / name)
    (root / "scripts/download_config.ps1").write_text(
        "function Initialize-ArtemisDownloads { $script:ArtemisDownloads = @{mode='Official'} }\n",
        encoding="ascii",
    )
    (root / "scripts/bootstrap_python.ps1").write_text(
        "function Initialize-ArtemisPython { return $env:TEST_PYTHON }\n", encoding="ascii"
    )
    (root / "scripts/update_installation.py").write_text(
        "import pathlib, sys\n"
        "root = pathlib.Path(__file__).parent.parent\n"
        "with (root / 'checks.txt').open('a') as f: f.write(sys.argv[1] + '\\n')\n"
        "sys.exit(0)\n",
        encoding="ascii",
    )
    (root / ".env").write_text("keep credentials")
    index = root / "apps/showcase_ui/dist/frontend/browser/index.html"
    index.parent.mkdir(parents=True)
    index.write_text("built frontend")
    tools = root / "stub-tools"
    tools.mkdir()
    for name in ("adb", "scrcpy", "ffmpeg"):
        (tools / f"{name}.cmd").write_text("@echo off\nexit /b 91\n")
    result = subprocess.run(
        [
            "powershell.exe",
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(root / "scripts/start.ps1"),
            "-PrepareOnly",
            "-SkipUpdate",
            "-NoOpen",
        ],
        env={
            **os.environ,
            "TEST_PYTHON": sys.executable,
            "PATH": str(tools) + os.pathsep + os.environ["PATH"],
        },
        capture_output=True,
        encoding="utf-8",
        timeout=15,
    )
    assert result.returncode == 0, result.stdout + result.stderr
    assert (root / "checks.txt").read_text().splitlines() == ["mcp", "frontend-check"]
    assert "Launching Artemis" not in result.stdout
    assert "ADB" not in result.stderr
    assert (root / ".env").read_text() == "keep credentials"


@pytest.mark.parametrize("initialized", [True, False])
def test_dirty_checkout_refuses_upgrade_before_stopping_processes(tmp_path, initialized):
    if not shutil.which("git"):
        pytest.skip("Git required")
    root = tmp_path / "dirty installation"
    (root / "scripts").mkdir(parents=True)
    for name in UPGRADE_FILES:
        shutil.copyfile(ROOT / "scripts" / name, root / "scripts" / name)
    (root / ".gitignore").write_text(".venv/\n")
    source = root / "user.txt"
    source.write_text("original")
    git(root, "init", "-b", "main")
    git(root, "config", "user.name", "Fixture")
    git(root, "config", "user.email", "fixture@example.test")
    git(root, "add", ".")
    git(root, "commit", "-m", "initial")
    python = make_venv(root) if initialized else sys.executable
    source.write_text("user changes")
    worker = subprocess.Popen([str(python), "-c", "import time; time.sleep(60)"])
    try:
        result = powershell(root / "scripts/upgrade.ps1")
        assert result.returncode == 1, result.stdout + result.stderr
        assert "发现本地修改" in result.stdout
        assert "[STOP]" not in result.stdout
        assert worker.poll() is None
        assert source.read_text() == "user changes"
        assert (root / ".venv").exists() == initialized
    finally:
        # Windows venv redirectors can have a second interpreter process.
        import psutil

        if worker.poll() is None:
            children = psutil.Process(worker.pid).children(recursive=True)
            worker.kill()
            for child in children:
                if child.is_running():
                    child.kill()
        worker.wait(timeout=5)


@pytest.mark.parametrize("launcher", ["update.bat", "update-cn.bat"])
def test_fresh_non_git_directory_reports_error_without_bootstrapping_or_hanging(tmp_path, launcher):
    root = tmp_path / "未初始化 source with spaces!"
    (root / "scripts").mkdir(parents=True)
    for name in UPGRADE_FILES:
        shutil.copyfile(ROOT / "scripts" / name, root / "scripts" / name)
    for name in ("update.bat", "update-cn.bat"):
        shutil.copyfile(ROOT / name, root / name)
    (root / ".env").write_text("existing config must stay local")
    result = subprocess.run(
        ["cmd.exe", "/d", "/c", launcher],
        cwd=root,
        # Redirected input must not wait for a key, even without ARTEMIS_NO_PAUSE.
        env={key: value for key, value in os.environ.items() if key != "ARTEMIS_NO_PAUSE"},
        input="",
        capture_output=True,
        encoding="utf-8",
        timeout=15,
    )
    assert result.returncode == 1
    assert result.stderr == ""
    assert "[升级失败]" in result.stdout
    assert "此目录不含 Git 信息" in result.stdout
    assert "首次准备" not in result.stdout
    assert "�" not in result.stdout
    assert not (root / ".venv").exists()
    assert (root / ".env").read_text() == "existing config must stay local"