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

from __future__ import annotations

import dataclasses
import itertools
import os.path
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import PurePath
from textwrap import dedent

import pytest

from pants.base.deprecated import warn_or_error
from pants.base.specs import Specs
from pants.base.specs_parser import SpecsParser
from pants.engine.addresses import Address, Addresses, AddressInput, UnparsedAddressInputs
from pants.engine.environment import EnvironmentName
from pants.engine.fs import CreateDigest, FileContent
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import (
    AmbiguousCodegenImplementationsException,
    CycleException,
    Owners,
    OwnersRequest,
    TransitiveExcludesNotSupportedError,
    _DependencyMapping,
    _DependencyMappingRequest,
    _TargetParametrizations,
    hydrate_sources,
    warn_deprecated_field_type,
)
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.internals.native_engine import AddressParseException
from pants.engine.internals.parametrize import Parametrize, _TargetParametrizationsRequest
from pants.engine.internals.scheduler import ExecutionError
from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents
from pants.engine.rules import concurrently, implicitly, rule
from pants.engine.target import (
    AllTargets,
    AllUnexpandedTargets,
    AlwaysTraverseDeps,
    AsyncFieldMixin,
    CoarsenedTargets,
    Dependencies,
    DependenciesRequest,
    DepsTraversalBehavior,
    ExplicitlyProvidedDependencies,
    Field,
    FieldDefaultFactoryRequest,
    FieldDefaultFactoryResult,
    FieldSet,
    GeneratedSources,
    GenerateSourcesRequest,
    HydratedSources,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
    InvalidFieldException,
    InvalidTargetException,
    MultipleSourcesField,
    OverridesField,
    ShouldTraverseDepsPredicate,
    SingleSourceField,
    SourcesPaths,
    SourcesPathsRequest,
    SpecialCasedDependencies,
    StringField,
    Tags,
    Target,
    TargetFilesGenerator,
    Targets,
    TransitivelyExcludeDependencies,
    TransitivelyExcludeDependenciesRequest,
    TransitiveTargets,
    TransitiveTargetsRequest,
)
from pants.engine.unions import UnionMembership, UnionRule
from pants.testutil.rule_runner import QueryRule, RuleRunner, engine_error
from pants.util.docutil import bin_name
from pants.util.ordered_set import FrozenOrderedSet


class MockSingleSourceField(SingleSourceField):
    pass


class MockDependencies(Dependencies):
    supports_transitive_excludes = True
    deprecated_alias = "deprecated_field"
    deprecated_alias_removal_version = "9.9.9.dev0"


class SpecialCasedDeps1(SpecialCasedDependencies):
    alias = "special_cased_deps1"


class SpecialCasedDeps2(SpecialCasedDependencies):
    alias = "special_cased_deps2"


class ResolveField(StringField, AsyncFieldMixin):
    alias = "resolve"
    default = None


_DEFAULT_RESOLVE = "default_test_resolve"


class ResolveFieldDefaultFactoryRequest(FieldDefaultFactoryRequest):
    field_type = ResolveField


@rule
async def resolve_field_default_factory(
    request: ResolveFieldDefaultFactoryRequest,
) -> FieldDefaultFactoryResult:
    return FieldDefaultFactoryResult(lambda f: f.value or _DEFAULT_RESOLVE)


class MockMultipleSourcesField(MultipleSourcesField):
    deprecated_alias = "deprecated_sources"
    deprecated_alias_removal_version = "9.9.9.dev0"


class MockPluginField(StringField):
    alias = "plugin_string"
    required = False


class MockTarget(Target):
    alias = "target"
    core_fields = (
        MockDependencies,
        MockMultipleSourcesField,
        SpecialCasedDeps1,
        SpecialCasedDeps2,
        ResolveField,
        Tags,
    )
    deprecated_alias = "deprecated_target"
    deprecated_alias_removal_version = "9.9.9.dev0"


class MockGeneratedTarget(Target):
    alias = "generated"
    core_fields = (MockDependencies, Tags, MockSingleSourceField, ResolveField)


class MockTargetGenerator(TargetFilesGenerator):
    alias = "generator"
    core_fields = (MockMultipleSourcesField, OverridesField)
    generated_target_cls = MockGeneratedTarget
    copied_fields = ()
    moved_fields = (MockDependencies, Tags, ResolveField)


