"""Offline behavior contracts for PM's caller-owned build operations."""
from __future__ import annotations

import json
import os
from pathlib import Path
import sys

import pytest

from pm.package import InstallError
from tests.pm._fixtures import (
    _run,
    _wheel,
    build_worker as build_worker,
    client as client,
    isolated_python as isolated_python,
    served as served,
)


def test_stage_tools_copies_verified_closure_without_acquiring_or_live_state(tmp_path, client, monkeypatch, served):
    import importlib.util
    import pm
    from pm.lock import Facts, Lockfile
    from pm.registry import _packages
    from pm.store import current_target, tree_digest

    target = current_target()
    source = tmp_path / "canonical"
    source.mkdir()
    lock = Lockfile(tmp_path / "lock.json")
    facts = Facts(source / "facts.json")
    definition = tmp_path / "copy_fixture.py"
    definition.write_text(
        "from pm.package import Package\n"
        "class CopyLeaf(Package):\n"
        "    name = 'copy-leaf'\n"
        "    def env(self, entry, target): return {'COPY_ROOT': str(entry)}\n"
        "class CopyTool(CopyLeaf):\n"
        "    name = 'copy-tool'\n"
        "    deps = ('copy-leaf',)\n", encoding="utf-8")
    spec = importlib.util.spec_from_file_location("copy_fixture", definition)
    module = importlib.util.module_from_spec(spec)
    monkeypatch.setitem(sys.modules, "copy_fixture", module)
    spec.loader.exec_module(module)
    for package in (module.CopyLeaf(), module.CopyTool()):
        monkeypatch.setitem(_packages, package.name, package)
        lock.set_pin(package.name, "1.0", {target: {"url": "https://invalid.test/tool.tgz", "sha256": "a" * 64}})
        entry = source / package.store_entry("1.0", target)
        entry.mkdir()
        (entry / "data").write_text(package.name, encoding="utf-8")
        facts.record(package.name, "1.0", entry.name, package.env(entry, target), source,
                     target=target, artifacts=["a" * 64], digest=tree_digest(entry))
    lock.save()
    from tests.pm._fixtures import make_tar
    import shutil
    directory, base = served
    for name in ("copy-leaf", "copy-tool"):
        archive, sha = make_tar(directory, name + ".tgz", {"data": name})
        lock.set_pin(name, "1.0", {target: {"url": base + "/" + archive, "sha256": sha}})
    lock.save()
    shutil.rmtree(source)
    assert pm.prepare_tools(["copy-tool"], out=source, target=target, cache=tmp_path / "python-cache") == source
    facts = Facts(source / "facts.json")
    before = (source / "facts.json").read_bytes()
    home_before = sorted(str(path) for path in (tmp_path / "home").rglob("*"))
    destination = tmp_path / "payload-tools"
    result = pm.stage_tools(["copy-tool"], source_store=source, out=destination, target=target)
    assert result == destination
    # An old output may have been populated with hardlinks. A fresh staging
    # request must establish independent bytes, not reuse those current facts.
    tool_entry = facts.get("copy-tool")["entry"]
    linked_copy = destination / tool_entry / "data"
    linked_copy.unlink()
    os.link(source / tool_entry / "data", linked_copy)
    pm.stage_tools(["copy-tool"], source_store=source, out=destination, target=target)
    staged = Facts(destination / "facts.json")
    for name in ("copy-leaf", "copy-tool"):
        entry = facts.get(name)["entry"]
        copied, original = destination / entry / "data", source / entry / "data"
        assert copied.read_bytes() == original.read_bytes()
        assert not copied.samefile(original)
        copied.write_text("signed output", encoding="utf-8")
        assert original.read_text() == name
        assert staged.get(name)["digest"] == facts.get(name)["digest"]
        assert staged.env_for(name, destination)["COPY_ROOT"] == str(destination / entry)
    assert (source / "facts.json").read_bytes() == before
    assert sorted(str(path) for path in (tmp_path / "home").rglob("*")) == home_before
    assert not (tmp_path / "home/config.yaml").exists()
    assert not (tmp_path / "store/facts.json").exists()

    # Every call must admit the source, even if destination facts are warm.
    original = source / facts.get("copy-tool")["entry"] / "data"
    original.write_text("corrupt", encoding="utf-8")
    with pytest.raises(InstallError, match="source failed verification"):
        pm.stage_tools(["copy-tool"], source_store=source, out=destination, target=target)
    original.write_text("copy-tool", encoding="utf-8")
    lock.set_pin("copy-tool", "1.0", {target: {"url": "https://invalid.test/repin", "sha256": "b" * 64}})
    lock.save()
    with pytest.raises(InstallError, match="source failed verification"):
        pm.stage_tools(["copy-tool"], source_store=source, out=destination, target=target)
    lock.set_pin("copy-tool", "1.0", {target: {"url": "https://invalid.test/tool.tgz", "sha256": facts.get("copy-tool")["artifacts"][0]}})
    lock.save()
    # A locally recorded link must not turn an independent copy into a shared
    # mutable file in the cache or another caller's tree.
    external = tmp_path / "external"
    external.write_text("shared", encoding="utf-8")
    original.unlink()
    original.symlink_to(external)
    package = module.CopyTool()
    entry = original.parent
    facts.record(package.name, "1.0", entry.name, package.env(entry, target), source,
                 target=target, artifacts=facts.get("copy-tool")["artifacts"], digest=tree_digest(entry))
    with pytest.raises(InstallError, match="source.*verification|escapes"):
        pm.stage_tools(["copy-tool"], source_store=source, out=tmp_path / "other-output", target=target)


