test/java/testutil/artifact_closure.bzl

"""Helper for computing the artifact closure of a target"""

# TODO: consider upstreaming this to @rules_testing

def _of_target(target):
    to_process = target[DefaultInfo].files.to_list()
    if _ArtifactActionMapInfo not in target:
        fail("Did you forget to add the aspect to analysis_test(extra_target_under_test_aspects = )?")
    map = target[_ArtifactActionMapInfo].map
    result = []
    visited = {}
    for __ in range(len(map)):
        if not to_process:
            break
        next_to_process = []
        for x in to_process:
            if x in visited:
                continue
            visited[x] = None
            result.append(x)
            if x not in map:
                # source file or not visible to us (toolchain?)
                continue
            a = map[x]
            next_to_process.extend([f for f in a.inputs.to_list() if f not in visited])
        to_process = next_to_process
    return result

_ArtifactActionMapInfo = provider(
    "Map of artifacts to actions",
    fields = ["map"],
)

def _aspect_impl(target, ctx):
    map = {}
    for action in target.actions:
        for output in action.outputs.to_list():
            map[output] = action

    # Rollup from all dep attributes
    for attr_name in dir(ctx.rule.attr):
        attr = getattr(ctx.rule.attr, attr_name)
        if type(attr) != "list":
            attr = [attr]
        for val in attr:
            if type(val) == "Target" and _ArtifactActionMapInfo in val:
                map = map | val[_ArtifactActionMapInfo].map
    return _ArtifactActionMapInfo(map = map)

_aspect = aspect(_aspect_impl, attr_aspects = ["*"])

artifact_closure = struct(
    aspect = _aspect,
    of_target = _of_target,
)