@pytest.fixture
def transitive_targets_rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            QueryRule(AllTargets, []),
            QueryRule(AllUnexpandedTargets, []),
            QueryRule(CoarsenedTargets, [Addresses]),
            QueryRule(Targets, [DependenciesRequest]),
            QueryRule(TransitiveTargets, [TransitiveTargetsRequest]),
        ],
        target_types=[MockTarget, MockTargetGenerator, MockGeneratedTarget],
        objects={"parametrize": Parametrize},
        # NB: The `graph` module masks the environment is most/all positions. We disable the
        # inherent environment so that the positions which do require the environment are
        # highlighted.
        inherent_environment=None,
    )


def test_transitive_targets(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1')
                target(name='t2', dependencies=[':t1'])
                target(name='d1', dependencies=[':t1'])
                target(name='d2', dependencies=[':t2'])
                target(name='d3')
                target(name='root', dependencies=[':d1', ':d2', ':d3'])
                """
            ),
        }
    )

    def get_target(name: str) -> Target:
        return transitive_targets_rule_runner.get_target(Address("", target_name=name))

    t1 = get_target("t1")
    t2 = get_target("t2")
    d1 = get_target("d1")
    d2 = get_target("d2")
    d3 = get_target("d3")
    root = get_target("root")

    direct_deps = transitive_targets_rule_runner.request(
        Targets, [DependenciesRequest(root[Dependencies])]
    )
    assert direct_deps == Targets([d1, d2, d3])

    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets, [TransitiveTargetsRequest([root.address, d2.address])]
    )
    assert transitive_targets.roots == (root, d2)
    # NB: `//:d2` is both a target root and a dependency of `//:root`.
    assert transitive_targets.dependencies == FrozenOrderedSet([d1, d2, d3, t2, t1])
    assert transitive_targets.closure == FrozenOrderedSet([root, d2, d1, d3, t2, t1])


def test_transitive_targets_transitive_exclude(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='base')
                target(name='intermediate', dependencies=[':base'])
                target(name='root', dependencies=[':intermediate', '!!:base'])
                """
            ),
        }
    )

    def get_target(name: str) -> Target:
        return transitive_targets_rule_runner.get_target(Address("", target_name=name))

    base = get_target("base")
    intermediate = get_target("intermediate")
    root = get_target("root")

    intermediate_direct_deps = transitive_targets_rule_runner.request(
        Targets, [DependenciesRequest(intermediate[Dependencies])]
    )
    assert intermediate_direct_deps == Targets([base])

    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets, [TransitiveTargetsRequest([root.address, intermediate.address])]
    )
    assert transitive_targets.roots == (root, intermediate)
    assert transitive_targets.dependencies == FrozenOrderedSet([intermediate])
    assert transitive_targets.closure == FrozenOrderedSet([root, intermediate])

    # Regression test that we work with deeply nested levels of excludes.
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1')
                target(name='t2', dependencies=[':t1'])
                target(name='t3', dependencies=[':t2'])
                target(name='t4', dependencies=[':t3'])
                target(name='t5', dependencies=[':t4'])
                target(name='t6', dependencies=[':t5'])
                target(name='t7', dependencies=[':t6'])
                target(name='t8', dependencies=[':t7'])
                target(name='t9', dependencies=[':t8'])
                target(name='t10', dependencies=[':t9'])
                target(name='t11', dependencies=[':t10'])
                target(name='t12', dependencies=[':t11'])
                target(name='t13', dependencies=[':t12'])
                target(name='t14', dependencies=[':t13'])
                target(name='t15', dependencies=[':t14', '!!:t1', '!!:t5'])
                """
            ),
        }
    )
    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets, [TransitiveTargetsRequest([get_target("t15").address])]
    )
    assert transitive_targets.dependencies == FrozenOrderedSet(
        [
            get_target("t14"),
            get_target("t13"),
            get_target("t12"),
            get_target("t11"),
            get_target("t10"),
            get_target("t9"),
            get_target("t8"),
            get_target("t7"),
            get_target("t6"),
            get_target("t4"),
            get_target("t3"),
            get_target("t2"),
        ]
    )


def test_special_cased_dependencies(transitive_targets_rule_runner: RuleRunner) -> None:
    """Test that subclasses of `SpecialCasedDependencies` show up if requested, but otherwise are
    left off.

    This uses the same test setup as `test_transitive_targets`, but does not use the `dependencies`
    field like normal.
    """
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1')
                target(name='t2', special_cased_deps1=[':t1'])
                target(name='d1', special_cased_deps1=[':t1'])
                target(name='d2', special_cased_deps2=[':t2'])
                target(name='d3')
                target(name='root', special_cased_deps1=[':d1', ':d2'], special_cased_deps2=[':d3'])
                """
            ),
        }
    )

    def get_target(name: str) -> Target:
        return transitive_targets_rule_runner.get_target(Address("", target_name=name))

    t1 = get_target("t1")
    t2 = get_target("t2")
    d1 = get_target("d1")
    d2 = get_target("d2")
    d3 = get_target("d3")
    root = get_target("root")

    direct_deps = transitive_targets_rule_runner.request(
        Targets, [DependenciesRequest(root[Dependencies])]
    )
    assert direct_deps == Targets()

    direct_deps = transitive_targets_rule_runner.request(
        Targets,
        [
            DependenciesRequest(
                root[Dependencies], should_traverse_deps_predicate=AlwaysTraverseDeps()
            )
        ],
    )
    assert direct_deps == Targets([d1, d2, d3])

    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets, [TransitiveTargetsRequest([root.address, d2.address])]
    )
    assert transitive_targets.roots == (root, d2)
    assert transitive_targets.dependencies == FrozenOrderedSet()
    assert transitive_targets.closure == FrozenOrderedSet([root, d2])

    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets,
        [
            TransitiveTargetsRequest(
                [root.address, d2.address],
                should_traverse_deps_predicate=AlwaysTraverseDeps(),
            )
        ],
    )
    assert transitive_targets.roots == (root, d2)
    assert transitive_targets.dependencies == FrozenOrderedSet([d1, d2, d3, t2, t1])
    assert transitive_targets.closure == FrozenOrderedSet([root, d2, d1, d3, t2, t1])


