#!/usr/bin/env python3
"""Tests for the extracted-source verification the generated INSTALL documents.

``tools/package_release.py`` used to tell a reader to run
``sha256sum --check source-manifest.json`` after extracting the source archive.
That could never work: ``source-manifest.json`` is JSON, not a ``sha256sum``
listing, and it is not inside the source archive at all — it ships in the release
directory and inside the *binary* archive. This module pins the replacement:

- the generated ``INSTALL.md`` documents ``tools/verify_extracted_source.py``
  with the archive-derived paths, no longer prints the impossible command, and
  keeps every step's directory change inside a subshell;
- the *documented block sequence itself* is executed end to end against a real
  source archive, a real binary archive and an executable stub CLI, so cwd
  leakage and a helper that leaves a ``__pycache__`` behind both fail the test;
- the helper verifies a genuinely extracted tree against a genuinely resolved
  index — first-party files, the vendor tree, the distribution-document alias and
  the runtime corresponding-source members — and requires the packager's
  generated members;
- it fails closed on a tampered file, a missing file, an added file, an added
  directory, a symlink, a hard link, a missing generated member, a mismatched
  provenance record, a wrong manifest kind, a misnamed source root, a duplicated
  member and a wrong-typed index entry.

The archive and index here are built with the packager's own writer and the same
member set a real ``--prepare-source`` produces, then extracted with ``tar``. No
network, toolchain, container or device is touched, and a green run is evidence
about the verifier and the instructions, never about a real release.
"""

from __future__ import annotations

import contextlib
import hashlib
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile
import unittest
from pathlib import Path
from unittest import mock

ROOT = Path(__file__).resolve().parents[2]
TOOLS = ROOT / "tools"
if str(TOOLS) not in sys.path:
    sys.path.insert(0, str(TOOLS))

import package_release as packaging  # noqa: E402
import verify_extracted_source as verifier  # noqa: E402
import verify_release_local as verify  # noqa: E402

SOURCE_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu-source"
BINARY_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu"
RESERVED_VERSION = "0.2.0-preview.1"

# The helper and the first-party modules it imports from the tree: a real package
# ships them, and running the helper out of the extracted tree needs them there.
TOOLS_MODULES = (
    "package_release.py",
    "verify_release_local.py",
    "build_release_source.py",
    "verify_extracted_source.py",
)

# A withheld repository document ships as its reviewed sanitized copy, aliased to
# the archive path the original would have occupied.
DISTRIBUTION_COPY = "tools/release-docs/RELEASING.md"
DISTRIBUTION_ARCHIVE_PATH = "docs/RELEASING.md"
DISTRIBUTION_BYTES = b"# Releasing (sanitized copy)\n"
VENDOR_CONFIG_BYTES = (
    b'[source.crates-io]\nreplace-with = "vendored-sources"\n\n'
    b'[source.vendored-sources]\ndirectory = "vendor"\n'
)
VENDOR_FILES = {
    "serde/Cargo.toml": b'[package]\nname = "serde"\n',
    "serde/src/lib.rs": b"// serde\n",
}
RUNTIME_MEMBERS = {
    "runtime-sources/bundle.tar.gz": b"runtime corresponding source\n",
    "runtime-sources/bundle.json": b'{"record_kind": "runtime_source_bundle"}\n',
}
SYNTHETIC_FILES = {
    "Cargo.toml": b'[workspace]\nmembers = ["crates/x"]\n',
    "Cargo.lock": b"# lock\n",
    "crates/x/Cargo.toml": b'[package]\nname = "x"\nversion = "0.2.0-preview.1"\n',
    DISTRIBUTION_COPY: DISTRIBUTION_BYTES,
}
# An executable stub so the documented ``--version``/``validate``/``doctor``/
# ``test`` commands can actually run; it is a real program, not a skipped step.
STUB_CLI = b"""#!/bin/sh
case "$1" in
  --version) echo "ultimate-tool (extracted-source test stub)"; exit 0 ;;
  --help) echo "usage: ultimate-tool <command>"; exit 0 ;;
esac
echo "stub: $*"
exit 0
"""


def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def tools_module_bytes() -> dict[str, bytes]:
    return {f"tools/{name}": (TOOLS / name).read_bytes() for name in TOOLS_MODULES}


def tree_files(root: Path) -> set[str]:
    return {
        path.relative_to(root).as_posix()
        for path in root.rglob("*")
        if path.is_file() or path.is_symlink()
    }


