26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.65

/src/python/pants/backend/python/lint/bandit/rules.py

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

from __future__ import annotations

from pants.backend.python.lint.bandit.subsystem import Bandit, BanditFieldSet
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex
from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions
from pants.core.util_rules.config_files import find_config_file
from pants.core.util_rules.partitions import Partition
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.engine.fs import CreateDigest, Directory, MergeDigests, RemovePrefix
from pants.engine.intrinsics import create_digest, execute_process, merge_digests, remove_prefix
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize


class BanditRequest(LintTargetsRequest):
    field_set_type = BanditFieldSet
    tool_subsystem = Bandit  # type: ignore[assignment]


def generate_argv(source_files: SourceFiles, bandit: Bandit) -> tuple[str, ...]:
    args: list[str] = []
    if bandit.config is not None:
        args.append(f"--config={bandit.config}")
    args.extend(bandit.args)
    args.extend(source_files.files)
    return tuple(args)


@rule
async def partition_bandit(
    request: BanditRequest.PartitionRequest[BanditFieldSet],
    bandit: Bandit,
    python_setup: PythonSetup,
) -> Partitions[BanditFieldSet, InterpreterConstraints]:
    if bandit.skip:
        return Partitions()

    # NB: Bandit output depends upon which Python interpreter version it's run with
    # ( https://github.com/PyCQA/bandit#under-which-version-of-python-should-i-install-bandit).
    # We batch targets by their constraints to ensure, for example, that all Python 2 targets run
    # together and all Python 3 targets run together.
    constraints_to_field_sets = InterpreterConstraints.group_field_sets_by_constraints(
        request.field_sets, python_setup
    )

    return Partitions(
        Partition(field_sets, constraints)
        for constraints, field_sets in constraints_to_field_sets.items()
    )


@rule(desc="Lint with Bandit", level=LogLevel.DEBUG)
async def bandit_lint(
    request: BanditRequest.Batch[BanditFieldSet, InterpreterConstraints], bandit: Bandit
) -> LintResult:
    assert request.partition_metadata is not None

    interpreter_constraints = request.partition_metadata
    bandit_pex_get = create_venv_pex(
        **implicitly(bandit.to_pex_request(interpreter_constraints=interpreter_constraints))
    )
    config_files_get = find_config_file(bandit.config_request)
    source_files_get = determine_source_files(
        SourceFilesRequest(field_set.source for field_set in request.elements)
    )
    # Ensure that the empty report dir exists.
    report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))

    bandit_pex, config_files, report_directory, source_files = await concurrently(
        bandit_pex_get, config_files_get, report_directory_digest_get, source_files_get
    )

    input_digest = await merge_digests(
        MergeDigests((source_files.snapshot.digest, config_files.snapshot.digest, report_directory))
    )

    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                bandit_pex,
                argv=generate_argv(source_files, bandit),
                input_digest=input_digest,
                description=f"Run Bandit on {pluralize(len(request.elements), 'file')}.",
                output_directories=(REPORT_DIR,),
                level=LogLevel.DEBUG,
            )
        )
    )
    report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))
    return LintResult.create(request, result, report=report)


def rules():
    return (
        *collect_rules(),
        *BanditRequest.rules(),
        *pex.rules(),
    )

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	from pants.backend.python.lint.bandit.subsystem import Bandit, BanditFieldSet	2✔
7	from pants.backend.python.subsystems.setup import PythonSetup	2✔
8	from pants.backend.python.util_rules import pex	2✔
9	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
10	from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex	2✔
11	from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions	2✔
12	from pants.core.util_rules.config_files import find_config_file	2✔
13	from pants.core.util_rules.partitions import Partition	2✔
14	from pants.core.util_rules.source_files import (	2✔
15	SourceFiles,
16	SourceFilesRequest,
17	determine_source_files,
18	)
19	from pants.engine.fs import CreateDigest, Directory, MergeDigests, RemovePrefix	2✔
20	from pants.engine.intrinsics import create_digest, execute_process, merge_digests, remove_prefix	2✔
21	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	2✔
22	from pants.util.logging import LogLevel	2✔
23	from pants.util.strutil import pluralize	2✔
24
25
26	class BanditRequest(LintTargetsRequest):	2✔
27	field_set_type = BanditFieldSet	2✔
28	tool_subsystem = Bandit # type: ignore[assignment]	2✔
29
30
31	def generate_argv(source_files: SourceFiles, bandit: Bandit) -> tuple[str, ...]:	2✔
32	args: list[str] = []	2✔
33	if bandit.config is not None:	2✔
34	args.append(f"--config={bandit.config}")	×
35	args.extend(bandit.args)	2✔
36	args.extend(source_files.files)	2✔
37	return tuple(args)	2✔
38
39
40	@rule	2✔
41	async def partition_bandit(	2✔
42	request: BanditRequest.PartitionRequest[BanditFieldSet],
43	bandit: Bandit,
44	python_setup: PythonSetup,
45	) -> Partitions[BanditFieldSet, InterpreterConstraints]:
46	if bandit.skip:	2✔
47	return Partitions()	×
48
49	# NB: Bandit output depends upon which Python interpreter version it's run with
50	# ( https://github.com/PyCQA/bandit#under-which-version-of-python-should-i-install-bandit).
51	# We batch targets by their constraints to ensure, for example, that all Python 2 targets run
52	# together and all Python 3 targets run together.
53	constraints_to_field_sets = InterpreterConstraints.group_field_sets_by_constraints(	2✔
54	request.field_sets, python_setup
55	)
56
57	return Partitions(	2✔
58	Partition(field_sets, constraints)
59	for constraints, field_sets in constraints_to_field_sets.items()
60	)
61
62
63	@rule(desc="Lint with Bandit", level=LogLevel.DEBUG)	2✔
64	async def bandit_lint(	2✔
65	request: BanditRequest.Batch[BanditFieldSet, InterpreterConstraints], bandit: Bandit
66	) -> LintResult:
67	assert request.partition_metadata is not None	2✔
68
69	interpreter_constraints = request.partition_metadata	2✔
70	bandit_pex_get = create_venv_pex(	2✔
71	**implicitly(bandit.to_pex_request(interpreter_constraints=interpreter_constraints))
72	)
73	config_files_get = find_config_file(bandit.config_request)	2✔
74	source_files_get = determine_source_files(	2✔
75	SourceFilesRequest(field_set.source for field_set in request.elements)
76	)
77	# Ensure that the empty report dir exists.
78	report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))	2✔
79
80	bandit_pex, config_files, report_directory, source_files = await concurrently(	2✔
81	bandit_pex_get, config_files_get, report_directory_digest_get, source_files_get
82	)
83
84	input_digest = await merge_digests(	2✔
85	MergeDigests((source_files.snapshot.digest, config_files.snapshot.digest, report_directory))
86	)
87
88	result = await execute_process(	2✔
89	**implicitly(
90	VenvPexProcess(
91	bandit_pex,
92	argv=generate_argv(source_files, bandit),
93	input_digest=input_digest,
94	description=f"Run Bandit on {pluralize(len(request.elements), 'file')}.",
95	output_directories=(REPORT_DIR,),
96	level=LogLevel.DEBUG,
97	)
98	)
99	)
100	report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))	2✔
101	return LintResult.create(request, result, report=report)	2✔
102
103
104	def rules():	2✔
105	return (	2✔
106	*collect_rules(),
107	*BanditRequest.rules(),
108	*pex.rules(),
109	)

pantsbuild / pants / 26342152999

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous