# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import hashlib
import os
import pkgutil
import shutil
import socket
import ssl
import tarfile
import time
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler
from io import BytesIO
from pathlib import Path
from typing import Any

import pytest

from pants.engine.console import Console
from pants.engine.fs import (
    EMPTY_DIGEST,
    EMPTY_SNAPSHOT,
    AddPrefix,
    CreateDigest,
    Digest,
    DigestContents,
    DigestEntries,
    DigestSubset,
    Directory,
    DownloadFile,
    FileContent,
    FileDigest,
    FileEntry,
    GlobMatchErrorBehavior,
    MergeDigests,
    PathGlobs,
    PathGlobsAndRoot,
    PathMetadataRequest,
    PathMetadataResult,
    RemovePrefix,
    Snapshot,
    SnapshotDiff,
    SymlinkEntry,
    Workspace,
)
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.native_engine import PathMetadata, PathMetadataKind, PathNamespace
from pants.engine.internals.scheduler import ExecutionError
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.rules import goal_rule, implicitly, rule
from pants.testutil.rule_runner import QueryRule, RuleRunner
from pants.util.collections import assert_single_element
from pants.util.contextutil import http_server, temporary_dir
from pants.util.dirutil import relative_symlink, safe_file_dump


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            QueryRule(Digest, [CreateDigest]),
            QueryRule(DigestContents, [PathGlobs]),
            QueryRule(DigestEntries, [Digest]),
            QueryRule(DigestEntries, [PathGlobs]),
            QueryRule(Snapshot, [CreateDigest]),
            QueryRule(Snapshot, [DigestSubset]),
            QueryRule(Snapshot, [PathGlobs]),
            QueryRule(PathMetadataResult, [PathMetadataRequest]),
        ],
        isolated_local_store=True,
    )


ROLAND_FILE_DIGEST = FileDigest(
    "693d8db7b05e99c6b7a7c0616456039d89c555029026936248085193559a0b5d", 16
)
ROLAND_DIGEST = Digest("63949aa823baf765eff07b946050d76ec0033144c785a94d3ebd82baa931cd16", 80)


def prime_store_with_roland_digest(rule_runner: RuleRunner) -> None:
    """Prime lmdb_store with a directory of a file named 'roland' and contents 'European
    Burmese'."""
    with temporary_dir() as temp_dir:
        Path(temp_dir, "roland").write_text("European Burmese")
        snapshot = rule_runner.scheduler.capture_snapshots(
            (PathGlobsAndRoot(PathGlobs(["*"]), temp_dir),)
        )[0]
    assert snapshot.files == ("roland",)
    assert snapshot.digest == ROLAND_DIGEST

    # NB: Capturing a Snapshot avoids persisting directory entries to disk, so we have to ensure
    # that independently.
    rule_runner.scheduler.ensure_directory_digest_persisted(snapshot.digest)


def setup_fs_test_tar(rule_runner: RuleRunner) -> None:
    """Extract fs_test.tar into the rule_runner's build root.

    Note that we use a tar, rather than rule_runner.write_files(), because it has symlinks set up a
    certain way.

    Contents:

        4.txt
        a
        ├── 3.txt
        ├── 4.txt.ln -> ../4.txt
        └── b
            ├── 1.txt
            └── 2
        c.ln -> a/b
        d.ln -> a

    NB: The RuleRunner injects a BUILDROOT file in the build_root.
    """
    data = pkgutil.get_data("pants.engine.internals", "fs_test_data/fs_test.tar")
    assert data is not None
    io = BytesIO()
    io.write(data)
    io.seek(0)
    with tarfile.open(fileobj=io) as tf:
        tf.extractall(rule_runner.build_root)


FS_TAR_ALL_FILES = (
    "BUILDROOT",  # injected by RuleRunner, not present in tar
    "4.txt",
    "a/3.txt",
    "a/4.txt.ln",
    "a/b/1.txt",
    "a/b/2",
    "c.ln/1.txt",
    "c.ln/2",
    "d.ln/3.txt",
    "d.ln/4.txt.ln",
    "d.ln/b/1.txt",
    "d.ln/b/2",
)
FS_TAR_ALL_DIRS = ("a", "a/b", "c.ln", "d.ln", "d.ln/b")


