"""PM-owned uv commands use the installed interpreter, never host discovery."""
from __future__ import annotations

import importlib
import os
from pathlib import Path
import shutil
import subprocess
import sys

import pytest

from pm._uv import _toolchain
from pm.lock import Facts, Lockfile
from pm.package import InstallError
from pm.packages import Python, Uv
from pm.store import current_target




@pytest.fixture
def installed_uv(tmp_path, monkeypatch):
    import pm.paths as paths
    import pm.registry as registry

    uv = shutil.which("uv")
    assert uv, "the interpreter selection contract requires real uv"
    store = tmp_path / "store"
    lock = Lockfile(tmp_path / "lock.json")
    target = current_target()
    digest = "1" * 64
    entry = store / "uv"
    entry.mkdir(parents=True)
    binary = entry / ("uv.exe" if os.name == "nt" else "uv")
    shutil.copy2(uv, binary)
    lock.set_pin("uv", "test", {target: {"url": "https://test.invalid/uv", "sha256": digest}})
    lock.set_pin("python", "test", {target: {"url": "https://test.invalid/python", "sha256": digest}})
    lock.save()
    facts = Facts(store / "facts.json")
    facts.record("uv", "test", entry.name, {}, store, target=target, artifacts=[digest])
    monkeypatch.setattr(paths, "lockfile_path", lambda: lock.path)
    monkeypatch.setattr(paths, "store_root", lambda: store)
    monkeypatch.setattr(paths, "writable_store_root", lambda: store)
    monkeypatch.setitem(registry._packages, "uv", Uv())
    return tmp_path, binary, facts, target, digest


def test_internal_tooling_cannot_escape_package_queries(installed_uv):
    from pm import env_for, installed_package

    _, binary, facts, target, digest = installed_uv
    assert "PATH" not in Uv().env(binary.parent, target)
    # Legacy facts can contain PATH even after the package stops exporting it.
    facts.record("uv", "test", binary.parent.name, {"PATH": [str(binary.parent)]},
                 binary.parent.parent, target=target, artifacts=[digest])
    assert str(binary.parent) not in env_for("uv", "venv", base_env={"PATH": "external"})["PATH"]
    with pytest.raises(ValueError, match="internal"):
        installed_package("uv")


def test_all_uv_commands_keep_the_pm_interpreter(installed_uv, monkeypatch):
    import pm.registry as registry

    root, uv, facts, target, digest = installed_uv
    selected = root / "store" / "selected-python"
    clean = {key: value for key, value in os.environ.items() if not key.startswith("UV_")}
    clean.update({"UV_NO_CONFIG": "1", "UV_OFFLINE": "1", "UV_PYTHON_DOWNLOADS": "never"})
    subprocess.run([str(uv), "venv", "--python", sys.executable, str(selected)],
                   cwd=root, env=clean, check=True, capture_output=True, timeout=60)
    python = selected / ("Scripts/python.exe" if os.name == "nt" else "bin/python")

    class FixturePython(Python):
        binary_rel = {"win32": "Scripts/python.exe", "posix": "bin/python"}

    monkeypatch.setitem(registry._packages, "python", FixturePython())
    facts.record("python", "test", selected.name, {}, selected.parent,
                 target=target, artifacts=[digest])
    monkeypatch.setenv("UV_PYTHON", str(root / "ambient-python"))
    monkeypatch.setenv("UV_PROJECT_ENVIRONMENT", str(root / "ambient-venv"))
    before = dict(os.environ)
    from pm._uv import _toolchain
    from pm.environment import managed_environment

    assert _toolchain(realize=False) == (uv, python)
    assert dict(os.environ) == before
    project = root / "project"
    project.mkdir()
    (project / "pyproject.toml").write_text(
        '[project]\nname="pm-python-proof"\nversion="1"\nrequires-python=">=3.11"\n'
        '[tool.uv]\npackage=false\n', encoding="utf-8",
    )
    (project / ".python-version").write_text(str(root / "host-only-python"), encoding="utf-8")
    environment = managed_environment(root / "candidate", offline=True)
    assert environment.python == python
    environment.create()
    environment.lock(project)
    environment.sync(project)
    environment.check()
    result = subprocess.run([str(environment.executable), "-I", "-c", "import sys; print(sys.prefix)"],
                            capture_output=True, text=True, check=True, timeout=30)
    assert Path(result.stdout.strip()).resolve() == environment.destination.resolve()
    assert dict(os.environ) == before