class Package:
    """A prepared-style source archive, its resolved index and the extracted tree."""

    def __init__(self, work: Path, *, vendored: bool = True, runtime: bool = True,
                 extract_tree: bool = True) -> None:
        self.work = work
        self.vendored = vendored
        self.runtime = runtime
        first_party = dict(SYNTHETIC_FILES)
        first_party.update(tools_module_bytes())
        self.files = {
            "first_party": [
                {"path": name, "sha256": sha256(content), "size": len(content)}
                for name, content in sorted(first_party.items())
            ],
            "vendor": [
                {"path": name, "sha256": sha256(content), "size": len(content)}
                for name, content in sorted(VENDOR_FILES.items())
            ] if vendored else [],
        }
        aggregate = verify.aggregate_digest(self.files["first_party"])
        provenance = {
            "schema": packaging.SCHEMA,
            "record_kind": "release_source_provenance",
            "files": {"count": len(self.files["first_party"]), "aggregate_sha256": aggregate},
        }
        members: list[tuple[str, bytes, int]] = []
        for name, content in sorted(first_party.items()):
            members.append((f"{SOURCE_PREFIX}/{name}", content, packaging.ARCHIVE_MODE_FILE))
        if runtime:
            for name, content in sorted(RUNTIME_MEMBERS.items()):
                members.append((f"{SOURCE_PREFIX}/{name}", content, packaging.ARCHIVE_MODE_FILE))
        if vendored:
            for name, content in sorted(VENDOR_FILES.items()):
                members.append((f"{SOURCE_PREFIX}/vendor/{name}", content, packaging.ARCHIVE_MODE_FILE))
            members.append((f"{SOURCE_PREFIX}/vendor-sources.toml", VENDOR_CONFIG_BYTES, packaging.ARCHIVE_MODE_FILE))
        members.append((f"{SOURCE_PREFIX}/{DISTRIBUTION_ARCHIVE_PATH}", DISTRIBUTION_BYTES, packaging.ARCHIVE_MODE_FILE))
        members.append(
            (f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", json.dumps(provenance, indent=2, sort_keys=True).encode("utf-8") + b"\n", packaging.ARCHIVE_MODE_FILE)
        )
        self.archive = work / f"{SOURCE_PREFIX}.tar.gz"
        self.archive_info = packaging.write_archive(self.archive, members)
        self.manifest = {
            "schema": packaging.SCHEMA,
            "record_kind": "release_source_manifest_resolved",
            "archive": {"name": self.archive.name, "prefix": SOURCE_PREFIX, "sha256": self.archive_info["sha256"]},
            "files": self.files["first_party"],
            "aggregate_sha256": aggregate,
            "vendor": {
                "files": self.files["vendor"],
                "aggregate_sha256": verify.aggregate_digest(self.files["vendor"]),
                "file_count": len(self.files["vendor"]),
                "status": "generated_cargo_vendor" if vendored else "not_included",
            },
            "distribution_docs": [
                {"copy": DISTRIBUTION_COPY, "archive_path": DISTRIBUTION_ARCHIVE_PATH,
                 "sha256": sha256(DISTRIBUTION_BYTES), "size": len(DISTRIBUTION_BYTES)}
            ],
            "runtime_bundle": (
                {"present": True, "source_members": [
                    {"path": name, "sha256": sha256(content), "size": len(content)}
                    for name, content in sorted(RUNTIME_MEMBERS.items())
                ]} if runtime else {"present": False}
            ),
        }
        self.manifest_path = work / "source-manifest.json"
        self.manifest_bytes = json.dumps(self.manifest, indent=2, sort_keys=True).encode("utf-8") + b"\n"
        self.manifest_path.write_bytes(self.manifest_bytes)
        self.tree = self.extract(parent=work) if extract_tree else work / SOURCE_PREFIX

    def expected_tree_files(self) -> set[str]:
        names = {entry["path"] for entry in self.files["first_party"]}
        names |= {entry["archive_path"] for entry in self.manifest["distribution_docs"]}
        names |= {f"vendor/{entry['path']}" for entry in self.files["vendor"]}
        names |= {entry["path"] for entry in self.manifest["runtime_bundle"].get("source_members", [])}
        names.add("SOURCE-PROVENANCE.json")
        if self.files["vendor"]:
            names.add("vendor-sources.toml")
        return names

    def extract(self, parent: Path | None = None, name: str = SOURCE_PREFIX) -> Path:
        # Members already carry the archive prefix, so a real `tar -xzf` lays the
        # tree down under <cwd>/<prefix>; extraction goes into the parent here too.
        parent = parent or self.work
        with tarfile.open(self.archive, "r:gz") as tar:
            for member in tar.getmembers():
                target = parent / member.name
                target.parent.mkdir(parents=True, exist_ok=True)
                source = tar.extractfile(member)
                assert source is not None
                with open(target, "wb") as handle:
                    shutil.copyfileobj(source, handle)
                os.chmod(target, member.mode & 0o777)
        return parent / name


def write_binary_archive(path: Path, index_bytes: bytes) -> dict:
    members = [
        (f"{BINARY_PREFIX}/bin/ultimate-tool", STUB_CLI, packaging.ARCHIVE_MODE_EXECUTABLE),
        (f"{BINARY_PREFIX}/bin/ultimate-tool-mcp", STUB_CLI, packaging.ARCHIVE_MODE_EXECUTABLE),
        (f"{BINARY_PREFIX}/examples/first-run/c64dev.toml", b"[project]\nname = \"first-run\"\n", packaging.ARCHIVE_MODE_FILE),
        (f"{BINARY_PREFIX}/source-manifest.json", index_bytes, packaging.ARCHIVE_MODE_FILE),
    ]
    return packaging.write_archive(path, members)


def write_sha256sums(directory: Path, names: list[str]) -> None:
    lines = [f"{sha256((directory / name).read_bytes())}  {name}" for name in names]
    (directory / "SHA256SUMS").write_text("\n".join(lines) + "\n", encoding="utf-8")


def sh_blocks(document: str) -> list[str]:
    return re.findall(r"```sh\n(.*?)```", document, flags=re.DOTALL)


INSTALL_CONTEXT = {
    "release": {
        "product_title": "Ultimate Tool",
        "license": "GPL-3.0-or-later",
        "runtime_requirements": [],
        "support_boundary": {"tested_platform": "Ubuntu 24.04"},
    },
    "version": RESERVED_VERSION,
    "archive": {"name": f"{BINARY_PREFIX}.tar.gz"},
    "source_archive": {"name": f"{SOURCE_PREFIX}.tar.gz"},
}


class ExtractedSourceTests(unittest.TestCase):
    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-source-"))
        self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)

    def package(self, **kwargs) -> Package:
        return Package(self.root, **kwargs)

    def assert_refused(self, tree: Path, manifest: Path, expected: str) -> None:
        with self.assertRaises(verify.HarnessError) as caught:
            verifier.verify_extracted_tree(tree, manifest)
        self.assertIn(expected, str(caught.exception))

    def test_a_faithful_extraction_verifies(self) -> None:
        package = self.package()
        result = verifier.verify_extracted_tree(package.tree, package.manifest_path)
        self.assertEqual(result["result"], "verified")
        self.assertEqual(result["prefix"], SOURCE_PREFIX)
        self.assertEqual(result["generated_members"], ["SOURCE-PROVENANCE.json", "vendor-sources.toml"])
        self.assertEqual(result["aggregate_sha256"], package.manifest["aggregate_sha256"])
        self.assertEqual(
            result["members"],
            len(package.files["first_party"]) + len(package.files["vendor"]) + 1 + len(RUNTIME_MEMBERS),
        )
        self.assertGreater(result["hashed_bytes"], 0)
        self.assertEqual(tree_files(package.tree), package.expected_tree_files())

    def test_absent_vendor_tree_does_not_require_a_vendor_config(self) -> None:
        package = self.package(vendored=False)
        result = verifier.verify_extracted_tree(package.tree, package.manifest_path)
        self.assertEqual(result["generated_members"], ["SOURCE-PROVENANCE.json"])
        self.assertFalse((package.tree / "vendor-sources.toml").exists())

    def test_a_tampered_member_is_refused(self) -> None:
        package = self.package()
        # Same length, different bytes: the size check cannot catch this, only the hash.
        target = package.tree / "crates/x/Cargo.toml"
        target.write_bytes(b"X" * len(target.read_bytes()))
        self.assert_refused(package.tree, package.manifest_path, "does not match the SHA-256")

    def test_a_missing_member_is_refused(self) -> None:
        package = self.package()
        (package.tree / "Cargo.lock").unlink()
        self.assert_refused(package.tree, package.manifest_path, "missing from the extracted tree")

    def test_a_size_change_is_refused(self) -> None:
        package = self.package()
        target = package.tree / "Cargo.toml"
        target.write_bytes(target.read_bytes() + b"# extra\n")
        self.assert_refused(package.tree, package.manifest_path, "the source index records")

    def test_an_added_file_is_refused(self) -> None:
        package = self.package()
        (package.tree / "crates/x/build.rs").write_bytes(b"fn main() {}\n")
        self.assert_refused(package.tree, package.manifest_path, "does not list")

    def test_an_added_directory_is_refused(self) -> None:
        package = self.package()
        (package.tree / "unlisted").mkdir()
        self.assert_refused(package.tree, package.manifest_path, "does not list")

    def test_a_symlink_is_refused(self) -> None:
        package = self.package()
        os.symlink("Cargo.toml", package.tree / "linked.toml")
        self.assert_refused(package.tree, package.manifest_path, "symbolic link")

    def test_a_hard_link_is_refused(self) -> None:
        package = self.package()
        source = self.root / "outside-inode"
        shutil.copyfile(package.tree / "Cargo.lock", source)
        (package.tree / "Cargo.lock").unlink()
        os.link(source, package.tree / "Cargo.lock")
        self.assert_refused(package.tree, package.manifest_path, "hard links")

    def test_a_missing_generated_member_is_refused(self) -> None:
        package = self.package()
        (package.tree / "SOURCE-PROVENANCE.json").unlink()
        self.assert_refused(package.tree, package.manifest_path, "SOURCE-PROVENANCE.json")

    def test_a_mismatched_provenance_aggregate_is_refused(self) -> None:
        package = self.package()
        path = package.tree / "SOURCE-PROVENANCE.json"
        document = json.loads(path.read_text(encoding="utf-8"))
        document["files"]["aggregate_sha256"] = "0" * 64
        path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        self.assert_refused(package.tree, package.manifest_path, "different first-party aggregate")

    def test_a_wrong_manifest_kind_is_refused(self) -> None:
        package = self.package()
        index = json.loads(package.manifest_path.read_text(encoding="utf-8"))
        index["record_kind"] = "release_source_manifest"
        package.manifest_path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        self.assert_refused(package.tree, package.manifest_path, "point --manifest at the resolved")

    def test_a_misnamed_source_root_is_refused(self) -> None:
        package = self.package(extract_tree=False)
        alt = self.root / "alt"
        package.extract(alt)
        renamed = alt / "renamed-tree"
        os.rename(alt / SOURCE_PREFIX, renamed)
        self.assert_refused(renamed, package.manifest_path, "records the archive prefix")

    def test_a_symlinked_source_root_is_refused(self) -> None:
        package = self.package()
        link = self.root / (SOURCE_PREFIX + "-link")
        os.symlink(package.tree, link)
        with self.assertRaises(verify.HarnessError):
            verifier.verify_extracted_tree(link, package.manifest_path)