def try_with_backoff(assertion_fn: Callable[[], bool], count: int = 4) -> bool:
    for i in range(count):
        time.sleep(0.1 * i)
        if assertion_fn():
            return True
    return False


# -----------------------------------------------------------------------------------------------
# `FileContent`
# -----------------------------------------------------------------------------------------------


def test_file_content_non_bytes():
    with pytest.raises(TypeError) as exc:
        FileContent(path="4.txt", content="four")

    assert str(exc.value) == "Expected 'content' to be bytes, but got str"


# -----------------------------------------------------------------------------------------------
# `PathGlobs`, including `GlobMatchErrorBehavior` and symlink handling
# -----------------------------------------------------------------------------------------------


def assert_path_globs(
    rule_runner: RuleRunner,
    globs: Iterable[str],
    *,
    expected_files: Iterable[str],
    expected_dirs: Iterable[str],
) -> None:
    snapshot = rule_runner.request(Snapshot, [PathGlobs(globs)])
    assert snapshot.files == tuple(sorted(expected_files))
    assert snapshot.dirs == tuple(sorted(expected_dirs))
    if expected_files or expected_dirs:
        assert snapshot.digest != EMPTY_DIGEST
    else:
        assert snapshot.digest == EMPTY_DIGEST


def test_path_globs_literal_files(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(rule_runner, ["4.txt"], expected_files=["4.txt"], expected_dirs=[])
    assert_path_globs(
        rule_runner,
        ["a/b/1.txt", "a/b/2"],
        expected_files=["a/b/1.txt", "a/b/2"],
        expected_dirs=["a", "a/b"],
    )
    assert_path_globs(rule_runner, ["c.ln/2"], expected_files=["c.ln/2"], expected_dirs=["c.ln"])
    assert_path_globs(
        rule_runner,
        ["d.ln/b/1.txt"],
        expected_files=["d.ln/b/1.txt"],
        expected_dirs=["d.ln", "d.ln/b"],
    )
    assert_path_globs(rule_runner, ["a/3.txt"], expected_files=["a/3.txt"], expected_dirs=["a"])
    assert_path_globs(rule_runner, ["z.fake"], expected_files=[], expected_dirs=[])


def test_path_globs_literal_directories(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(rule_runner, ["c.ln"], expected_files=[], expected_dirs=["c.ln"])
    assert_path_globs(rule_runner, ["a"], expected_files=[], expected_dirs=["a"])
    assert_path_globs(rule_runner, ["a/b"], expected_files=[], expected_dirs=["a", "a/b"])
    assert_path_globs(rule_runner, ["z"], expected_files=[], expected_dirs=[])


def test_path_globs_glob_pattern(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(rule_runner, ["*.txt"], expected_files=["4.txt"], expected_dirs=[])
    assert_path_globs(
        rule_runner, ["a/b/*.txt"], expected_files=["a/b/1.txt"], expected_dirs=["a", "a/b"]
    )
    assert_path_globs(
        rule_runner, ["c.ln/*.txt"], expected_files=["c.ln/1.txt"], expected_dirs=["c.ln"]
    )
    assert_path_globs(
        rule_runner, ["a/b/*"], expected_files=["a/b/1.txt", "a/b/2"], expected_dirs=["a", "a/b"]
    )
    assert_path_globs(rule_runner, ["*/0.txt"], expected_files=[], expected_dirs=[])
    assert_path_globs(
        rule_runner,
        ["*"],
        expected_files=["BUILDROOT", "4.txt"],
        expected_dirs=["a", "c.ln", "d.ln"],
    )
    assert_path_globs(
        rule_runner,
        ["*/*"],
        expected_files=[
            "a/3.txt",
            "a/4.txt.ln",
            "c.ln/1.txt",
            "c.ln/2",
            "d.ln/3.txt",
            "d.ln/4.txt.ln",
        ],
        expected_dirs=FS_TAR_ALL_DIRS,
    )
    assert_path_globs(
        rule_runner,
        ["*/*/*"],
        expected_files=["a/b/1.txt", "a/b/2", "d.ln/b/1.txt", "d.ln/b/2"],
        expected_dirs=["a", "a/b", "d.ln", "d.ln/b"],
    )


def test_path_globs_rglob_pattern(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(
        rule_runner,
        ["**/*.txt.ln"],
        expected_files=["a/4.txt.ln", "d.ln/4.txt.ln"],
        expected_dirs=["a", "d.ln"],
    )
    assert_path_globs(
        rule_runner,
        ["**/*.txt"],
        expected_files=[
            "4.txt",
            "a/3.txt",
            "a/b/1.txt",
            "c.ln/1.txt",
            "d.ln/3.txt",
            "d.ln/b/1.txt",
        ],
        expected_dirs=FS_TAR_ALL_DIRS,
    )
    assert_path_globs(
        rule_runner,
        ["**/3.t*t"],
        expected_files=["a/3.txt", "d.ln/3.txt"],
        expected_dirs=["a", "d.ln"],
    )
    assert_path_globs(rule_runner, ["**/*.fake"], expected_files=[], expected_dirs=[])
    assert_path_globs(
        rule_runner, ["**"], expected_files=FS_TAR_ALL_FILES, expected_dirs=FS_TAR_ALL_DIRS
    )
    assert_path_globs(
        rule_runner, ["**/*"], expected_files=FS_TAR_ALL_FILES, expected_dirs=FS_TAR_ALL_DIRS
    )
    assert_path_globs(
        rule_runner,
        ["a/**"],
        expected_files=["a/3.txt", "a/4.txt.ln", "a/b/1.txt", "a/b/2"],
        expected_dirs=["a", "a/b"],
    )
    assert_path_globs(
        rule_runner,
        ["d.ln/**"],
        expected_files=["d.ln/3.txt", "d.ln/4.txt.ln", "d.ln/b/1.txt", "d.ln/b/2"],
        expected_dirs=["d.ln", "d.ln/b"],
    )
    assert_path_globs(rule_runner, ["a/**/3.txt"], expected_files=["a/3.txt"], expected_dirs=["a"])
    assert_path_globs(
        rule_runner, ["a/**/b/1.txt"], expected_files=["a/b/1.txt"], expected_dirs=["a", "a/b"]
    )
    assert_path_globs(rule_runner, ["a/**/2"], expected_files=["a/b/2"], expected_dirs=["a", "a/b"])


def test_path_globs_ignore_pattern(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(
        rule_runner,
        ["**", "!*.ln"],
        expected_files=["BUILDROOT", "4.txt", "a/3.txt", "a/b/1.txt", "a/b/2"],
        expected_dirs=["a", "a/b"],
    )


def test_path_globs_ignore_sock(rule_runner: RuleRunner) -> None:
    sock_path = os.path.join(rule_runner.build_root, "sock.sock")
    with socket.socket(socket.AF_UNIX) as sock:
        sock.bind(sock_path)
    assert os.path.exists(sock_path)
    assert not os.path.isfile(sock_path)

    rule_runner.write_files({"non-sock.txt": ""})
    assert_path_globs(
        rule_runner,
        ["**"],
        expected_files=["BUILDROOT", "non-sock.txt"],
        expected_dirs=[],
    )


def test_path_globs_remove_duplicates(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(
        rule_runner, ["*", "**"], expected_files=FS_TAR_ALL_FILES, expected_dirs=FS_TAR_ALL_DIRS
    )
    assert_path_globs(
        rule_runner,
        ["**/*.txt", "a/b/1.txt", "4.txt"],
        expected_files=[
            "4.txt",
            "a/3.txt",
            "c.ln/1.txt",
            "d.ln/3.txt",
            "a/b/1.txt",
            "d.ln/b/1.txt",
        ],
        expected_dirs=FS_TAR_ALL_DIRS,
    )


def test_path_globs_parent_link(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    assert_path_globs(
        rule_runner,
        ["c.ln/../3.txt"],
        expected_files=["c.ln/../3.txt"],
        expected_dirs=["c.ln", "c.ln/.."],
    )


def test_path_globs_symlink_escaping_errors(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    link = os.path.join(rule_runner.build_root, "subdir/escaping")
    dest = os.path.join(rule_runner.build_root, "../../..")
    relative_symlink(dest, link)

    exc_reg = r".*While expanding link.*subdir/escaping.*may not traverse outside of the buildroot"
    with pytest.raises(Exception, match=exc_reg):
        assert_path_globs(rule_runner, ["subdir/escaping"], expected_files=[], expected_dirs=[])


def test_path_globs_symlink_dead(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    link = os.path.join(rule_runner.build_root, "subdir/dead")
    dest = os.path.join(rule_runner.build_root, "this_file_does_not_exist")
    relative_symlink(dest, link)

    # Because the symlink does not escape, it should be ignored, rather than cause an error.
    assert_path_globs(rule_runner, ["subdir/dead"], expected_files=[], expected_dirs=[])


def test_path_globs_symlink_dead_nested(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)
    link = os.path.join(rule_runner.build_root, "subdir/dead")
    dest = os.path.join(
        rule_runner.build_root, "this_folder_does_not_exist/this_file_does_not_exist"
    )
    relative_symlink(dest, link)

    # Because the symlink does not escape, it should be ignored, rather than cause an error.
    assert_path_globs(rule_runner, ["subdir/dead"], expected_files=[], expected_dirs=[])


def test_path_globs_symlink_loop(rule_runner: RuleRunner) -> None:
    # Matching a recursive glob against a link which points to its parent directory would cause
    # infinite recursion, so we eagerly error instead.
    setup_fs_test_tar(rule_runner)
    link = os.path.join(rule_runner.build_root, "subdir/link.ln")
    dest = os.path.join(rule_runner.build_root, "subdir")
    relative_symlink(dest, link)

    exc_reg = r".*Maximum link depth exceeded"
    with pytest.raises(Exception, match=exc_reg):
        assert_path_globs(rule_runner, ["**"], expected_files=[], expected_dirs=[])


def test_path_globs_to_digest_contents(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)

    def get_contents(globs: Iterable[str]) -> set[FileContent]:
        return set(rule_runner.request(DigestContents, [PathGlobs(globs)]))

    assert get_contents(["4.txt", "a/4.txt.ln"]) == {
        FileContent("4.txt", b"four\n"),
        FileContent("a/4.txt.ln", b"four\n"),
    }
    assert get_contents(["c.ln/../3.txt"]) == {FileContent("c.ln/../3.txt", b"three\n")}

    # Directories are empty.
    assert not get_contents(["a/b"])
    assert not get_contents(["c.ln"])


def test_path_globs_to_digest_entries(rule_runner: RuleRunner) -> None:
    setup_fs_test_tar(rule_runner)

    def get_entries(globs: Iterable[str]) -> set[FileEntry | Directory | SymlinkEntry]:
        return set(rule_runner.request(DigestEntries, [PathGlobs(globs)]))

    assert get_entries(["4.txt", "a/4.txt.ln"]) == {
        FileEntry(
            "4.txt",
            FileDigest("ab929fcd5594037960792ea0b98caf5fdaf6b60645e4ef248c28db74260f393e", 5),
        ),
        FileEntry(
            "a/4.txt.ln",
            FileDigest("ab929fcd5594037960792ea0b98caf5fdaf6b60645e4ef248c28db74260f393e", 5),
        ),
    }
    assert get_entries(["c.ln/../3.txt"]) == {
        FileEntry(
            "c.ln/../3.txt",
            FileDigest("f6936912184481f5edd4c304ce27c5a1a827804fc7f329f43d273b8621870776", 6),
        )
    }

    # Directories are empty.
    assert get_entries(["a/b"]) == {Directory("a/b")}
    assert get_entries(["c.ln"]) == {Directory("c.ln")}


def test_digest_entries_handles_empty_directory(rule_runner: RuleRunner) -> None:
    digest = rule_runner.request(
        Digest, [CreateDigest([Directory("a/b"), FileContent("a/foo.txt", b"four\n")])]
    )
    entries = rule_runner.request(DigestEntries, [digest])
    assert entries == DigestEntries(
        [
            Directory("a/b"),
            FileEntry(
                "a/foo.txt",
                FileDigest("ab929fcd5594037960792ea0b98caf5fdaf6b60645e4ef248c28db74260f393e", 5),
            ),
        ]
    )


def test_digest_entries_handles_symlinks(rule_runner: RuleRunner) -> None:
    digest = rule_runner.request(
        Digest,
        [
            CreateDigest(
                [
                    SymlinkEntry("a.ln", "a.txt"),
                    SymlinkEntry("b.ln", "b.txt"),
                    FileContent("a.txt", b"four\n"),
                ]
            )
        ],
    )
    entries = rule_runner.request(DigestEntries, [digest])
    assert entries == DigestEntries(
        [
            SymlinkEntry("a.ln", "a.txt"),
            FileEntry(
                "a.txt",
                FileDigest("ab929fcd5594037960792ea0b98caf5fdaf6b60645e4ef248c28db74260f393e", 5),
            ),
            SymlinkEntry("b.ln", "b.txt"),
        ]
    )


@pytest.mark.parametrize(
    "create_digest, files, dirs",
    [
        pytest.param(
            CreateDigest(
                [
                    FileContent("file.txt", b"four\n"),
                    SymlinkEntry("symlink", "file.txt"),
                    SymlinkEntry("relsymlink", "./file.txt"),
                    SymlinkEntry("a/symlink", "../file.txt"),
                    SymlinkEntry("a/b/symlink", "../../file.txt"),
                ]
            ),
            ("a/b/symlink", "a/symlink", "file.txt", "relsymlink", "symlink"),
            ("a", "a/b"),
            id="simple",
        ),
        pytest.param(
            CreateDigest(
                [
                    FileContent("file.txt", b"four\n"),
                    SymlinkEntry(
                        "circular1", "./circular1"
                    ),  # After so many traversals, we give up
                    SymlinkEntry("circular2", "circular2"),  # After so many traversals, we give up
                    SymlinkEntry("chain1", "chain2"),
                    SymlinkEntry("chain2", "chain3"),
                    SymlinkEntry("chain3", "chain1"),
                    SymlinkEntry(
                        "a/symlink", "file.txt"
                    ),  # looks for a/file.txt, which doesn't exist
                    SymlinkEntry("a/too-far.ln", "../../file.txt"),  # went too far up
                    SymlinkEntry("a/parent", ".."),
                    SymlinkEntry("too-far.ln", "../file.txt"),  # went too far up
                    SymlinkEntry("absolute1.ln", str(Path(__file__).resolve())),  # absolute path
                    SymlinkEntry("absolute2.ln", "/file.txt"),
                ]
            ),
            ("file.txt",),
            ("a",),
            id="ignored",
        ),
        pytest.param(
            CreateDigest(
                [
                    FileContent("file.txt", b"four\n"),
                    SymlinkEntry("a/b/parent-file.ln", "../../file.txt"),
                    SymlinkEntry("dirlink", "a"),
                ]
            ),
            ("a/b/parent-file.ln", "dirlink/b/parent-file.ln", "file.txt"),
            ("a", "a/b", "dirlink", "dirlink/b"),
            id="parentdir-in-symlink-target",
        ),
        pytest.param(
            CreateDigest(
                [
                    FileContent("a/file.txt", b"four\n"),
                    SymlinkEntry("dirlink", "a"),
                    SymlinkEntry("double-dirlink", "dirlink"),
                ]
            ),
            ("a/file.txt", "dirlink/file.txt", "double-dirlink/file.txt"),
            ("a", "dirlink", "double-dirlink"),
            id="double-dirlink",
        ),
        pytest.param(
            CreateDigest(
                [
                    FileContent("a/file.txt", b"four\n"),
                    SymlinkEntry("a/self", "."),
                ]
            ),
            tuple(f"a/{'self/' * count}file.txt" for count in range(64)),
            ("a",),
            id="self-dir",
        ),
    ],
)
def test_snapshot_and_contents_are_symlink_oblivious(
    rule_runner: RuleRunner,
    create_digest: CreateDigest,
    files: tuple[str, ...],
    dirs: tuple[str, ...],
) -> None:
    digest = rule_runner.request(Digest, [create_digest])
    snapshot = rule_runner.request(Snapshot, [digest])
    assert snapshot.files == files
    assert snapshot.dirs == dirs
    contents = rule_runner.request(DigestContents, [digest])
    assert tuple(content.path for content in contents) == files


def test_glob_match_error_behavior(rule_runner: RuleRunner, caplog) -> None:
    setup_fs_test_tar(rule_runner)
    test_name = f"{__name__}.{test_glob_match_error_behavior.__name__}()"

    def evaluate_path_globs(globs: Iterable[str], error_behavior: GlobMatchErrorBehavior) -> None:
        pg = PathGlobs(
            globs,
            glob_match_error_behavior=error_behavior,
            description_of_origin=(
                test_name if error_behavior != GlobMatchErrorBehavior.ignore else None
            ),
        )
        rule_runner.request(Snapshot, [pg])

    with pytest.raises(Exception) as exc:
        evaluate_path_globs(["not-a-file.txt"], GlobMatchErrorBehavior.error)
    assert f'Unmatched glob from {test_name}: "not-a-file.txt"' in str(exc.value)

    with pytest.raises(Exception) as exc:
        evaluate_path_globs(["not-a-file.txt", "!ignore.txt"], GlobMatchErrorBehavior.error)
    assert f'Unmatched glob from {test_name}: "not-a-file.txt", exclude: "ignore.txt"' in str(
        exc.value
    )

    # TODO: get Rust logging working with RuleRunner.
    # caplog.clear()
    # evaluate_path_globs(["not-a-file.txt"], GlobMatchErrorBehavior.warn)
    # assert len(caplog.records) == 1
    # assert f'Unmatched glob from {test_name}: "not-a-file.txt"' in caplog.text

    caplog.clear()
    evaluate_path_globs(["not-a-file.txt"], GlobMatchErrorBehavior.ignore)
    assert len(caplog.records) == 0


# -----------------------------------------------------------------------------------------------
# `PathGlobsAndRoot`
# -----------------------------------------------------------------------------------------------


def test_snapshot_from_outside_buildroot(rule_runner: RuleRunner) -> None:
    with temporary_dir() as temp_dir:
        Path(temp_dir, "roland").write_text("European Burmese")
        snapshot = rule_runner.scheduler.capture_snapshots(
            [PathGlobsAndRoot(PathGlobs(["*"]), temp_dir)]
        )[0]
    assert snapshot.files == ("roland",)
    assert snapshot.digest == ROLAND_DIGEST


def test_multiple_snapshots_from_outside_buildroot(rule_runner: RuleRunner) -> None:
    with temporary_dir() as temp_dir:
        Path(temp_dir, "roland").write_text("European Burmese")
        Path(temp_dir, "susannah").write_text("I don't know")
        snapshots = rule_runner.scheduler.capture_snapshots(
            [
                PathGlobsAndRoot(PathGlobs(["roland"]), temp_dir),
                PathGlobsAndRoot(PathGlobs(["susannah"]), temp_dir),
                PathGlobsAndRoot(PathGlobs(["doesnotexist"]), temp_dir),
            ]
        )
    assert len(snapshots) == 3
    assert snapshots[0].files == ("roland",)
    assert snapshots[0].digest == ROLAND_DIGEST
    assert snapshots[1].files == ("susannah",)
    assert snapshots[1].digest == Digest(
        "d3539cfc21eb4bab328ca9173144a8e932c515b1b9e26695454eeedbc5a95f6f", 82
    )
    assert snapshots[2] == EMPTY_SNAPSHOT


def test_snapshot_from_outside_buildroot_failure(rule_runner: RuleRunner) -> None:
    with temporary_dir() as temp_dir:
        with pytest.raises(Exception) as exc:
            rule_runner.scheduler.capture_snapshots(
                [PathGlobsAndRoot(PathGlobs(["*"]), os.path.join(temp_dir, "doesnotexist"))]
            )
    assert "doesnotexist" in str(exc.value)


# -----------------------------------------------------------------------------------------------
# `CreateDigest`
# -----------------------------------------------------------------------------------------------


def test_create_empty_directory(rule_runner: RuleRunner) -> None:
    res = rule_runner.request(Snapshot, [CreateDigest([Directory("a/")])])
    assert res.dirs == ("a",)
    assert not res.files
    assert res.digest != EMPTY_DIGEST

    res = rule_runner.request(
        Snapshot, [CreateDigest([Directory("x/y/z"), Directory("m"), Directory("m/n")])]
    )
    assert res.dirs == ("m", "m/n", "x", "x/y", "x/y/z")
    assert not res.files
    assert res.digest != EMPTY_DIGEST


def test_create_digest_with_file_entries(rule_runner: RuleRunner) -> None:
    # Retrieve some known FileEntry's from the test tar.
    setup_fs_test_tar(rule_runner)
    file_entries = rule_runner.request(DigestEntries, [PathGlobs(["4.txt", "a/4.txt.ln"])])

    # Make a snapshot with just those files.
    snapshot = rule_runner.request(Snapshot, [CreateDigest(file_entries)])
    assert snapshot.dirs == ("a",)
    assert snapshot.files == ("4.txt", "a/4.txt.ln")
    assert snapshot.digest != EMPTY_DIGEST


# -----------------------------------------------------------------------------------------------
# `MergeDigests`
# -----------------------------------------------------------------------------------------------


def test_merge_digests(rule_runner: RuleRunner) -> None:
    with temporary_dir() as temp_dir:
        Path(temp_dir, "roland").write_text("European Burmese")
        Path(temp_dir, "susannah").write_text("Not sure actually")
        (
            empty_snapshot,
            roland_snapshot,
            susannah_snapshot,
            both_snapshot,
        ) = rule_runner.scheduler.capture_snapshots(
            (
                PathGlobsAndRoot(PathGlobs(["doesnotmatch"]), temp_dir),
                PathGlobsAndRoot(PathGlobs(["roland"]), temp_dir),
                PathGlobsAndRoot(PathGlobs(["susannah"]), temp_dir),
                PathGlobsAndRoot(PathGlobs(["*"]), temp_dir),
            )
        )

    empty_merged = rule_runner.request(Digest, [MergeDigests((empty_snapshot.digest,))])
    assert empty_snapshot.digest == empty_merged

    roland_merged = rule_runner.request(
        Digest, [MergeDigests((roland_snapshot.digest, empty_snapshot.digest))]
    )
    assert roland_snapshot.digest == roland_merged

    both_merged = rule_runner.request(
        Digest, [MergeDigests((roland_snapshot.digest, susannah_snapshot.digest))]
    )
    assert both_snapshot.digest == both_merged


# -----------------------------------------------------------------------------------------------
# `DigestSubset`
# -----------------------------------------------------------------------------------------------


def generate_original_digest(rule_runner: RuleRunner) -> Digest:
    files = [
        FileContent(path, b"dummy content")
        for path in [
            "a.txt",
            "b.txt",
            "c.txt",
            "subdir/a.txt",
            "subdir/b.txt",
            "subdir2/a.txt",
            "subdir2/nested_subdir/x.txt",
        ]
    ]
    return rule_runner.request(
        Digest,
        [CreateDigest(files)],
    )


def test_digest_subset_empty(rule_runner: RuleRunner) -> None:
    subset_snapshot = rule_runner.request(
        Snapshot, [DigestSubset(generate_original_digest(rule_runner), PathGlobs(()))]
    )
    assert subset_snapshot.digest == EMPTY_DIGEST
    assert subset_snapshot.files == ()
    assert subset_snapshot.dirs == ()


def test_digest_subset_globs(rule_runner: RuleRunner) -> None:
    subset_snapshot = rule_runner.request(
        Snapshot,
        [
            DigestSubset(
                generate_original_digest(rule_runner),
                PathGlobs(("a.txt", "c.txt", "subdir2/**")),
            )
        ],
    )
    assert set(subset_snapshot.files) == {
        "a.txt",
        "c.txt",
        "subdir2/a.txt",
        "subdir2/nested_subdir/x.txt",
    }
    assert set(subset_snapshot.dirs) == {"subdir2", "subdir2/nested_subdir"}

    expected_files = [
        FileContent(path, b"dummy content")
        for path in [
            "a.txt",
            "c.txt",
            "subdir2/a.txt",
            "subdir2/nested_subdir/x.txt",
        ]
    ]
    subset_digest = rule_runner.request(Digest, [CreateDigest(expected_files)])
    assert subset_snapshot.digest == subset_digest


def test_digest_subset_globs_2(rule_runner: RuleRunner) -> None:
    subset_snapshot = rule_runner.request(
        Snapshot,
        [
            DigestSubset(
                generate_original_digest(rule_runner), PathGlobs(("a.txt", "c.txt", "subdir2/*"))
            )
        ],
    )
    assert set(subset_snapshot.files) == {"a.txt", "c.txt", "subdir2/a.txt"}
    assert set(subset_snapshot.dirs) == {"subdir2", "subdir2/nested_subdir"}


def test_digest_subset_nonexistent_filename_globs(rule_runner: RuleRunner) -> None:
    # We behave according to the `GlobMatchErrorBehavior`.
    original_digest = generate_original_digest(rule_runner)
    globs = ["some_file_not_in_snapshot.txt", "a.txt"]
    subset_snapshot = rule_runner.request(
        Snapshot, [DigestSubset(original_digest, PathGlobs(globs))]
    )
    assert set(subset_snapshot.files) == {"a.txt"}
    expected_digest = rule_runner.request(
        Digest, [CreateDigest([FileContent("a.txt", b"dummy content")])]
    )
    assert subset_snapshot.digest == expected_digest

    # TODO: Fix this to actually error.
    # with pytest.raises(ExecutionError):
    #     rule_runner.request(
    #         Snapshot,
    #         [
    #             DigestSubset(
    #                 original_digest,
    #                 PathGlobs(
    #                     globs,
    #                     glob_match_error_behavior=GlobMatchErrorBehavior.error,
    #                     conjunction=GlobExpansionConjunction.all_match,
    #                     description_of_origin="test",
    #                 ),
    #             )
    #         ],
    #     )


# -----------------------------------------------------------------------------------------------
# `Digest` -> `Snapshot`
# -----------------------------------------------------------------------------------------------


def test_lift_digest_to_snapshot(rule_runner: RuleRunner) -> None:
    prime_store_with_roland_digest(rule_runner)
    snapshot = rule_runner.request(Snapshot, [ROLAND_DIGEST])
    assert snapshot.files == ("roland",)
    assert snapshot.digest == ROLAND_DIGEST


def test_error_lifting_file_digest_to_snapshot(rule_runner: RuleRunner) -> None:
    prime_store_with_roland_digest(rule_runner)
    # A file digest is not a directory digest. Here, we hash the file that was primed as part of
    # that directory, and show that we can't turn it into a Snapshot.
    text = b"European Burmese"
    hasher = hashlib.sha256()
    hasher.update(text)
    digest = Digest(fingerprint=hasher.hexdigest(), serialized_bytes_length=len(text))
    with pytest.raises(ExecutionError) as exc:
        rule_runner.request(Snapshot, [digest])
    assert "unknown directory" in str(exc.value)


# -----------------------------------------------------------------------------------------------
# `AddPrefix` and `RemovePrefix`
# -----------------------------------------------------------------------------------------------


def test_add_prefix(rule_runner: RuleRunner) -> None:
    digest = rule_runner.request(
        Digest,
        [CreateDigest([FileContent("main.ext", b""), FileContent("subdir/sub.ext", b"")])],
    )

    # Two components.
    output_digest = rule_runner.request(Digest, [AddPrefix(digest, "outer_dir/middle_dir")])
    snapshot = rule_runner.request(Snapshot, [output_digest])
    assert sorted(snapshot.files) == [
        "outer_dir/middle_dir/main.ext",
        "outer_dir/middle_dir/subdir/sub.ext",
    ]
    assert sorted(snapshot.dirs) == [
        "outer_dir",
        "outer_dir/middle_dir",
        "outer_dir/middle_dir/subdir",
    ]

    # Empty.
    output_digest = rule_runner.request(Digest, [AddPrefix(digest, "")])
    assert digest == output_digest

    # Illegal.
    with pytest.raises(Exception, match=r"The `prefix` must be relative."):
        rule_runner.request(Digest, [AddPrefix(digest, "../something")])


def test_remove_prefix(rule_runner: RuleRunner) -> None:
    relevant_files = (
        "characters/dark_tower/roland",
        "characters/dark_tower/susannah",
    )
    all_files = (
        "books/dark_tower/gunslinger",
        "characters/altered_carbon/kovacs",
        *relevant_files,
        "index",
    )

    with temporary_dir() as temp_dir:
        safe_file_dump(os.path.join(temp_dir, "index"), "books\ncharacters\n")
        safe_file_dump(
            os.path.join(temp_dir, "characters", "altered_carbon", "kovacs"),
            "Envoy",
            makedirs=True,
        )

        tower_dir = os.path.join(temp_dir, "characters", "dark_tower")
        safe_file_dump(os.path.join(tower_dir, "roland"), "European Burmese", makedirs=True)
        safe_file_dump(os.path.join(tower_dir, "susannah"), "Not sure actually", makedirs=True)

        safe_file_dump(
            os.path.join(temp_dir, "books", "dark_tower", "gunslinger"),
            "1982",
            makedirs=True,
        )

        snapshot, snapshot_with_extra_files = rule_runner.scheduler.capture_snapshots(
            [