def test_project_environment_replaces_generation_when_pinned_python_moves(installed_uv, tmp_path, monkeypatch):
    """An unchanged dependency pin cannot reuse a venv made by another tools store."""
    from pm import operations
    import pm.registry as registry
    from tests.pm._fixtures import _run, _wheel, stage_host_python

    root, uv, facts, target, digest = installed_uv
    store = root / "store"
    from pm.store import tree_digest
    facts.record("uv", "test", uv.parent.name, {}, store, target=target,
                 artifacts=[digest], digest=tree_digest(uv.parent))

    class FixturePython(Python):
        binary_rel = {"win32": "Scripts/python.exe", "posix": "bin/python"}

    monkeypatch.setitem(registry._packages, "python", FixturePython())
    interpreters = [stage_host_python(store / name / "bin" / "python")
                    for name in ("python-a", "python-b")]
    project = tmp_path / "project"
    project.mkdir()
    wheel = _wheel(tmp_path, "side_dep")
    (project / "pyproject.toml").write_text(
        '[project]\nname="pm-generation-proof"\nversion="1"\nrequires-python=">=3.11"\n'
        'dependencies=["side-dep==1.0"]\n[tool.uv]\npackage=false\nno-index=true\n'
        f'find-links=["{wheel.parent.as_posix()}"]\n', encoding="utf-8",
    )
    env = {key: value for key, value in os.environ.items()
           if not key.startswith(("UV_", "PYTHON")) and key != "VIRTUAL_ENV"}
    env.update(UV_CACHE_DIR=str(tmp_path / "cache"), UV_PYTHON_DOWNLOADS="never")
    _run([str(uv), "lock", "--python", str(interpreters[0])], cwd=project, env=env)
    locked = (project / "uv.lock").read_bytes()
    selection_root = tmp_path / "selection"
    chosen = []
    for interpreter in (*interpreters, interpreters[0]):
        facts.record("python", "test", interpreter.parent.parent.name, {}, store,
                     target=target, artifacts=[digest], digest=tree_digest(interpreter.parent.parent))
        selected = operations.ensure_project_environment(
            "test-environment", project, root=selection_root, explicit=True)
        chosen.append(selected)
        assert Path(_run([str(selected), "-I", "-c",
                          "import sys, side_dep; print(sys.base_prefix)"], cwd=project, env=env)) == interpreter.parent.parent
        assert (project / "uv.lock").read_bytes() == locked
    assert chosen[0] != chosen[1] != chosen[2]

    # A failed replacement must leave the previously selected environment intact.
    facts.record("python", "test", interpreters[1].parent.parent.name, {}, store,
                 target=target, artifacts=[digest], digest=tree_digest(interpreters[1].parent.parent))
    active = (selection_root / "active.json").read_bytes()
    monkeypatch.setattr(operations, "build_environment", lambda **kwargs: (_ for _ in ()).throw(RuntimeError("build failed")))
    with pytest.raises(RuntimeError, match="build failed"):
        operations.ensure_project_environment("test-environment", project, root=selection_root, explicit=True)
    assert (selection_root / "active.json").read_bytes() == active
    assert operations.environment_python("test-environment", root=selection_root) == chosen[-1]


def test_uv_refuses_discovery_when_pm_python_is_missing(installed_uv, monkeypatch):
    root, _, facts, target, digest = installed_uv
    ensure = importlib.import_module("pm.install")
    monkeypatch.setattr(ensure, "lazy_installs_allowed", lambda: False)
    monkeypatch.setenv("UV_PYTHON", str(root / "ambient-python"))
    assert _toolchain(realize=False) is None
    assert not (root / "store" / "python-test").exists()
    with pytest.raises(InstallError, match="python"):
        _toolchain()

    entry = root / "store" / "missing-binary"
    entry.mkdir()
    facts.record("python", "test", entry.name, {}, entry.parent,
                 target=target, artifacts=[digest])
    recorded = facts.path.read_bytes()
    assert _toolchain(realize=False) is None
    with pytest.raises(InstallError, match="lazy installs are disabled: python"):
        _toolchain()
    assert facts.path.read_bytes() == recorded
    assert not list(entry.iterdir())