@pytest.mark.parametrize("damage", [None, "entry", "env", "bytes", "pin", "binary"])
def test_verified_tools_admits_only_locked_entries_without_execution(tmp_path, monkeypatch, damage):
    import subprocess
    from pm import build_operations
    from pm.lock import Facts, Lockfile
    from pm.registry import get_package
    from pm.store import current_target, tree_digest

    target, store = current_target(), tmp_path / "tools"
    lock = Lockfile(tmp_path / "lock.json")
    package = get_package("python")
    entry = store / package.store_entry("1.0", target)
    binary = package.binary(entry, target)
    assert binary is not None
    binary.parent.mkdir(parents=True)
    binary.write_bytes(b"already verified at publication")
    lock.set_pin("python", "1.0", {target: {"url": "https://invalid.test/python", "sha256": "a" * 64}})
    facts = Facts(store / "facts.json")
    env = package.env(entry, target)
    if damage == "entry":
        renamed = entry.with_name("noncanonical")
        entry.rename(renamed)
        entry = renamed
        env = package.env(entry, target)
    if damage == "env":
        env["PATH"] = [str(tmp_path / "unverified")]
    if damage == "binary":
        binary.unlink()
    facts.record("python", "1.0", entry.name, env, store, target=target,
                 artifacts=["a" * 64], digest=tree_digest(entry))
    if damage == "bytes":
        binary.write_bytes(b"changed")
    if damage == "pin":
        lock.set_pin("python", "1.0", {target: {"url": "https://invalid.test/python", "sha256": "b" * 64}})
    before = {p.relative_to(store): p.read_bytes() for p in store.rglob("*") if p.is_file()}
    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: pytest.fail("read-only admission executed a process"))
    if damage:
        with pytest.raises(InstallError, match="source failed verification"):
            build_operations.verified_tools(["python"], source_store=store, target=target, lock=lock)
    else:
        selected = build_operations.verified_tools(["python"], source_store=store, target=target, lock=lock)
        assert selected.entries["python"].path == entry
        assert selected.entries["python"].binary == binary
        assert selected.environment({"PATH": "inherited"})["PATH"] == str(binary.parent) + os.pathsep + "inherited"
    assert {p.relative_to(store): p.read_bytes() for p in store.rglob("*") if p.is_file()} == before


