Expose ipc_main with arguments instead of argparse

fixit/cli/__init__.py

@@ -199,26 +199,24 @@ class IPCResult:
     paths: List[str]
 
 
-def ipc_main(opts: LintOpts) -> IPCResult:
+def run_ipc(
+    opts: LintOpts,
+    paths: List[str],
+    prefix: Optional[str] = None,
+    workers: LintWorkers = LintWorkers.CPU_COUNT,
+) -> IPCResult:
     """
     Given a LintOpts config with lint rules and lint success/failure report formatter,
-    this IPC helper took paths of source files from either a path file (with @paths arg)
-    or a list of paths as args. Results are formed as JSON and delimited by newlines.
+    this IPC helper takes a path of source files (with an optional `prefix` that will be prepended).
+    Results are formed as JSON and delimited by newlines.
     It uses a multiprocess pool and the results are streamed to stdout as soon
     as they're available.
 
     Returns an IPCResult object.
     """
-    parser = argparse.ArgumentParser(
-        description="Runs Fixit lint rules and print results as console output.",
-        fromfile_prefix_chars="@",
-        parents=[get_multiprocessing_parser()],
-    )
-    parser.add_argument("paths", nargs="*", help="List of paths to run lint rules on.")
-    parser.add_argument("--prefix", help="A prefix to be added to all paths.")
-    args: argparse.Namespace = parser.parse_args()
+
     paths: Generator[str, None, None] = (
-        os.path.join(args.prefix, p) if args.prefix else p for p in args.paths
+        os.path.join(prefix, p) if prefix else p for p in paths
     )
 
     full_repo_metadata_config = opts.full_repo_metadata_config
@@ -230,7 +228,7 @@ def ipc_main(opts: LintOpts) -> IPCResult:
         get_file_lint_result_json,
         paths,
         opts,
-        workers=args.workers,
+        workers=workers,
         metadata_caches=metadata_caches,
     )
     for results in results_iter:
@@ -239,4 +237,26 @@ def ipc_main(opts: LintOpts) -> IPCResult:
         for result in results:
             print(result)
 
-    return IPCResult(args.paths)
+    return IPCResult(list(paths))
+
+
+def ipc_main(opts: LintOpts) -> IPCResult:
+    """
+    Like `run_ipc` instead this function expects arguments to be collected through argparse.
+    This IPC helper takes paths of source files from either a path file (with @paths arg)
+    or a list of paths as args.
+
+    Returns an IPCResult object.
+    """
+    parser = argparse.ArgumentParser(
+        description="Runs Fixit lint rules and print results as console output.",
+        fromfile_prefix_chars="@",
+        parents=[get_multiprocessing_parser()],
+    )
+    parser.add_argument("paths", nargs="*", help="List of paths to run lint rules on.")
+    parser.add_argument("--prefix", help="A prefix to be added to all paths.")
+    args: argparse.Namespace = parser.parse_args()
+
+    return run_ipc(
+        opts=opts, paths=args.paths, prefix=args.prefix, workers=args.workers
+    )

fixit/cli/tests/test_ipc.py