# TODO(#12871): Fix this to not be based on generated targets.
def test_transitive_targets_tolerates_generated_target_cycles(
    transitive_targets_rule_runner: RuleRunner,
) -> None:
    """For certain file-level targets like `python_source`, we should tolerate cycles because the
    underlying language tolerates them."""
    transitive_targets_rule_runner.write_files(
        {
            "dep.txt": "",
            "t1.txt": "",
            "t2.txt": "",
            "BUILD": dedent(
                """\
                generator(name='dep', sources=['dep.txt'])
                generator(name='t1', sources=['t1.txt'], dependencies=['dep.txt:dep', 't2.txt:t2'])
                generator(name='t2', sources=['t2.txt'], dependencies=['t1.txt:t1'])
                """
            ),
        }
    )
    result = transitive_targets_rule_runner.request(
        TransitiveTargets,
        [TransitiveTargetsRequest([Address("", target_name="t2")])],
    )
    assert len(result.roots) == 1
    assert result.roots[0].address == Address("", target_name="t2")
    assert [tgt.address for tgt in result.dependencies] == [
        Address("", relative_file_path="t2.txt", target_name="t2"),
        Address("", relative_file_path="t1.txt", target_name="t1"),
        Address("", relative_file_path="dep.txt", target_name="dep"),
    ]