class MalformedIndexTests(unittest.TestCase):
    """A wrong-typed or contradictory index is refused, never half-used."""

    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-index-"))
        self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)

    def mutate(self, mutate) -> tuple[Package, Path]:
        package = Package(self.root)
        index = json.loads(package.manifest_path.read_text(encoding="utf-8"))
        mutate(index)
        path = self.root / "mutated.json"
        path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        return package, path

    def refused(self, mutate, expected: str) -> None:
        package, path = self.mutate(mutate)
        with self.assertRaises(verify.HarnessError) as caught:
            verifier.verify_extracted_tree(package.tree, path)
        self.assertIn(expected, str(caught.exception))

    def refused_provenance(self, mutate, expected: str) -> None:
        package = Package(self.root)
        path = package.tree / "SOURCE-PROVENANCE.json"
        document = json.loads(path.read_text(encoding="utf-8"))
        mutate(document)
        path.write_text(json.dumps(document), encoding="utf-8")
        with self.assertRaises(verify.HarnessError) as caught:
            verifier.verify_extracted_tree(package.tree, package.manifest_path)
        self.assertIn(expected, str(caught.exception))

    # -- first-party files -------------------------------------------------
    def test_a_non_list_files_record_is_refused(self) -> None:
        self.refused(lambda index: index.update({"files": "Cargo.toml"}), "is not a list")

    def test_a_non_object_entry_is_refused(self) -> None:
        self.refused(lambda index: index["files"].append("Cargo.toml"), "is not an object")

    def test_a_non_string_path_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"path": 7}), "not a non-empty string")

    def test_a_parent_segment_path_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"path": "../escape"}), "parent-directory segment")

    def test_a_backslash_path_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"path": "crates\\x"}), "backslash")

    def test_a_non_hex_digest_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"sha256": "zz"}), "lowercase SHA-256")

    def test_an_uppercase_digest_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"sha256": "A" * 64}), "lowercase SHA-256")

    def test_a_boolean_size_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"size": True}), "non-negative integer size")

    def test_a_string_size_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"size": "12"}), "non-negative integer size")

    def test_a_float_size_is_refused(self) -> None:
        self.refused(lambda index: index["files"][0].update({"size": 1.5}), "non-negative integer size")

    def test_a_duplicate_inside_one_category_is_refused(self) -> None:
        self.refused(lambda index: index["files"].append(dict(index["files"][0])), "more than once")

    def test_a_duplicate_member_across_categories_is_refused(self) -> None:
        self.refused(
            lambda index: index["distribution_docs"][0].update({"archive_path": "Cargo.toml"}),
            "more than once",
        )

    # -- distribution documents -------------------------------------------
    def test_a_non_list_distribution_docs_record_is_refused(self) -> None:
        self.refused(lambda index: index.update({"distribution_docs": "x"}), "is not a list")

    def test_a_non_object_distribution_doc_is_refused(self) -> None:
        self.refused(lambda index: index["distribution_docs"].append("x"), "is not an object")

    def test_a_distribution_doc_absolute_path_is_refused(self) -> None:
        self.refused(
            lambda index: index["distribution_docs"][0].update({"archive_path": "/etc/passwd"}),
            "absolute member name",
        )

    def test_a_distribution_doc_bad_digest_is_refused(self) -> None:
        self.refused(lambda index: index["distribution_docs"][0].update({"sha256": "nope"}), "lowercase SHA-256")

    def test_a_distribution_doc_negative_size_is_refused(self) -> None:
        self.refused(lambda index: index["distribution_docs"][0].update({"size": -1}), "non-negative integer size")

    # -- vendored crate files ---------------------------------------------
    def test_a_non_list_vendor_files_record_is_refused(self) -> None:
        self.refused(lambda index: index["vendor"].update({"files": {"a": 1}}), "is not a list")

    def test_a_non_object_vendor_entry_is_refused(self) -> None:
        self.refused(lambda index: index["vendor"]["files"].append(3), "is not an object")

    def test_a_vendor_empty_segment_path_is_refused(self) -> None:
        self.refused(lambda index: index["vendor"]["files"][0].update({"path": "a//b"}), "empty path segment")

    def test_a_vendor_bad_digest_is_refused(self) -> None:
        self.refused(lambda index: index["vendor"]["files"][0].update({"sha256": "xyz"}), "lowercase SHA-256")

    def test_a_vendor_bad_size_is_refused(self) -> None:
        self.refused(lambda index: index["vendor"]["files"][0].update({"size": True}), "non-negative integer size")

    def test_a_duplicate_vendor_member_is_refused(self) -> None:
        self.refused(
            lambda index: index["vendor"]["files"].append(dict(index["vendor"]["files"][0])),
            "more than once",
        )

    # -- runtime corresponding-source members ------------------------------
    def test_a_non_list_runtime_members_record_is_refused(self) -> None:
        self.refused(lambda index: index["runtime_bundle"].update({"source_members": 5}), "is not a list")

    def test_a_non_object_runtime_entry_is_refused(self) -> None:
        self.refused(lambda index: index["runtime_bundle"]["source_members"].append([]), "is not an object")

    def test_a_runtime_parent_segment_path_is_refused(self) -> None:
        self.refused(
            lambda index: index["runtime_bundle"]["source_members"][0].update({"path": "../x"}),
            "parent-directory segment",
        )

    def test_a_runtime_bad_digest_is_refused(self) -> None:
        self.refused(
            lambda index: index["runtime_bundle"]["source_members"][0].update({"sha256": "q"}),
            "lowercase SHA-256",
        )

    def test_a_runtime_float_size_is_refused(self) -> None:
        self.refused(
            lambda index: index["runtime_bundle"]["source_members"][0].update({"size": 1.5}),
            "non-negative integer size",
        )

    def test_a_duplicate_runtime_member_is_refused(self) -> None:
        self.refused(
            lambda index: index["runtime_bundle"]["source_members"].append(
                dict(index["runtime_bundle"]["source_members"][0])
            ),
            "more than once",
        )

    # -- strict scalar types ----------------------------------------------
    def test_a_boolean_schema_is_refused(self) -> None:
        self.refused(lambda index: index.update({"schema": True}), "expected the integer")

    def test_a_float_schema_is_refused(self) -> None:
        self.refused(lambda index: index.update({"schema": 1.0}), "expected the integer")

    def test_a_string_runtime_present_is_refused(self) -> None:
        self.refused(lambda index: index["runtime_bundle"].update({"present": "true"}), "not a boolean")

    def test_a_numeric_runtime_present_is_refused(self) -> None:
        self.refused(lambda index: index["runtime_bundle"].update({"present": 1}), "not a boolean")

    def test_a_contradictory_aggregate_is_refused(self) -> None:
        self.refused(lambda index: index.update({"aggregate_sha256": "0" * 64}), "does not match the aggregate it records")

    # -- the archive's own bound record ------------------------------------
    def test_a_boolean_provenance_schema_is_refused(self) -> None:
        self.refused_provenance(lambda document: document.update({"schema": True}), "expected the integer")

    def test_a_float_provenance_schema_is_refused(self) -> None:
        self.refused_provenance(lambda document: document.update({"schema": 1.0}), "expected the integer")

    def test_a_boolean_provenance_count_is_refused(self) -> None:
        self.refused_provenance(lambda document: document["files"].update({"count": True}), "non-integer file count")

    def test_a_float_provenance_count_is_refused(self) -> None:
        self.refused_provenance(lambda document: document["files"].update({"count": 1.0}), "non-integer file count")


