Add support for bandit output file.

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

@@ -15,14 +15,14 @@
 from pants.backend.python.subsystems import python_native_code, subprocess_environment
 from pants.backend.python.subsystems.subprocess_environment import SubprocessEncodingEnvironment
 from pants.backend.python.target_types import PythonInterpreterCompatibility, PythonSources
-from pants.core.goals.lint import LintRequest, LintResult, LintResults
+from pants.core.goals.lint import LintOptions, LintRequest, LintResult, LintResultFile, LintResults
 from pants.core.util_rules import determine_source_files, strip_source_roots
 from pants.core.util_rules.determine_source_files import (
     AllSourceFilesRequest,
     SourceFiles,
     SpecifiedSourceFilesRequest,
 )
-from pants.engine.fs import Digest, MergeDigests, PathGlobs, Snapshot
+from pants.engine.fs import Digest, MergeDigests, PathGlobs, Snapshot, SnapshotSubset
 from pants.engine.process import FallibleProcessResult, Process
 from pants.engine.rules import SubsystemRule, rule
 from pants.engine.selectors import Get, MultiGet
@@ -51,10 +51,14 @@ class BanditPartition:
     interpreter_constraints: PexInterpreterConstraints
 
 
-def generate_args(*, specified_source_files: SourceFiles, bandit: Bandit) -> Tuple[str, ...]:
+def generate_args(
+    *, specified_source_files: SourceFiles, bandit: Bandit, output_file: Optional[str]
+) -> Tuple[str, ...]:
     args = []
     if bandit.options.config is not None:
         args.append(f"--config={bandit.options.config}")
+    if output_file:
+        args.append(f"--output={output_file}")
     args.extend(bandit.options.args)
     args.extend(specified_source_files.files)
     return tuple(args)