def test_transitive_targets_with_should_traverse_deps_predicate(
    transitive_targets_rule_runner: RuleRunner,
) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1')
                target(name='t2', dependencies=[':t1'])
                target(name='t3')
                target(name='d1', dependencies=[':t1'])
                target(name='d2', dependencies=[':t2'])
                target(name='d3')
                target(name='skipped', dependencies=[':t3'])
                target(name='d4', tags=['skip_deps'], dependencies=[':skipped'])
                target(name='root', dependencies=[':d1', ':d2', ':d3', ':d4'])
                """
            ),
        }
    )

    def get_target(name: str) -> Target:
        return transitive_targets_rule_runner.get_target(Address("", target_name=name))

    t1 = get_target("t1")
    t2 = get_target("t2")
    t3 = get_target("t3")
    d1 = get_target("d1")
    d2 = get_target("d2")
    d3 = get_target("d3")
    skipped = get_target("skipped")
    d4 = get_target("d4")
    root = get_target("root")

    direct_deps = transitive_targets_rule_runner.request(
        Targets, [DependenciesRequest(root[Dependencies])]
    )
    assert direct_deps == Targets([d1, d2, d3, d4])

    class SkipDepsTagOrTraverse(ShouldTraverseDepsPredicate):
        def __call__(
            self, target: Target, field: Dependencies | SpecialCasedDependencies
        ) -> DepsTraversalBehavior:
            if "skip_deps" in (target[Tags].value or []):
                return DepsTraversalBehavior.EXCLUDE
            return DepsTraversalBehavior.INCLUDE

    predicate = SkipDepsTagOrTraverse()
    # Assert the class is frozen even though it was not decorated with @dataclass(frozen=True)
    with pytest.raises(dataclasses.FrozenInstanceError):
        # The dataclass(frozen=True) decorator is only needed if the subclass adds fields.
        predicate._callable = SkipDepsTagOrTraverse.__call__  # type: ignore[misc] # noqa

    transitive_targets = transitive_targets_rule_runner.request(
        TransitiveTargets,
        [
            TransitiveTargetsRequest(
                [root.address, d2.address], should_traverse_deps_predicate=predicate
            )
        ],
    )
    assert transitive_targets.roots == (root, d2)
    # NB: `//:d2` is both a target root and a dependency of `//:root`.
    assert transitive_targets.dependencies == FrozenOrderedSet([d1, d2, d3, d4, t2, t1])
    assert transitive_targets.closure == FrozenOrderedSet([root, d2, d1, d3, d4, t2, t1])
    # `//:d4` depends on `//:skipped` which depends on `//:t3`.
    # Nothing else depends on `//:skipped` or `//:t3`, so they should not
    # be present in the list of transitive deps thanks to `should_traverse_deps_predicate`.
    assert skipped not in transitive_targets.dependencies
    assert t3 not in transitive_targets.dependencies
    assert skipped not in transitive_targets.closure
    assert t3 not in transitive_targets.closure


def test_coarsened_targets(transitive_targets_rule_runner: RuleRunner) -> None:
    """CoarsenedTargets should "coarsen" a cycle into a single CoarsenedTarget instance."""
    transitive_targets_rule_runner.write_files(
        {
            "dep.txt": "",
            "t1.txt": "",
            "t2.txt": "",
            # Cycles are only tolerated for file-level targets like `python_source`.
            # TODO(#12871): Stop relying on only generated targets having cycle tolerance.
            "BUILD": dedent(
                """\
                generator(name='dep', sources=['dep.txt'])
                generator(name='t1', sources=['t1.txt'], dependencies=['dep.txt:dep', 't2.txt:t2'])
                generator(name='t2', sources=['t2.txt'], dependencies=['t1.txt:t1'])
                """
            ),
        }
    )

    def assert_coarsened(
        a: Address, expected_members: list[Address], expected_dependencies: list[Address]
    ) -> None:
        coarsened_targets = transitive_targets_rule_runner.request(
            CoarsenedTargets,
            [Addresses([a])],
        )
        assert sorted(t.address for t in coarsened_targets[0].members) == expected_members
        # NB: Only the direct dependencies are compared.
        assert (
            sorted(d.address for ct in coarsened_targets[0].dependencies for d in ct.members)
            == expected_dependencies
        )

    # Non-file-level targets are already validated to not have cycles, so they coarsen to
    # themselves.
    assert_coarsened(
        Address("", target_name="dep"),
        [Address("", target_name="dep")],
        [Address("", relative_file_path="dep.txt", target_name="dep")],
    )
    assert_coarsened(
        Address("", target_name="t1"),
        [Address("", target_name="t1")],
        [
            Address("", relative_file_path="t1.txt", target_name="t1"),
            Address("", relative_file_path="t2.txt", target_name="t2"),
        ],
    )
    assert_coarsened(
        Address("", target_name="t2"),
        [Address("", target_name="t2")],
        [
            Address("", relative_file_path="t1.txt", target_name="t1"),
            Address("", relative_file_path="t2.txt", target_name="t2"),
        ],
    )

    # File-level targets not involved in cycles coarsen to themselves.
    assert_coarsened(
        Address("", relative_file_path="dep.txt", target_name="dep"),
        [Address("", relative_file_path="dep.txt", target_name="dep")],
        [],
    )

    # File-level targets involved in cycles will coarsen to the cycle, and have only dependencies
    # outside of the cycle.
    cycle_files = [
        Address("", relative_file_path="t1.txt", target_name="t1"),
        Address("", relative_file_path="t2.txt", target_name="t2"),
    ]
    assert_coarsened(
        Address("", relative_file_path="t1.txt", target_name="t1"),
        cycle_files,
        [Address("", relative_file_path="dep.txt", target_name="dep")],
    )
    assert_coarsened(
        Address("", relative_file_path="t2.txt", target_name="t2"),
        cycle_files,
        [Address("", relative_file_path="dep.txt", target_name="dep")],
    )


def assert_failed_cycle(
    rule_runner: RuleRunner,
    *,
    root_target_name: str,
    subject_target_name: str,
    path_target_names: tuple[str, ...],
) -> None:
    with pytest.raises(ExecutionError) as e:
        rule_runner.request(
            TransitiveTargets,
            [TransitiveTargetsRequest([Address("", target_name=root_target_name)])],
        )
    (cycle_exception,) = e.value.wrapped_exceptions
    assert isinstance(cycle_exception, CycleException)
    assert cycle_exception.subject == Address("", target_name=subject_target_name)
    assert cycle_exception.path == tuple(Address("", target_name=p) for p in path_target_names)


def test_dep_cycle_self(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files({"BUILD": "target(name='t1', dependencies=[':t1'])"})
    assert_failed_cycle(
        transitive_targets_rule_runner,
        root_target_name="t1",
        subject_target_name="t1",
        path_target_names=("t1", "t1"),
    )


def test_dep_cycle_direct(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1', dependencies=[':t2'])
                target(name='t2', dependencies=[':t1'])
                """
            )
        }
    )
    assert_failed_cycle(
        transitive_targets_rule_runner,
        root_target_name="t1",
        subject_target_name="t1",
        path_target_names=("t1", "t2", "t1"),
    )
    assert_failed_cycle(
        transitive_targets_rule_runner,
        root_target_name="t2",
        subject_target_name="t2",
        path_target_names=("t2", "t1", "t2"),
    )