@pytest.fixture
def build_tools(tmp_path, monkeypatch, build_worker):
    env = {key: value for key, value in os.environ.items()
           if not key.startswith(("UV_", "PYTHON")) and key != "VIRTUAL_ENV"}
    home = tmp_path / "home"
    home.mkdir()
    env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(home / ".hermes"))
    monkeypatch.setenv("HERMES_HOME", env["HERMES_HOME"])
    monkeypatch.setattr(Path, "home", lambda: home)
    return env


@pytest.fixture
def locked_source(tmp_path, build_tools):
    from pm import lock_project

    wheels = tmp_path / "wheelhouse"
    wheels.mkdir()
    for name in ("base_dep", "chosen_dep", "dev_dep"):
        _wheel(wheels, name)
    source = tmp_path / "source with spaces"
    source.mkdir()
    (source / "pyproject.toml").write_text(
        '[project]\nname="export-root"\nversion="1"\nrequires-python=">=3.11"\n'
        'dependencies=["base-dep==1.0"]\n'
        '[project.optional-dependencies]\nchosen=["chosen-dep==1.0; sys_platform == \'win32\'"]\n'
        '[dependency-groups]\ndev=["dev-dep==1.0"]\n'
        '[tool.uv]\npackage=false\nno-index=true\n'
        f'find-links=[{json.dumps(wheels.as_posix())}]\n', encoding="utf-8",
    )
    lock_project(source, python=Path(sys.executable), cache=tmp_path / "cache",
                 env=build_tools, offline=True, explicit=True)
    return source


@pytest.mark.parametrize("lock_state", ["current", "stale", "missing"])
def test_check_lock_never_changes_source(locked_source, tmp_path, build_tools, lock_state):
    from pm import check_project_lock

    source = locked_source
    if lock_state == "stale":
        manifest = source / "pyproject.toml"
        manifest.write_text(manifest.read_text().replace('version="1"', 'version="2"'), encoding="utf-8")
    elif lock_state == "missing":
        (source / "uv.lock").unlink()
    before = {p.relative_to(source): p.read_bytes() for p in source.rglob("*") if p.is_file()}
    if lock_state == "current":
        check_project_lock(source, python=Path(sys.executable), cache=tmp_path / "cache",
                           env=build_tools, offline=True, explicit=True)
    else:
        with pytest.raises(InstallError):
            check_project_lock(source, python=Path(sys.executable), cache=tmp_path / "cache",
                               env=build_tools, offline=True, explicit=True)
    assert {p.relative_to(source): p.read_bytes() for p in source.rglob("*") if p.is_file()} == before
    assert not (source / ".venv").exists()


def test_frozen_export_keeps_markers_and_excludes_build_metadata(locked_source, tmp_path, build_tools):
    from packaging.requirements import Requirement
    from pm import export_requirements

    source = locked_source
    # Frozen export must use yesterday's resolution, not refresh today's edits.
    manifest = source / "pyproject.toml"
    manifest.write_text(manifest.read_text().replace("base-dep==1.0", "base-dep==2.0"), encoding="utf-8")
    before = {p.name: p.read_bytes() for p in source.iterdir() if p.is_file()}
    out = tmp_path / "export directory" / "requirements.txt"
    export_requirements(source, out, extras=["chosen", "chosen"], python=Path(sys.executable),
                        cache=tmp_path / "cache", env=build_tools, explicit=True)
    text = out.read_text(encoding="utf-8")
    requirements = {r.name: r for r in map(Requirement, text.splitlines())}
    assert set(requirements) == {"base-dep", "chosen-dep"}
    assert str(requirements["base-dep"].specifier) == "==1.0"
    marker = requirements["chosen-dep"].marker
    assert marker is not None
    assert marker.evaluate({"sys_platform": "win32"})
    assert not marker.evaluate({"sys_platform": "linux"})
    assert "#" not in text and "--hash" not in text
    assert {p.name: p.read_bytes() for p in source.iterdir() if p.is_file()} == before
    assert not (source / ".venv").exists()