@@ -63,6 +67,7 @@ def generate_args(*, specified_source_files: SourceFiles, bandit: Bandit) -> Tup
 @rule
 async def bandit_lint_partition(
     partition: BanditPartition,
+    lint_options: LintOptions,
     bandit: Bandit,
     python_setup: PythonSetup,
     subprocess_encoding_environment: SubprocessEncodingEnvironment,
@@ -117,19 +122,37 @@ async def bandit_lint_partition(
     address_references = ", ".join(
         sorted(field_set.address.reference() for field_set in partition.field_sets)
     )
-
+    output_dir = lint_options.output_dir
+    output_path = output_dir / "bandit_report.txt" if output_dir else None
+    bandit_args = generate_args(
+        specified_source_files=specified_source_files,
+        bandit=bandit,
+        output_file=output_path.name if output_path else None,
+    )
     process = requirements_pex.create_process(
         python_setup=python_setup,
         subprocess_encoding_environment=subprocess_encoding_environment,
         pex_path="./bandit.pex",
-        pex_args=generate_args(specified_source_files=specified_source_files, bandit=bandit),
+        pex_args=bandit_args,
+        output_files=(output_path.name,) if output_path else None,
         input_digest=input_digest,
         description=(
             f"Run Bandit on {pluralize(len(partition.field_sets), 'target')}: {address_references}."
         ),
     )
     result = await Get(FallibleProcessResult, Process, process)
-    return LintResult.from_fallible_process_result(result, linter_name="Bandit")
+    results_file = None
+    if output_path:
+        report_file_snapshot = await Get(
+            Snapshot, SnapshotSubset(result.output_digest, PathGlobs([output_path.name]))
+        )
+        if report_file_snapshot.is_empty or len(report_file_snapshot.files) != 1:
+            raise Exception(f"Unexpected report file snapshot: {report_file_snapshot}")
+        results_file = LintResultFile(output_path=output_path, digest=report_file_snapshot.digest)
+
+    return LintResult.from_fallible_process_result(
+        result, linter_name="Bandit", results_file=results_file
+    )
 
 
 @rule(desc="Lint using Bandit")

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

@@ -7,12 +7,13 @@
 from pants.backend.python.lint.bandit.rules import rules as bandit_rules
 from pants.backend.python.target_types import PythonInterpreterCompatibility, PythonLibrary
 from pants.base.specs import FilesystemLiteralSpec, OriginSpec, SingleAddress
-from pants.core.goals.lint import LintResults
+from pants.core.goals.lint import LintOptions, LintResults
 from pants.engine.addresses import Address
-from pants.engine.fs import FileContent
+from pants.engine.fs import DigestContents, FileContent
 from pants.engine.rules import RootRule
 from pants.engine.selectors import Params
 from pants.engine.target import TargetWithOrigin
+from pants.testutil.engine.util import create_subsystem
 from pants.testutil.external_tool_test_base import ExternalToolTestBase
 from pants.testutil.interpreter_selection_utils import skip_unless_python27_and_python3_present
 from pants.testutil.option.util import create_options_bootstrapper
@@ -27,7 +28,7 @@ class BanditIntegrationTest(ExternalToolTestBase):
 
     @classmethod
     def rules(cls):
-        return (*super().rules(), *bandit_rules(), RootRule(BanditRequest))
+        return (*super().rules(), *bandit_rules(), RootRule(BanditRequest), RootRule(LintOptions))
 
     def make_target_with_origin(
         self,
@@ -54,6 +55,7 @@ def run_bandit(
         passthrough_args: Optional[str] = None,
         skip: bool = False,
         additional_args: Optional[List[str]] = None,
+        reports_dir: Optional[str] = None,
     ) -> LintResults:
         args = ["--backend-packages=pants.backend.python.lint.bandit"]
         if config:
@@ -69,6 +71,7 @@ def run_bandit(
             LintResults,
             Params(
                 BanditRequest(BanditFieldSet.create(tgt) for tgt in targets),
+                create_subsystem(LintOptions, reports_dir=reports_dir),  # type: ignore[type-var]
                 create_options_bootstrapper(args=args),
             ),
         )
@@ -79,13 +82,15 @@ def test_passing_source(self) -> None:
         assert len(result) == 1
         assert result[0].exit_code == 0
         assert "No issues identified." in result[0].stdout.strip()
+        assert result[0].results_file is None
 
     def test_failing_source(self) -> None:
         target = self.make_target_with_origin([self.bad_source])
         result = self.run_bandit([target])
         assert len(result) == 1
         assert result[0].exit_code == 1
         assert "Issue: [B303:blacklist] Use of insecure MD2, MD4, MD5" in result[0].stdout
+        assert result[0].results_file is None
 
     def test_mixed_sources(self) -> None:
         target = self.make_target_with_origin([self.good_source, self.bad_source])
@@ -94,6 +99,7 @@ def test_mixed_sources(self) -> None:
         assert result[0].exit_code == 1
         assert "good.py" not in result[0].stdout
         assert "Issue: [B303:blacklist] Use of insecure MD2, MD4, MD5" in result[0].stdout
+        assert result[0].results_file is None
 
     def test_multiple_targets(self) -> None:
         targets = [
@@ -105,6 +111,7 @@ def test_multiple_targets(self) -> None:
         assert result[0].exit_code == 1
         assert "good.py" not in result[0].stdout
         assert "Issue: [B303:blacklist] Use of insecure MD2, MD4, MD5" in result[0].stdout
+        assert result[0].results_file is None
 
     def test_precise_file_args(self) -> None:
         target = self.make_target_with_origin(
@@ -115,6 +122,7 @@ def test_precise_file_args(self) -> None:
         assert len(result) == 1
         assert result[0].exit_code == 0
         assert "No issues identified." in result[0].stdout
+        assert result[0].results_file is None
 
     @skip_unless_python27_and_python3_present
     def test_uses_correct_python_version(self) -> None:
@@ -152,13 +160,15 @@ def test_respects_config_file(self) -> None:
         assert len(result) == 1
         assert result[0].exit_code == 0
         assert "No issues identified." in result[0].stdout.strip()
+        assert result[0].results_file is None
 
     def test_respects_passthrough_args(self) -> None:
         target = self.make_target_with_origin([self.bad_source])
         result = self.run_bandit([target], passthrough_args="--skip B303")
         assert len(result) == 1
         assert result[0].exit_code == 0
         assert "No issues identified." in result[0].stdout.strip()
+        assert result[0].results_file is None
 
     def test_skip(self) -> None:
         target = self.make_target_with_origin([self.bad_source])
@@ -175,3 +185,18 @@ def test_3rdparty_plugin(self) -> None:
         assert len(result) == 1
         assert result[0].exit_code == 1
         assert "Issue: [C100:hardcoded_aws_key]" in result[0].stdout
+        assert result[0].results_file is None
+
+    def test_output_file(self) -> None:
+        target = self.make_target_with_origin([self.bad_source])
+        result = self.run_bandit([target], reports_dir=".")
+        assert len(result) == 1
+        assert result[0].exit_code == 1
+        assert result[0].stdout.strip() == ""
+        assert result[0].results_file is not None
+        output_files = self.request_single_product(DigestContents, result[0].results_file.digest)
+        assert len(output_files) == 1
+        assert (
+            "Issue: [B303:blacklist] Use of insecure MD2, MD4, MD5"
+            in output_files[0].content.decode()
+        )