class BoundedReadTests(unittest.TestCase):
    """The per-member read cannot run past the size the index recorded."""

    def test_a_member_that_grows_after_open_is_refused(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="ut-test-install-growth-"))
        self.addCleanup(shutil.rmtree, root, ignore_errors=True)
        package = Package(root)
        expected = next(entry for entry in package.files["first_party"] if entry["path"] == "Cargo.toml")

        @contextlib.contextmanager
        def grown(path, *, max_bytes, description):
            # A stream that keeps yielding past the declared size, standing in for a
            # file another process appended to between the size check and the read.
            yield io.BytesIO(b"\0" * (expected["size"] + 64))

        with mock.patch.object(verifier.packaging, "open_regular_readonly", grown):
            with self.assertRaises(verify.HarnessError) as caught:
                verifier.verify_extracted_tree(package.tree, package.manifest_path)
        self.assertIn("larger than", str(caught.exception))

    def test_a_member_of_exactly_the_recorded_size_is_accepted(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="ut-test-install-exact-"))
        self.addCleanup(shutil.rmtree, root, ignore_errors=True)
        package = Package(root)
        expected = next(entry for entry in package.files["first_party"] if entry["path"] == "Cargo.toml")

        @contextlib.contextmanager
        def exact(path, *, max_bytes, description):
            yield io.BytesIO(b"\0" * expected["size"])

        with mock.patch.object(verifier.packaging, "open_regular_readonly", exact):
            # Same length, different bytes: the read is bounded, so Cargo.toml itself
            # is a hash mismatch, not an overlong read.
            with self.assertRaises(verify.HarnessError) as caught:
                verifier.verify_extracted_tree(package.tree, package.manifest_path)
        self.assertIn("  Cargo.toml does not match the SHA-256", str(caught.exception))