@@ -0,0 +1,105 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+import contextlib
+import io
+import json
+import os
+import tempfile
+from typing import Any, Dict
+
+from libcst.testing.utils import UnitTest
+
+from fixit.cli import run_ipc
+from fixit.cli.args import LintWorkers
+from fixit.cli.tests.test_lint_opts import generate_mock_lint_opt
+
+
+EXPECTED_SUCCESS_REPORT: Dict[str, Any] = json.loads(
+    """{"path": "fill-this-out", "status": "success", "reports": ["fake picklable report"]}"""
+)
+EXPECTED_FAILURE_REPORT: Dict[str, Any] = json.loads(
+    """{"path": "fill-this-out", "status": "failure", "reports": ["fake picklable report"]}"""
+)
+
+
+class IpcTest(UnitTest):
+    def setUp(self) -> None:
+        self.opts = generate_mock_lint_opt()
+
+    def test_single_path_ipc(self) -> None:
+        with io.StringIO() as buffer, tempfile.TemporaryDirectory() as prefix, contextlib.redirect_stdout(
+            buffer
+        ):
+            # create a valid file for the test to run against
+            path = "path.py"
+
+            with open(os.path.join(prefix, path), "w") as fd:
+                fd.write("""test_str = 'hello world'""")
+
+            run_ipc(
+                opts=self.opts,
+                paths=[path],
+                prefix=prefix,
+                workers=LintWorkers.USE_CURRENT_THREAD,
+            )
+
+            # get values from the buffer before we close it
+            buffer.flush()
+            output = buffer.getvalue()
+
+        report = json.loads(output)
+
+        target_report = EXPECTED_SUCCESS_REPORT.copy()
+        target_report["path"] = os.path.join(prefix, path)
+
+        self.assertDictEqual(report, target_report)
+
+    def test_multi_path_ipc(self) -> None:
+        with io.StringIO() as buffer, tempfile.TemporaryDirectory() as prefix, contextlib.redirect_stdout(
+            buffer
+        ):
+            path_a = "path_a.py"
+            path_b = "path_b.py"
+            # this path doesn't exist at all, but the runner should still handle it gracefully
+            path_c = "does_not_exist.tmp"
+
+            # create a valid file for the test to run against
+            with open(os.path.join(prefix, path_a), "w") as fd_a:
+                fd_a.write("""test_str = 'hello world'""")
+
+            # now create an invalid one
+            # mismatched tab-indent will do the trick
+            with open(os.path.join(prefix, path_b), "w") as fd_b:
+                fd_b.write("""\ta = 1\nb = 2""")
+
+            run_ipc(
+                opts=self.opts,
+                paths=[path_a, path_b, path_c],
+                prefix=prefix,
+                workers=LintWorkers.USE_CURRENT_THREAD,
+            )
+
+            # get values from the buffer before we close it
+            buffer.flush()
+            output = buffer.getvalue()
+
+        # each report is separated by a new-line
+        reports = output.strip().split("\n")
+        self.assertEqual(len(reports), 3)
+        report_a, report_b, report_c = [json.loads(report) for report in reports]
+
+        target_report_a = EXPECTED_SUCCESS_REPORT.copy()
+        target_report_a["path"] = os.path.join(prefix, path_a)
+
+        target_report_b = EXPECTED_FAILURE_REPORT.copy()
+        target_report_b["path"] = os.path.join(prefix, path_b)
+
+        target_report_c = EXPECTED_FAILURE_REPORT.copy()
+        target_report_c["path"] = os.path.join(prefix, path_c)
+
+        self.assertDictEqual(report_a, target_report_a)
+        self.assertDictEqual(report_b, target_report_b)
+        self.assertDictEqual(report_c, target_report_c)

fixit/cli/tests/test_lint_opts.py

@@ -5,7 +5,7 @@
 
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Collection, List, Sequence, cast
+from typing import Collection, List, Optional, Sequence, cast
 
 from libcst import Module
 from libcst.testing.utils import UnitTest
@@ -37,6 +37,22 @@ def create_reports(
         return [FakeLintSuccessReport(str(path), "success", ["fake picklable report"])]
 
 
+@dataclass(frozen=True)
+class FakeLintFailureReport(LintFailureReportBase):
+    path: str
+    status: str
+    reports: Collection[str]
+
+    @staticmethod
+    def create_reports(
+        path: Path,
+        reports: Collection[BaseLintRuleReport],
+        global_list: List[str],
+    ) -> Sequence["FakeLintFailureReport"]:
+        global_list.append(str(path))
+        return [FakeLintFailureReport(str(path), "failure", ["fake picklable report"])]
+
+
 class FakeRule(CstLintRule):
     def visit_Module(self, node: Module) -> None:
         self.report(node, "Dummy message")
@@ -55,15 +71,22 @@ def mock_operation(
     return cast(Sequence[FakeLintSuccessReport], results)
 
 
+def generate_mock_lint_opt(global_list: Optional[List[str]] = None) -> LintOpts:
+    if global_list is None:
+        global_list = []
+
+    return LintOpts(
+        {FakeRule},
+        FakeLintSuccessReport,
+        FakeLintFailureReport,
+        extra={"global_list": global_list},
+    )
+
+
 class LintOptsTest(UnitTest):
     def setUp(self) -> None:
         self.global_list = []
-        self.opts = LintOpts(
-            {FakeRule},
-            FakeLintSuccessReport,
-            LintFailureReportBase,
-            extra={"global_list": self.global_list},
-        )
+        self.opts = generate_mock_lint_opt(global_list=self.global_list)
 
     def test_extra_opts(self) -> None:
         path = "path.py"