def test_frozen_export_preserves_git_commit_pin(locked_source, tmp_path, build_tools):
    from packaging.requirements import Requirement
    from pm import export_requirements, lock_project

    dependency = tmp_path / "git-dependency"
    dependency.mkdir()
    (dependency / "pyproject.toml").write_text(
        '[project]\nname="git-dep"\nversion="1.0"\nrequires-python=">=3.11"\n'
        'dependencies=[]\n', encoding="utf-8",
    )
    _run(["git", "init", "--quiet"], cwd=dependency, env=build_tools)
    _run(["git", "add", "pyproject.toml"], cwd=dependency, env=build_tools)
    _run(["git", "-c", "user.name=PM test", "-c", "user.email=pm@example.invalid",
          "-c", "commit.gpgsign=false", "-c", f"core.hooksPath={tmp_path / 'empty-hooks'}",
          "commit", "--quiet", "-m", "fixture"], cwd=dependency, env=build_tools)
    revision = _run(["git", "rev-parse", "HEAD"], cwd=dependency, env=build_tools)
    url = f"git+{dependency.as_uri()}@{revision}"
    manifest = locked_source / "pyproject.toml"
    manifest.write_text(manifest.read_text().replace('"base-dep==1.0"', json.dumps(f"git-dep @ {url}")),
                        encoding="utf-8")
    lock_project(locked_source, python=Path(sys.executable), cache=tmp_path / "cache", env=build_tools, explicit=True)
    before = (locked_source / "uv.lock").read_bytes()
    out = tmp_path / "git-requirements.txt"
    export_requirements(locked_source, out, cache=tmp_path / "cache", env=build_tools, explicit=True)
    requirement = Requirement(out.read_text(encoding="utf-8").strip())
    assert requirement.name == "git-dep"
    assert requirement.url == url
    assert (locked_source / "uv.lock").read_bytes() == before


@pytest.mark.parametrize("sealed", [False, True])
def test_requirements_build_installs_offline_markers_and_seals_only_build_pth(tmp_path, build_tools, sealed):
    from pm import build_requirements_environment

    wheels = tmp_path / "wheels with spaces"
    wheels.mkdir()
    _wheel(wheels, "leaf_dep")
    _wheel(wheels, "app_dep", requirements=["leaf-dep==1.0"])
    out = tmp_path / "destination with spaces"
    # Ambient project and interpreter settings cannot escape the explicit build.
    (tmp_path / "uv.toml").write_text('required-version="<0.1"\n', encoding="utf-8")
    poison = tmp_path / "poison"
    env = dict(build_tools, UV_PROJECT_ENVIRONMENT=str(poison), VIRTUAL_ENV=str(poison),
               UV_PYTHON=str(poison), UV_CACHE_DIR=str(poison), UV_SYSTEM_PYTHON="true")
    before = dict(os.environ)
    executable = build_requirements_environment(
        ["app-dep==1.0; python_version >= '3'", "missing-dep==1.0; python_version < '2'"],
        out=out, python=Path(sys.executable), wheelhouse=wheels, cache=tmp_path / "cache",
        env=env, offline=True, sealed=sealed, explicit=True)
    result = json.loads(_run(
        [str(executable), "-I", "-c", "import sys, json, app_dep, leaf_dep, importlib.util; "
         "print(json.dumps([app_dep.__version__, leaf_dep.__version__, sys.base_prefix, "
         "importlib.util.find_spec('missing_dep') is None]))"], cwd=tmp_path, env=build_tools,
    ))
    assert result == ["1.0", "1.0", sys.base_prefix, True]
    assert executable.parent.parent == out
    assert bool(list(out.rglob("_virtualenv.pth"))) is not sealed
    assert not poison.exists()
    assert not Path(build_tools["HERMES_HOME"]).exists()
    assert dict(os.environ) == before


