#!/usr/bin/env python3
"""Validate prebuilt monocoat release artifacts in isolated environments.

This is the local replacement for the wheel-smoke release gate. It never
builds, downloads, publishes, or touches the caller's desktop or config
directories. Every scenario gets its own throwaway HOME/XDG tree and a
constructor-built environment.
"""

from __future__ import annotations

import argparse
import json
import re
import shlex
import shutil
import subprocess
import sys
import tarfile
import tempfile
import zipfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath

ROOT = Path(__file__).resolve().parent.parent
OPTIONAL_TOOLS = (
    "sassc",
    "magick",
    "rsvg-convert",
    "xfconf-query",
    "picom",
    "polybar",
    "kitty",
)
NO_TRACEBACK = "Traceback (most recent call last)"


class ArtifactError(ValueError):
    """The --dist directory does not hold exactly one wheel and one sdist."""


class CheckFailure(AssertionError):
    """A scenario assertion failed."""


class SkipCheck(RuntimeError):
    """A scenario cannot run on this host."""


@dataclass(frozen=True)
class CommandResult:
    argv: tuple[str, ...]
    code: int
    stdout: str
    stderr: str


@dataclass(frozen=True)
class Check:
    name: str
    status: str
    detail: str


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse the public command line."""
    parser = argparse.ArgumentParser(
        prog="release-smoke",
        description=(
            "Validate a prebuilt monocoat wheel/sdist in isolated fresh "
            "environments."
        ),
    )
    parser.add_argument(
        "--dist",
        required=True,
        type=Path,
        metavar="DIR",
        help="directory holding one monocoat-*.whl and one monocoat-*.tar.gz",
    )
    parser.add_argument(
        "--keep",
        action="store_true",
        help="keep the temporary environments and print their location",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="print one JSON document instead of check lines",
    )
    return parser.parse_args(argv)


def clean_env(base: Path) -> dict[str, str]:
    """Return an environment containing only values rooted inside *base*.

    The desktop/session variables named in the release-smoke contract are
    deliberately absent, not cleared strings. The caller's environment is
    never consulted.
    """
    base = Path(base)
    return {
        "HOME": str(base / "home"),
        "XDG_CONFIG_HOME": str(base / "home" / ".config"),
        "XDG_CACHE_HOME": str(base / "home" / ".cache"),
        "XDG_DATA_HOME": str(base / "home" / ".local" / "share"),
        "XDG_STATE_HOME": str(base / "home" / ".local" / "state"),
        "XDG_RUNTIME_DIR": str(base / "runtime"),
        "PATH": f"{base / 'interpreter-bin'}:/usr/bin:/bin",
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
        "NO_COLOR": "1",
        "PYTHONNOUSERSITE": "1",
        "PYTHONDONTWRITEBYTECODE": "1",
        "PIP_DISABLE_PIP_VERSION_CHECK": "1",
        "PIP_NO_INPUT": "1",
        "TMPDIR": str(base / "tmp"),
    }


def _prepare_env(base: Path) -> dict[str, str]:
    """Create the directories and interpreter shims named by clean_env."""
    base.mkdir(parents=True, exist_ok=True)
    env = clean_env(base)
    for key in (
        "HOME",
        "XDG_CONFIG_HOME",
        "XDG_CACHE_HOME",
        "XDG_DATA_HOME",
        "XDG_STATE_HOME",
        "XDG_RUNTIME_DIR",
        "TMPDIR",
    ):
        Path(env[key]).mkdir(parents=True, exist_ok=True)
    Path(env["XDG_RUNTIME_DIR"]).chmod(0o700)
    bindir = Path(env["PATH"].split(":", 1)[0])
    bindir.mkdir(parents=True, exist_ok=True)
    executable = Path(sys.executable).resolve()
    for name in ("python", "python3"):
        link = bindir / name
        if not link.exists():
            link.symlink_to(executable)
    return env


def find_artifacts(directory: Path) -> tuple[Path, Path]:
    """Find exactly one wheel and one sdist in *directory*."""
    directory = Path(directory)
    if not directory.is_dir():
        raise ArtifactError(f"--dist is not a directory: {directory}")
    wheels = sorted(p for p in directory.glob("monocoat-*.whl") if p.is_file())
    sdists = sorted(p for p in directory.glob("monocoat-*.tar.gz") if p.is_file())
    if len(wheels) != 1:
        found = ", ".join(p.name for p in wheels) or "none"
        raise ArtifactError(
            f"expected one monocoat-*.whl in {directory}; found {len(wheels)} ({found})"
        )
    if len(sdists) != 1:
        found = ", ".join(p.name for p in sdists) or "none"
        raise ArtifactError(
            f"expected one monocoat-*.tar.gz in {directory}; found {len(sdists)} ({found})"
        )
    return wheels[0], sdists[0]


def wheel_version(wheel: Path) -> str:
    """Return the version encoded in a monocoat wheel filename."""
    wheel = Path(wheel)
    parts = wheel.name.removesuffix(".whl").split("-")
    if len(parts) != 5 or parts[0] != "monocoat" or not parts[1]:
        raise ArtifactError(f"not a monocoat wheel filename: {wheel.name}")
    return parts[1]


def _run(
    argv: list[str] | tuple[str, ...],
    env: dict[str, str],
    *,
    cwd: Path | None = None,
    timeout: int = 180,
) -> CommandResult:
    """Run one command with an explicit environment and captured output."""
    try:
        proc = subprocess.run(
            [str(arg) for arg in argv],
            cwd=cwd,
            env=env,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise CheckFailure(
            f"could not run {shlex.join([str(arg) for arg in argv])}: {exc}"
        ) from exc
    return CommandResult(
        tuple(str(arg) for arg in argv),
        proc.returncode,
        proc.stdout,
        proc.stderr,
    )


def _proc_detail(proc: CommandResult) -> str:
    """A one-line diagnostic with the failing command and its output."""
    command = shlex.join(proc.argv)
    return (
        f"command={command!r} exit={proc.code} "
        f"stdout={proc.stdout!r} stderr={proc.stderr!r}"
    )


def _no_traceback(proc: CommandResult) -> None:
    if NO_TRACEBACK in proc.stdout or NO_TRACEBACK in proc.stderr:
        raise CheckFailure(f"traceback from command; {_proc_detail(proc)}")


def _expect(
    proc: CommandResult,
    expected: int | tuple[int, ...] | set[int] | frozenset[int],
    what: str,
) -> CommandResult:
    _no_traceback(proc)
    allowed = {expected} if isinstance(expected, int) else set(expected)
    if proc.code not in allowed:
        wanted = ", ".join(str(code) for code in sorted(allowed))
        raise CheckFailure(f"{what} exited {proc.code}, expected {wanted}; {_proc_detail(proc)}")
    return proc


def _json_doc(proc: CommandResult, what: str) -> dict:
    """Parse stdout as one JSON object, preserving the full output on failure."""
    _no_traceback(proc)
    try:
        doc = json.loads(proc.stdout)
    except json.JSONDecodeError as exc:
        raise CheckFailure(
            f"{what} did not print valid JSON ({exc}); {_proc_detail(proc)}"
        ) from exc
    if not isinstance(doc, dict):
        raise CheckFailure(f"{what} printed JSON that is not an object; {_proc_detail(proc)}")
    return doc


def _one_line(text: str) -> str:
    return text.replace("\r", "\\r").replace("\n", "\\n")


def _venv(
    base: Path,
    env: dict[str, str],
    wheel: Path,
    *,
    system_site: bool = False,
) -> tuple[Path, Path]:
    """Create a venv, install the wheel offline, and return (python, cli)."""
    venv = base / "venv"
    argv = [str(sys.executable), "-m", "venv"]
    if system_site:
        argv.append("--system-site-packages")
    argv.append(str(venv))
    _expect(_run(argv, env), 0, f"create venv at {venv}")
    python = venv / "bin" / "python"
    cli = venv / "bin" / "monocoat"
    install = _run(
        [
            str(python),
            "-m",
            "pip",
            "install",
            "--no-index",
            "--no-deps",
            str(wheel),
        ],
        env,
    )
    _expect(install, 0, f"install {wheel.name}")
    if not python.is_file() or not cli.is_file():
        raise CheckFailure(f"venv install did not create {python} and {cli}")
    return python, cli


class ReleaseSmoke:
    """Run the release scenarios, preserving each result independently."""

    def __init__(self, dist: Path, *, keep: bool = False) -> None:
        self.wheel, self.sdist = find_artifacts(dist)
        self.expected_version = wheel_version(self.wheel)
        self.keep = keep
        self.root = Path(tempfile.mkdtemp(prefix="monocoat-release-smoke-"))
        self.checks: list[Check] = []

    def run(self) -> list[Check]:
        scenarios = (
            ("wheel-cli", self._wheel_cli),
            ("first-run", self._first_run),
            ("migration", self._migration),
            ("missing-tools", self._missing_tools),
            ("sdist", self._sdist),
            ("clone-no-install", self._clone_no_install),
            ("settings-import", self._settings_import),
        )
        for name, scenario in scenarios:
            base = self.root / name
            try:
                detail = scenario(base)
            except SkipCheck as exc:
                self.checks.append(Check(name, "SKIP", _one_line(str(exc))))
            except CheckFailure as exc:
                self.checks.append(Check(name, "FAIL", _one_line(str(exc))))
            except Exception as exc:  # keep one bad scenario from hiding the rest
                self.checks.append(
                    Check(
                        name,
                        "FAIL",
                        _one_line(f"{type(exc).__name__}: {exc}"),
                    )
                )
            else:
                self.checks.append(Check(name, "PASS", _one_line(detail)))
        return self.checks

    def _wheel_cli(self, base: Path) -> str:
        env = _prepare_env(base)
        python, cli = _venv(base, env, self.wheel)

        version = _run([cli, "--version"], env)
        _expect(version, 0, "monocoat --version")
        expected = f"monocoat {self.expected_version}\n"
        if version.stdout != expected:
            raise CheckFailure(
                f"monocoat --version printed {version.stdout!r}, expected {expected!r}; "
                f"{_proc_detail(version)}"
            )

        _expect(_run([cli, "--help"], env), 0, "monocoat --help")

        bare = _run([cli], env)
        _expect(bare, 2, "bare monocoat without a display")
        if "usage: monocoat" not in bare.stdout:
            raise CheckFailure(f"bare monocoat did not print help; {_proc_detail(bare)}")

        for module in ("rich", "textual"):
            imported = _run([python, "-c", f"import {module}"], env)
            if imported.code == 0:
                raise CheckFailure(f"{module} is importable in the zero-dependency wheel")

        gui = _run([python, "-c", "import monocoat.gui"], env)
        _expect(gui, 0, "import monocoat.gui without gi")

        init = _run([cli, "init", "--json"], env)
        _expect(init, 0, "wheel monocoat init before doctor")
        _json_doc(init, "wheel monocoat init --json")
        doctor = _run([cli, "doctor", "--json"], env)
        _expect(doctor, {0, 1, 5}, "wheel monocoat doctor --json")
        _json_doc(doctor, "wheel monocoat doctor --json")
        return (
            f"version={self.expected_version}; help/bare/zero-dep imports ok; "
            f"doctor rc={doctor.code}"
        )

    def _first_run(self, base: Path) -> str:
        env = _prepare_env(base)
        _python, cli = _venv(base, env, self.wheel)

        missing = _run([cli, "apply", "--dry-run"], env)
        _expect(missing, 3, "first apply --dry-run without a config")
        if "monocoat init" not in missing.stderr:
            raise CheckFailure(
                f"missing-config hint does not name `monocoat init`; {_proc_detail(missing)}"
            )

        init = _run([cli, "init", "--json"], env)
        _expect(init, 0, "monocoat init")
        init_doc = _json_doc(init, "monocoat init --json")
        if init_doc.get("name") != "default":
            raise CheckFailure(f"init chose preset {init_doc.get('name')!r}, expected 'default'")

        listed = _run([cli, "list"], env)
        _expect(listed, 0, "monocoat list after init")
        if "default" not in listed.stdout:
            raise CheckFailure(f"monocoat list did not name the initialized preset; "
                               f"{_proc_detail(listed)}")

        dry = _run([cli, "apply", "--dry-run", "--json"], env)
        _expect(dry, {0, 1, 5}, "first apply --dry-run --json")
        _json_doc(dry, "first apply --dry-run --json")
        return f"init ok; list rc=0; first dry-run rc={dry.code}"

    def _migration(self, base: Path) -> str:
        env = _prepare_env(base)
        _python, cli = _venv(base, env, self.wheel)
        config_home = Path(env["XDG_CONFIG_HOME"])
        legacy = config_home / "theme" / "theme.toml"
        legacy.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(ROOT / "examples" / "theme.toml", legacy)

        dry = _run([cli, "migrate", "--dry-run", "--json"], env)
        _expect(dry, 0, "monocoat migrate --dry-run --json")
        dry_doc = _json_doc(dry, "monocoat migrate --dry-run --json")
        if dry_doc.get("exit") != 0 or dry_doc.get("status") != "planned":
            raise CheckFailure(f"unexpected migration dry-run document: {dry_doc!r}")
        active = config_home / "monocoat" / "monocoat.toml"
        if active.exists() or active.is_symlink():
            raise CheckFailure("migration dry-run wrote the active config")

        migrated = _run([cli, "migrate", "--json"], env)
        _expect(migrated, 0, "monocoat migrate --json")
        doc = _json_doc(migrated, "monocoat migrate --json")
        if doc.get("exit") != 0 or doc.get("status") != "migrated":
            raise CheckFailure(f"unexpected migration document: {doc!r}")
        if not active.is_symlink() or not active.exists():
            raise CheckFailure(f"migration did not create a working {active}")
        return f"dry-run planned; migrate rc=0; active={active.name}"

    def _missing_tools(self, base: Path) -> str:
        env = _prepare_env(base)
        _python, cli = _venv(base, env, self.wheel)
        # This host may have the optional tools in /usr/bin. Use only the
        # interpreter shim and an empty directory so doctor observes absence.
        empty = base / "empty-bin"
        empty.mkdir()
        interp_bin = Path(env["PATH"].split(":", 1)[0])
        env["PATH"] = f"{interp_bin}:{empty}"
        for tool in OPTIONAL_TOOLS:
            if shutil.which(tool, path=env["PATH"]):
                raise CheckFailure(f"missing-tools PATH unexpectedly contains {tool}")

        init = _run([cli, "init", "--json"], env)
        _expect(init, 0, "monocoat init for missing-tools scenario")
        _json_doc(init, "monocoat init --json")

        doctor = _run([cli, "doctor", "--json"], env)
        _expect(doctor, {0, 1, 3, 5}, "monocoat doctor --json without optional tools")
        doc = _json_doc(doctor, "monocoat doctor --json without optional tools")
        missing = {
            row.get("tool")
            for row in doc.get("tools", [])
            if isinstance(row, dict) and row.get("status") != "ok"
        }
        absent = [tool for tool in OPTIONAL_TOOLS if tool not in missing]
        if absent:
            raise CheckFailure(
                "doctor did not list these optional tools as missing: "
                + ", ".join(absent)
                + f"; tools={doc.get('tools')!r}"
            )

        dry = _run([cli, "apply", "--dry-run"], env)
        _expect(dry, {0, 1, 3, 5}, "apply --dry-run without optional tools")
        return (
            f"doctor rc={doctor.code}; all {len(OPTIONAL_TOOLS)} optional tools missing; "
            f"dry-run rc={dry.code}"
        )

    def _sdist(self, _base: Path) -> str:
        wheel_modules: set[str] = set()
        with zipfile.ZipFile(self.wheel) as wheel:
            for name in wheel.namelist():
                path = PurePosixPath(name)
                if path.parts and path.parts[0] == "monocoat" and path.suffix == ".py":
                    wheel_modules.add(str(path.relative_to("monocoat")))

        with tarfile.open(self.sdist, "r:gz") as archive:
            names = {
                str(PurePosixPath(member.name))
                for member in archive.getmembers()
                if member.isfile()
            }
            roots = {PurePosixPath(name).parts[0] for name in names if PurePosixPath(name).parts}
            if len(roots) != 1:
                raise CheckFailure(f"sdist has unexpected top-level entries: {sorted(roots)}")
            root = next(iter(roots))
            required = {
                f"{root}/src/monocoat/__init__.py",
                f"{root}/LICENSE",
                f"{root}/README.md",
                f"{root}/pyproject.toml",
            }
            absent = sorted(required - names)
            if absent:
                raise CheckFailure(f"sdist lacks required files: {', '.join(absent)}")

            prefix = f"{root}/src/monocoat/"
            sdist_modules = {
                name[len(prefix):]
                for name in names
                if name.startswith(prefix) and name.endswith(".py")
            }
            not_in_sdist = sorted(wheel_modules - sdist_modules)
            if not_in_sdist:
                raise CheckFailure(
                    "wheel modules missing from sdist: " + ", ".join(not_in_sdist)
                )
            init_name = f"{root}/src/monocoat/__init__.py"
            member = archive.getmember(init_name)
            source_file = archive.extractfile(member)
            if source_file is None:
                raise CheckFailure(f"cannot read {init_name} from the sdist")
            source = source_file.read().decode("utf-8")

        match = re.search(r'^__version__\s*=\s*["\']([^"\']+)["\']', source, re.MULTILINE)
        if match is None:
            raise CheckFailure("sdist src/monocoat/__init__.py has no __version__")
        if match.group(1) != self.expected_version:
            raise CheckFailure(
                f"sdist version {match.group(1)!r} does not match wheel "
                f"version {self.expected_version!r}"
            )
        return (
            f"{len(wheel_modules)} wheel modules present under src/monocoat; "
            f"version={self.expected_version}"
        )

    def _clone_no_install(self, base: Path) -> str:
        env = _prepare_env(base)
        clone = base / "tmp" / "clone"
        clone.parent.mkdir(parents=True, exist_ok=True)
        cloned = _run(["git", "clone", "--quiet", str(ROOT), str(clone)], env)
        _expect(cloned, 0, "git clone local checkout")
        version = _run(["python3", str(clone / "bin" / "monocoat"), "--version"], env)
        _expect(version, 0, "clone bin/monocoat --version without installing")
        expected = f"monocoat {self.expected_version}\n"
        if version.stdout != expected:
            raise CheckFailure(
                f"clone printed {version.stdout!r}, expected {expected!r}; "
                f"{_proc_detail(version)}"
            )
        return f"local clone runs from its shim at version={self.expected_version}"

    def _settings_import(self, base: Path) -> str:
        env = _prepare_env(base)
        system_venv = base / "probe-venv"
        created = _run(
            [str(sys.executable), "-m", "venv", "--system-site-packages", str(system_venv)],
            env,
        )
        _expect(created, 0, "create system-site-packages venv")
        probe_python = system_venv / "bin" / "python"
        gi = _run(
            [
                str(probe_python),
                "-c",
                "import gi; gi.require_version('Gtk', '3.0')",
            ],
            env,
        )
        if gi.code != 0:
            raise SkipCheck("system python has no PyGObject GTK 3")
        _no_traceback(gi)

        # The final environment is a fresh one; the probe venv above only
        # decided whether this host can exercise the Settings entry point.
        shutil.rmtree(system_venv)
        _python, cli = _venv(base, env, self.wheel, system_site=True)
        settings = _run([cli, "settings"], env)
        _no_traceback(settings)
        if settings.code == 0:
            raise CheckFailure("monocoat settings with no DISPLAY exited 0")
        message = (settings.stderr + settings.stdout).strip()
        if not message:
            raise CheckFailure(f"settings failure had no message; {_proc_detail(settings)}")
        return f"GTK importable; settings rc={settings.code} with a clear headless message"


def _print_checks(checks: list[Check]) -> None:
    for check in checks:
        print(f"{check.status} {check.name}: {check.detail}")
    passed = sum(check.status == "PASS" for check in checks)
    failed = sum(check.status == "FAIL" for check in checks)
    skipped = sum(check.status == "SKIP" for check in checks)
    print(
        f"summary: {len(checks)} checks, {passed} passed, "
        f"{failed} failed, {skipped} skipped"
    )


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        smoke = ReleaseSmoke(args.dist, keep=args.keep)
    except ArtifactError as exc:
        print(f"release-smoke: {exc}", file=sys.stderr)
        return 2

    try:
        checks = smoke.run()
        failed = sum(check.status == "FAIL" for check in checks)
        exit_code = 1 if failed else 0
        if args.json:
            print(
                json.dumps(
                    {
                        "checks": [
                            {
                                "name": check.name,
                                "status": check.status,
                                "detail": check.detail,
                            }
                            for check in checks
                        ],
                        "failed": failed,
                        "exit": exit_code,
                    },
                    indent=2,
                    sort_keys=True,
                )
            )
            if args.keep:
                print(f"kept: {smoke.root}", file=sys.stderr)
        else:
            _print_checks(checks)
            if args.keep:
                print(f"kept: {smoke.root}")
        return exit_code
    finally:
        if not args.keep:
            shutil.rmtree(smoke.root, ignore_errors=True)


if __name__ == "__main__":
    raise SystemExit(main())