class ExtractedHelperCliTests(unittest.TestCase):
    """The helper run *out of the extracted tree*, as the instructions say."""

    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-cli-"))
        self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)

    def run_extracted(self, package: Package, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess:
        helper = package.tree / "tools" / "verify_extracted_source.py"
        return subprocess.run(
            [sys.executable, str(helper), *args],
            capture_output=True, text=True, cwd=str(cwd or self.root), timeout=180,
        )

    def test_the_extracted_helper_verifies_and_exits_zero(self) -> None:
        package = Package(self.root)
        result = self.run_extracted(package, "--source-root", str(package.tree),
                                    "--manifest", str(package.manifest_path), "--json")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["result"], "verified")

    def test_running_the_extracted_helper_leaves_no_bytecode_in_the_tree(self) -> None:
        package = Package(self.root)
        before = tree_files(package.tree)
        result = self.run_extracted(package, "--source-root", str(package.tree),
                                    "--manifest", str(package.manifest_path))
        self.assertEqual(result.returncode, 0, result.stderr)
        after = tree_files(package.tree)
        self.assertEqual(after, before)
        self.assertFalse([name for name in after if "__pycache__" in name])

    def test_the_default_source_root_is_the_current_directory(self) -> None:
        package = Package(self.root)
        result = self.run_extracted(package, "--manifest", str(package.manifest_path), cwd=package.tree)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("verified", result.stdout)

    def test_a_tampered_tree_exits_nonzero(self) -> None:
        package = Package(self.root)
        target = package.tree / "Cargo.toml"
        target.write_bytes(b"X" * len(target.read_bytes()))
        result = self.run_extracted(package, "--source-root", str(package.tree),
                                    "--manifest", str(package.manifest_path))
        self.assertEqual(result.returncode, 1)
        self.assertIn("does not match the SHA-256", result.stderr)

    def test_a_wrong_typed_index_exits_nonzero_without_a_traceback(self) -> None:
        package = Package(self.root)
        index = json.loads(package.manifest_path.read_text(encoding="utf-8"))
        index["files"][0]["size"] = "12"
        bad = self.root / "bad.json"
        bad.write_text(json.dumps(index), encoding="utf-8")
        result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(bad))
        self.assertEqual(result.returncode, 1)
        self.assertNotIn("Traceback", result.stderr)
        self.assertIn("non-negative integer size", result.stderr)

    def test_a_wrong_typed_runtime_present_exits_nonzero_without_a_traceback(self) -> None:
        package = Package(self.root)
        index = json.loads(package.manifest_path.read_text(encoding="utf-8"))
        index["runtime_bundle"]["present"] = "yes"
        bad = self.root / "bad-runtime.json"
        bad.write_text(json.dumps(index), encoding="utf-8")
        result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(bad))
        self.assertEqual(result.returncode, 1)
        self.assertNotIn("Traceback", result.stderr)
        self.assertIn("not a boolean", result.stderr)

    def test_a_missing_manifest_is_a_usage_error(self) -> None:
        package = Package(self.root)
        helper = package.tree / "tools" / "verify_extracted_source.py"
        result = subprocess.run([sys.executable, str(helper), "--source-root", str(package.tree)],
                                capture_output=True, text=True, cwd=str(self.root), timeout=60)
        self.assertEqual(result.returncode, 2)