@pytest.mark.parametrize("failure", ["create", "install", "check"])
def test_failed_requirement_build_removes_only_its_candidate(tmp_path, build_tools, failure):
    import zipfile
    from pm import build_requirements_environment

    wheels = tmp_path / "wheels"
    wheels.mkdir()
    wheel = _wheel(wheels, "app_dep")
    previous = tmp_path / "previous"
    executable = build_requirements_environment(["app-dep==1.0"], out=previous, wheelhouse=wheels,
                                               env=build_tools, cache=tmp_path / "cache", offline=True, explicit=True)
    previous_cfg = (previous / "pyvenv.cfg").read_bytes()
    python = Path(sys.executable)
    if failure == "create":
        python = tmp_path / "missing-python"
    elif failure == "install":
        wheel.unlink()
    else:
        # The archive carries an extra distribution invisible to resolution.
        # pip check must reject its unsatisfied dependency after installation.
        with zipfile.ZipFile(wheel, "a") as archive:
            archive.writestr("app_dep-1.0.data/purelib/ghost-1.0.dist-info/METADATA",
                             "Metadata-Version: 2.1\nName: ghost\nVersion: 1.0\n"
                             "Requires-Dist: missing-dep==1.0\n")
    out = tmp_path / "candidate"
    with pytest.raises(InstallError, match="dependency validation" if failure == "check" else None):
        build_requirements_environment(["app-dep==1.0"], out=out, python=python, wheelhouse=wheels,
                                       env=build_tools, cache=tmp_path / "cold-cache", offline=True, explicit=True)
    assert not out.exists()
    with pytest.raises(FileExistsError, match="already exists"):
        build_requirements_environment(["app-dep==1.0"], out=previous, wheelhouse=wheels,
                                       env=build_tools, cache=tmp_path / "cache", offline=True, explicit=True)
    assert (previous / "pyvenv.cfg").read_bytes() == previous_cfg
    assert _run([str(executable), "-I", "-c", "import app_dep; print(app_dep.__version__)"],
                cwd=tmp_path, env=build_tools) == "1.0"


def test_wheelhouse_cannot_fall_back_to_index_or_build_source(tmp_path, build_tools):
    import base64
    import io
    import tarfile
    from pm import build_requirements_environment

    index = tmp_path / "index" / "leaf-dep"
    index.mkdir(parents=True)
    wheel = _wheel(index, "leaf_dep")
    (index / "index.html").write_text(f'<a href="{wheel.name}">{wheel.name}</a>', encoding="utf-8")
    wheels = tmp_path / "wheelhouse"
    wheels.mkdir()
    # A real, self-contained sdist: relaxing binary-only would make it install.
    backend = (
        "from pathlib import Path\nimport base64\n"
        "def build_wheel(wheel_directory, config_settings=None, metadata_directory=None):\n"
        f"    Path(wheel_directory, {wheel.name!r}).write_bytes(base64.b64decode("
        f"{base64.b64encode(wheel.read_bytes())!r}))\n"
        f"    return {wheel.name!r}\n"
    )
    entries = {
        "pyproject.toml": '[project]\nname="leaf-dep"\nversion="1.0"\n'
        '[build-system]\nrequires=[]\nbuild-backend="backend"\nbackend-path=["."]\n',
        "backend.py": backend,
    }
    with tarfile.open(wheels / "leaf_dep-1.0.tar.gz", "w:gz") as archive:
        for name, text in entries.items():
            data = text.encode("utf-8")
            info = tarfile.TarInfo(f"leaf_dep-1.0/{name}")
            info.size = len(data)
            archive.addfile(info, io.BytesIO(data))
    env = dict(build_tools, UV_DEFAULT_INDEX=index.parent.as_uri())
    out = tmp_path / "candidate"
    with pytest.raises(InstallError):
        build_requirements_environment(["leaf-dep==1.0"], out=out, wheelhouse=wheels,
                                       env=env, cache=tmp_path / "cache", offline=True, explicit=True)
    assert not out.exists()
    # Both alternate sources really work when there is no wheelhouse restriction.
    for name, settings in (("index", env), ("source", dict(build_tools, UV_NO_INDEX="1", UV_FIND_LINKS=str(wheels)))):
        executable = build_requirements_environment(
            ["leaf-dep==1.0"], out=tmp_path / f"from-{name}", env=settings,
            cache=tmp_path / f"{name}-cache", offline=True, explicit=True)
        assert _run([str(executable), "-I", "-c", "import leaf_dep; print(leaf_dep.__version__)"],
                    cwd=tmp_path, env=build_tools) == "1.0"