def test_dep_cycle_indirect(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                target(name='t1', dependencies=[':t2'])
                target(name='t2', dependencies=[':t3'])
                target(name='t3', dependencies=[':t2'])
                """
            )
        }
    )
    assert_failed_cycle(
        transitive_targets_rule_runner,
        root_target_name="t1",
        subject_target_name="t2",
        path_target_names=("t1", "t2", "t3", "t2"),
    )
    assert_failed_cycle(
        transitive_targets_rule_runner,
        root_target_name="t2",
        subject_target_name="t2",
        path_target_names=("t2", "t3", "t2"),
    )


def test_dep_no_cycle_indirect(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "t1.txt": "",
            "t2.txt": "",
            # TODO(#12871): Stop relying on only generated targets having cycle tolerance.
            "BUILD": dedent(
                """\
                generator(name='t1', dependencies=['t2.txt:t2'], sources=['t1.txt'])
                generator(name='t2', dependencies=[':t1'], sources=['t2.txt'])
                """
            ),
        }
    )
    result = transitive_targets_rule_runner.request(
        TransitiveTargets,
        [TransitiveTargetsRequest([Address("", target_name="t1")])],
    )
    assert len(result.roots) == 1
    assert result.roots[0].address == Address("", target_name="t1")
    assert {tgt.address for tgt in result.dependencies} == {
        Address("", relative_file_path="t1.txt", target_name="t1"),
        Address("", relative_file_path="t2.txt", target_name="t2"),
    }


def test_name_explicitly_set(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "dir1/f.txt": "",
            "dir1/BUILD": dedent(
                """\
                generator(sources=['f.txt'])
                generator(name='nombre', sources=['f.txt'])
                """
            ),
            "dir2/BUILD": dedent(
                """\
                target()
                target(name='nombre')
                """
            ),
            "dir3/BUILD": dedent(
                """\
                target(resolve=parametrize('r1', 'r2'))
                target(name='nombre', resolve=parametrize('r1', 'r2'))
                """
            ),
            "same_name/BUILD": "target(name='same_name')",
        }
    )

    def assert_is_set(addr: Address, expected: bool) -> None:
        tgt = transitive_targets_rule_runner.get_target(addr)
        assert tgt.name_explicitly_set is expected

    assert_is_set(Address("dir1"), False)
    assert_is_set(Address("dir1", target_name="nombre"), True)

    # We don't bother with generated targets.
    assert_is_set(Address("dir1", relative_file_path="f.txt"), True)
    assert_is_set(Address("dir1", target_name="nombre", relative_file_path="f.txt"), True)

    assert_is_set(Address("dir2"), False)
    assert_is_set(Address("dir2", target_name="nombre"), True)

    for r in ("r1", "r2"):
        assert_is_set(Address("dir3", parameters={"resolve": r}), False)
        assert_is_set(Address("dir3", target_name="nombre", parameters={"resolve": r}), True)

    # Even if the name is the same as the default, we should recognize when it's explicitly set.
    assert_is_set(Address("same_name"), True)


def test_deprecated_field_name(transitive_targets_rule_runner: RuleRunner, caplog) -> None:
    transitive_targets_rule_runner.write_files({"BUILD": "target(name='t', deprecated_field=[])"})
    transitive_targets_rule_runner.get_target(Address("", target_name="t"))
    assert len(caplog.records) == 1
    assert "Instead, use `dependencies`" in caplog.text


def test_deprecated_field_name_on_generator_issue_20627(
    transitive_targets_rule_runner: RuleRunner, caplog
) -> None:
    warn_deprecated_field_type.forget(MockDependencies)  # type: ignore[attr-defined]
    warn_or_error.forget(  # type: ignore[attr-defined]
        removal_version=MockDependencies.deprecated_alias_removal_version,
        entity=f"the field name {MockDependencies.deprecated_alias}",
        hint=(
            f"Instead, use `{MockDependencies.alias}`, which behaves the same. Run `{bin_name()} "
            "update-build-files` to automatically fix your BUILD files."
        ),
    )
    transitive_targets_rule_runner.write_files(
        {
            "f1.txt": "",
            "BUILD": "generator(name='t', deprecated_sources=['f1.txt'], deprecated_field=[])",
        }
    )
    transitive_targets_rule_runner.get_target(
        Address("", target_name="t", relative_file_path="f1.txt")
    )
    assert len(caplog.records) == 2
    assert "Instead, use `sources`" in caplog.text
    assert "Instead, use `dependencies`" in caplog.text


def test_resolve_deprecated_target_name(transitive_targets_rule_runner: RuleRunner, caplog) -> None:
    transitive_targets_rule_runner.write_files({"BUILD": "deprecated_target(name='t')"})
    transitive_targets_rule_runner.get_target(Address("", target_name="t"))
    assert len(caplog.records) == 1
    assert "Instead, use `target`" in caplog.text


def test_resolve_generated_target(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "f1.txt": "",
            "f2.txt": "",
            "f3.txt": "",
            "no_owner.txt": "",
            "BUILD": dedent(
                """\
                generator(name='generator', sources=['f1.txt', 'f2.txt'])
                target(name='non-generator', sources=['f1.txt'])
                """
            ),
        }
    )
    generated_target_address = Address("", target_name="generator", relative_file_path="f1.txt")
    assert transitive_targets_rule_runner.get_target(
        generated_target_address
    ) == MockGeneratedTarget({SingleSourceField.alias: "f1.txt"}, generated_target_address)

    # The target generator must actually generate the requested target.
    with pytest.raises(ExecutionError):
        transitive_targets_rule_runner.get_target(
            Address("", target_name="generator", relative_file_path="no_owner.txt")
        )

    # TODO: Using a "file address" on a target that does not generate file-level targets will fall
    # back to the target generator. See https://github.com/pantsbuild/pants/issues/14419.
    non_generator_file_address = Address(
        "", target_name="non-generator", relative_file_path="f1.txt"
    )
    assert transitive_targets_rule_runner.get_target(non_generator_file_address) == MockTarget(
        {MultipleSourcesField.alias: ["f1.txt"]},
        non_generator_file_address.maybe_convert_to_target_generator(),
    )


def test_find_all_targets(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "f1.txt": "",
            "f2.txt": "",
            "f3.txt": "",
            "no_owner.txt": "",
            "BUILD": dedent(
                """\
                generator(name='generator', sources=['f1.txt', 'f2.txt'])
                target(name='non-generator', sources=['f1.txt'])
                """
            ),
            "dir/BUILD": "target()",
        }
    )
    all_tgts = transitive_targets_rule_runner.request(AllTargets, [])
    expected = {
        Address("", target_name="generator", relative_file_path="f1.txt"),
        Address("", target_name="generator", relative_file_path="f2.txt"),
        Address("", target_name="non-generator"),
        Address("dir"),
    }
    assert {t.address for t in all_tgts} == expected

    all_unexpanded = transitive_targets_rule_runner.request(AllUnexpandedTargets, [])
    assert {t.address for t in all_unexpanded} == {*expected, Address("", target_name="generator")}


def test_invalid_target(transitive_targets_rule_runner: RuleRunner) -> None:
    transitive_targets_rule_runner.write_files(
        {
            "path/to/BUILD": dedent(
                """\
                target(name='t1', frobnot='is-unknown')
                """
            ),
        }
    )

    def get_target(name: str) -> Target:
        return transitive_targets_rule_runner.get_target(Address("path/to", target_name=name))

    with engine_error(InvalidTargetException, contains="path/to/BUILD:1: Unrecognized field"):
        get_target("t1")


@pytest.fixture
def owners_rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            QueryRule(Owners, [OwnersRequest]),
        ],
        target_types=[
            MockTarget,
            MockTargetGenerator,
            MockGeneratedTarget,
        ],
        # NB: The `graph` module masks the environment is most/all positions. We disable the
        # inherent environment so that the positions which do require the environment are
        # highlighted.
        inherent_environment=None,
    )


def assert_owners(
    rule_runner: RuleRunner,
    requested: Iterable[str],
    *,
    expected: set[Address],
    match_if_owning_build_file_included_in_sources: bool = False,
) -> None:
    result = rule_runner.request(
        Owners,
        [
            OwnersRequest(
                tuple(requested),
                match_if_owning_build_file_included_in_sources=match_if_owning_build_file_included_in_sources,
            )
        ],
    )
    assert set(result) == expected


def test_owners_source_file_does_not_exist(owners_rule_runner: RuleRunner) -> None:
    """Test when a source file belongs to a target, even though the file does not actually exist.

    This happens, for example, when the file is deleted and we're computing `--changed-since`. In
    this case, we can only use target generators rather than their generated targets.
    """
    owners_rule_runner.write_files(
        {
            "demo/f.txt": "",
            "demo/BUILD": dedent(
                """\
                target(name='not-generator', sources=['*.txt'])
                generator(name='generator', sources=['*.txt'])
                """
            ),
        }
    )
    assert_owners(
        owners_rule_runner,
        ["demo/deleted.txt"],
        expected={
            Address("demo", target_name="generator"),
            Address("demo", target_name="not-generator"),
        },
    )

    # For files that do exist, we should use generated targets when possible.
    assert_owners(
        owners_rule_runner,
        ["demo/f.txt"],
        expected={
            Address("demo", target_name="generator", relative_file_path="f.txt"),
            Address("demo", target_name="not-generator"),
        },
    )

    # If another generated target comes from the same target generator, then both that generated
    # target and its generator should be used.
    assert_owners(
        owners_rule_runner,
        ["demo/f.txt", "demo/deleted.txt"],
        expected={
            Address("demo", target_name="generator", relative_file_path="f.txt"),
            Address("demo", target_name="generator"),
            Address("demo", target_name="not-generator"),
        },
    )


def test_owners_multiple_owners(owners_rule_runner: RuleRunner) -> None:
    owners_rule_runner.write_files(