def snapshot(root: Path) -> dict[str, tuple[str, str]]:
    """Every path under ``root``: its kind and, for a file, its content hash.

    Directories are recorded too, so a verification run that creates or removes
    even an empty directory is visible. ``__pycache__`` would show up here.
    """
    recorded: dict[str, tuple[str, str]] = {}
    for current, dirnames, filenames in os.walk(root):
        for name in dirnames:
            path = Path(current) / name
            relative = path.relative_to(root).as_posix()
            recorded[relative] = ("link", os.readlink(path)) if path.is_symlink() else ("dir", "")
        for name in filenames:
            path = Path(current) / name
            relative = path.relative_to(root).as_posix()
            if path.is_symlink():
                recorded[relative] = ("link", os.readlink(path))
            else:
                recorded[relative] = ("file", hashlib.sha256(path.read_bytes()).hexdigest())
    return recorded


def verification_command(block: str) -> str:
    """The verification invocation from a documented block, without its extraction.

    The block extracts the source archive and then runs the helper. Re-running
    only the helper against an already-extracted tree is what proves the check is
    read-only; re-extracting would overwrite whatever was tampered with first.
    """
    return "\n".join(line for line in block.splitlines() if not line.strip().startswith("tar -xzf"))


class DocumentedSequenceTests(unittest.TestCase):
    """Execute the generated instructions exactly, in one shell session."""

    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-seq-"))
        self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
        self.shim = self.root / "shim"
        self.shim.mkdir()
        (self.shim / "python3").symlink_to(sys.executable)

    def environment(self) -> dict:
        return {**os.environ, "PATH": f"{self.shim}{os.pathsep}{os.environ['PATH']}"}

    def prepare(self) -> tuple[Package, list[str]]:
        package = Package(self.root, extract_tree=False)
        binary_archive = self.root / f"{BINARY_PREFIX}.tar.gz"
        write_binary_archive(binary_archive, package.manifest_bytes)
        write_sha256sums(self.root, [binary_archive.name, package.archive.name])
        document = packaging.render_install(INSTALL_CONTEXT).decode("utf-8")
        relevant = [
            block for block in sh_blocks(document)
            if "sha256sum -c SHA256SUMS" in block
            or "bin/ultimate-tool --version" in block
            or "examples/first-run" in block
            or "verify_extracted_source.py" in block
        ]
        self.assertEqual(len(relevant), 4, relevant)
        return package, relevant

    def run_blocks(self, blocks: list[str], *, extra: str = "") -> subprocess.CompletedProcess:
        script = "\n".join(part.rstrip() for part in blocks) + "\n" + extra
        return subprocess.run(
            ["bash", "-eu", "-c", script],
            cwd=self.root, capture_output=True, text=True, timeout=300, env=self.environment(),
        )

    def assert_tree_matches_index(self, package: Package) -> Path:
        tree = self.root / SOURCE_PREFIX
        self.assertEqual(tree_files(tree), package.expected_tree_files())
        self.assertFalse([name for name in tree_files(tree) if "__pycache__" in name])
        for entry in package.files["first_party"]:
            self.assertEqual(sha256((tree / entry["path"]).read_bytes()), entry["sha256"])
        for entry in package.files["vendor"]:
            self.assertEqual(sha256((tree / "vendor" / entry["path"]).read_bytes()), entry["sha256"])
        for entry in package.manifest["distribution_docs"]:
            self.assertEqual(sha256((tree / entry["archive_path"]).read_bytes()), entry["sha256"])
        for entry in package.manifest["runtime_bundle"]["source_members"]:
            self.assertEqual(sha256((tree / entry["path"]).read_bytes()), entry["sha256"])
        return tree

    def test_the_documented_block_sequence_runs_from_one_directory(self) -> None:
        package, relevant = self.prepare()
        verify_block = next(block for block in relevant if "verify_extracted_source.py" in block)
        self.assertIn("-B ", verify_block)
        result = self.run_blocks(relevant, extra='printf "PWD=%s\\n" "$PWD"\n')
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertIn("verified", result.stdout)
        self.assertIn(f"PWD={self.root}", result.stdout)
        self.assert_tree_matches_index(package)
        self.assertTrue((self.root / BINARY_PREFIX / "examples" / "first-run").is_dir())

    def test_the_verification_command_is_read_only_and_rerunnable(self) -> None:
        package, relevant = self.prepare()
        verify_block = next(block for block in relevant if "verify_extracted_source.py" in block)
        first = self.run_blocks(relevant)
        self.assertEqual(first.returncode, 0, first.stdout + first.stderr)
        tree = self.root / SOURCE_PREFIX
        before = snapshot(tree)
        self.assertIn(("dir", ""), before.values())
        self.assertEqual(snapshot(tree), before)
        second = self.run_blocks([verification_command(verify_block)])
        self.assertEqual(second.returncode, 0, second.stdout + second.stderr)
        self.assertIn("verified", second.stdout)
        self.assertEqual(snapshot(tree), before)

    def test_tampering_an_indexed_file_fails_the_same_verification_command(self) -> None:
        package, relevant = self.prepare()
        verify_block = next(block for block in relevant if "verify_extracted_source.py" in block)
        first = self.run_blocks(relevant)
        self.assertEqual(first.returncode, 0, first.stdout + first.stderr)
        tree = self.root / SOURCE_PREFIX
        target = tree / "Cargo.toml"
        target.write_bytes(b"X" * len(target.read_bytes()))
        tampered = snapshot(tree)
        result = self.run_blocks([verification_command(verify_block)])
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("does not match the SHA-256", result.stderr)
        # The failed check wrote nothing: the tampered tree is byte for byte as it was.
        self.assertEqual(snapshot(tree), tampered)