@pytest.mark.platforms("windows")
def test_bundled_uv_uses_a_verified_writable_python_without_changing_runtime(installed_uv, monkeypatch):
    import pm.paths as paths
    import pm.registry as registry
    from pm.store import Store, tree_digest

    root, uv_binary, facts, target, digest = installed_uv
    ensure = importlib.import_module("pm.install")
    shipped = uv_binary.parent.parent
    writable = root / "writable-tools"
    monkeypatch.setattr(paths, "writable_store_root", lambda: writable)
    (shipped.parent / "manifest.json").write_text("{}", encoding="utf-8")

    class FixturePython(Python):
        def verify(self, entry, target):
            return "" if self.binary(entry, target).read_bytes() == b"pinned interpreter" else "damaged Python"

    class FixtureUv(Uv):
        emulated_arch_targets = {target}

    monkeypatch.setitem(registry._packages, "uv", FixtureUv())
    facts.record("uv", "test", uv_binary.parent.name, {}, shipped,
                 target=target, artifacts=[digest], digest=tree_digest(uv_binary.parent))
    python = FixturePython()
    monkeypatch.setitem(registry._packages, "python", python)
    entry = shipped / python.store_entry("test", target)
    entry.mkdir()
    binary = entry / "python.exe"
    binary.write_bytes(b"pinned interpreter")
    (entry / "python.dll").write_bytes(b"pinned runtime")
    facts.record("python", "test", entry.name, python.env(entry, target), shipped,
                 target=target, artifacts=[digest], digest=tree_digest(entry))
    before = facts.path.read_bytes()
    shipped_digest = tree_digest(entry)
    monkeypatch.setattr(ensure, "lazy_installs_allowed", lambda: False)

    def no_download(*args, **kwargs):
        raise AssertionError("the verified Python must be copied without downloading")

    monkeypatch.setattr(Store, "fetch_many", no_download)
    assert _toolchain(realize=False) is None
    assert not writable.exists()
    with pytest.raises(InstallError, match="lazy installs are disabled"):
        _toolchain()
    assert not writable.exists()

    resolved_uv, python = _toolchain(explicit=True)
    copied = writable / entry.name
    assert resolved_uv == uv_binary
    assert python == copied / "python.exe"
    assert tree_digest(copied) == shipped_digest
    copied_fact = Facts(writable / "facts.json").get("python")
    assert copied_fact["digest"] == shipped_digest
    assert copied_fact["artifacts"] == [digest]
    assert copied_fact["target"] == target
    assert ensure.installed_package("python").binary == binary
    assert str(copied) not in ensure.env_for("python")["PATH"]

    def no_copy(*args, **kwargs):
        raise AssertionError("a matching copy must be reused")

    with monkeypatch.context() as reuse:
        reuse.setattr(shutil, "copytree", no_copy)
        assert _toolchain(realize=False)[1] == python
        assert _toolchain()[1] == python

    assert facts.path.read_bytes() == before
    assert tree_digest(entry) == shipped_digest


@pytest.mark.parametrize("damage", [None, "source", "copy", "publication"])
def test_copy_failure_preserves_previous_python(installed_uv, monkeypatch, damage):
    import pm.registry as registry
    from pm.store import Store, tree_digest

    root, _, facts, target, digest = installed_uv
    ensure = importlib.import_module("pm.install")
    shipped = facts.path.parent
    writable = root / "writable-tools"
    writable.mkdir()
    python = Python()
    monkeypatch.setattr(python, "verify", lambda *_: "")
    monkeypatch.setitem(registry._packages, "python", python)
    entry_name = python.store_entry("test", target)
    source = shipped / entry_name
    source.mkdir()
    binary_rel = python.binary(source, target).relative_to(source)
    (source / binary_rel).parent.mkdir(parents=True, exist_ok=True)
    (source / binary_rel).write_bytes(b"new interpreter")
    facts.record("python", "test", entry_name, {}, shipped,
                 target=target, artifacts=[digest], digest=tree_digest(source))
    previous = writable / entry_name
    previous.mkdir()
    (previous / binary_rel).parent.mkdir(parents=True, exist_ok=True)
    (previous / binary_rel).write_bytes(b"previous interpreter")
    previous_facts = Facts(writable / "facts.json")
    previous_facts.record("python", "previous", entry_name, {}, writable,
                          target=target, artifacts=[digest], digest=tree_digest(previous))
    before = previous_facts.path.read_bytes()
    copytree = shutil.copytree

    if damage == "source":
        (source / binary_rel).write_bytes(b"damaged source")
    elif damage == "copy":
        def damaged_copy(src, dest, *args, **kwargs):
            result = copytree(src, dest, *args, **kwargs)
            if Path(src) == source:
                (dest / binary_rel).write_bytes(b"damaged copy")
            return result
        monkeypatch.setattr(shutil, "copytree", damaged_copy)
    elif damage == "publication":
        def failed_publication(*args, **kwargs):
            raise OSError("publication refused")
        monkeypatch.setattr(Store, "publish", failed_publication)

    if damage is None:
        monkeypatch.setattr(Store, "fetch_many", lambda *args, **kwargs: pytest.fail("copy downloaded bytes"))
        ensure._install(python, ensure._lockfile(), previous_facts, Store(writable), target,
                        copy_from=(facts, Store(shipped)))
        assert tree_digest(previous) == tree_digest(source)
        assert previous_facts.get("python")["digest"] == facts.get("python")["digest"]
    else:
        with pytest.raises(InstallError, match="verification|copied bytes|publication refused"):
            ensure._install(python, ensure._lockfile(), previous_facts, Store(writable), target,
                            copy_from=(facts, Store(shipped)))
        assert (previous / binary_rel).read_bytes() == b"previous interpreter"
        assert previous_facts.path.read_bytes() == before