@pytest.mark.parametrize("operation", ["check", "export", "build"])
def test_ready_tools_do_not_bypass_disabled_lazy_operations(locked_source, tmp_path, build_tools, monkeypatch, operation):
    import importlib
    from pm import build_requirements_environment, check_project_lock, export_requirements

    monkeypatch.setattr(importlib.import_module("pm.install"), "lazy_installs_allowed", lambda: False)
    out = tmp_path / "blocked-output"
    before = {p.name: p.read_bytes() for p in locked_source.iterdir() if p.is_file()}
    actions = {
        "check": lambda: check_project_lock(locked_source, env=build_tools, cache=tmp_path / "cache"),
        "export": lambda: export_requirements(locked_source, out, env=build_tools, cache=tmp_path / "cache"),
        "build": lambda: build_requirements_environment(["base-dep==1.0"], out=out, env=build_tools,
                                                       cache=tmp_path / "cache", offline=True),
    }
    with pytest.raises(InstallError, match="lazy installs are disabled"):
        actions[operation]()
    assert not out.exists()
    assert {p.name: p.read_bytes() for p in locked_source.iterdir() if p.is_file()} == before
    # Passive lock validation is safe with already-ready tools and no network.
    check_project_lock(locked_source, env=build_tools, cache=tmp_path / "cache", offline=True)


def test_prune_cache_does_not_acquire_a_missing_toolchain(tmp_path, monkeypatch):
    import importlib
    import pm.paths
    from pm import prune_cache

    monkeypatch.setattr("pm.client.is_runtime", lambda: True)
    store = tmp_path / "empty-store"
    monkeypatch.setattr(pm.paths, "store_root", lambda: store)
    monkeypatch.setattr(pm.paths, "writable_store_root", lambda: store)
    monkeypatch.setattr(pm.paths, "facts_path", lambda: store / "facts.json")
    monkeypatch.setattr(importlib.import_module("pm.install"), "ensure",
                        lambda *args, **kwargs: pytest.fail("cache pruning must not acquire tools"))
    with pytest.raises(InstallError, match="pinned toolchain is unavailable"):
        prune_cache(tmp_path / "cache")
    assert not store.exists()


@pytest.mark.parametrize("ci", [False, True])
def test_prune_cache_preserves_downloaded_wheels_unless_ci(tmp_path, build_tools, ci):
    from functools import partial
    from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
    import threading
    from pm import build_requirements_environment, prune_cache

    wheels = tmp_path / "wheels"
    wheels.mkdir()
    wheel = _wheel(wheels, "cached_dep")
    server = ThreadingHTTPServer(("127.0.0.1", 0), partial(SimpleHTTPRequestHandler, directory=str(wheels)))
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    cache = tmp_path / "cache"
    requirement = f"cached-dep @ http://127.0.0.1:{server.server_port}/{wheel.name}"
    try:
        executable = build_requirements_environment(
            [requirement], out=tmp_path / "first", cache=cache, env=build_tools, explicit=True)
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)
    assert _run([str(executable), "-I", "-c", "import cached_dep; print(cached_dep.__version__)"],
                cwd=tmp_path, env=build_tools) == "1.0"
    prune_cache(cache, ci=ci)
    out = tmp_path / "second"
    if ci:
        with pytest.raises(InstallError):
            build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True, explicit=True)
        assert not out.exists()
    else:
        second = build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True, explicit=True)
        assert _run([str(second), "-I", "-c", "import cached_dep; print(cached_dep.__version__)"],
                    cwd=tmp_path, env=build_tools) == "1.0"