class GeneratedInstallTests(unittest.TestCase):
    """The documented procedure must name the shipped helper and real inputs."""

    def install(self) -> str:
        return packaging.render_install(INSTALL_CONTEXT).decode("utf-8")

    def test_the_impossible_command_is_gone(self) -> None:
        document = self.install()
        self.assertNotIn("sha256sum --check source-manifest.json", document)

    def test_the_procedure_names_the_shipped_helper_and_real_inputs(self) -> None:
        document = self.install()
        self.assertIn(f"python3 -B {SOURCE_PREFIX}/tools/verify_extracted_source.py", document)
        self.assertIn(f"--source-root {SOURCE_PREFIX}", document)
        self.assertIn(f"--manifest {BINARY_PREFIX}/source-manifest.json", document)
        self.assertIn(f"tar -xzf {SOURCE_PREFIX}.tar.gz", document)

    def test_no_documented_step_leaves_the_download_directory(self) -> None:
        document = self.install()
        for block in sh_blocks(document):
            for line in block.splitlines():
                stripped = line.strip()
                self.assertFalse(
                    stripped.startswith("cd "),
                    f"a documented step changes directory outside a subshell: {stripped!r}",
                )

    def test_the_documented_helper_is_the_one_in_the_tree(self) -> None:
        self.assertTrue((ROOT / "tools" / "verify_extracted_source.py").is_file())
        self.assertIn("tools/verify_extracted_source.py", self.install())

    def test_the_source_section_still_points_at_the_checksum_chain(self) -> None:
        document = self.install()
        self.assertIn("sha256sum -c SHA256SUMS", document)
        self.assertIn("SOURCE-AND-LICENSES.md", document)


if __name__ == "__main__":
    unittest.main()