Refactor output code (#156)

chore(output_handler): refactor output handling

- Remove unnecessary empty methods from OutputHandler
- Move code writing to stdout or to the specified file to OutputHandler
- Move code computing the exit code to OutputHandler
- Rename concrete classes from ${FORMAT}Handler to ${FORMAT}OutputHandler and their files from ${format}_output.py to ${format}_output_handler.py
- Add docstrings to OutputHandler methods

Author: agateau-gg

ggshield/cmd.py

@@ -13,7 +13,7 @@
 from .hook_cmd import precommit_cmd, prepush_cmd
 from .ignore import ignore
 from .install import install
-from .output import JSONHandler, OutputHandler, TextHandler
+from .output import JSONOutputHandler, OutputHandler, TextOutputHandler
 from .pre_receive_cmd import prereceive_cmd
 from .quota import quota
 from .status import status
@@ -150,9 +150,9 @@ def scan(
     if max_commits:
         config.max_commits_for_hook = max_commits
 
-    output_handler_cls: Type[OutputHandler] = TextHandler
+    output_handler_cls: Type[OutputHandler] = TextOutputHandler
     if json_output:
-        output_handler_cls = JSONHandler
+        output_handler_cls = JSONOutputHandler
 
     ctx.obj["output_handler"] = output_handler_cls(
         show_secrets=config.show_secrets, verbose=config.verbose, output=output

ggshield/dev_scan.py

@@ -176,7 +176,7 @@ def path_cmd(
         )
         scan = ScanCollection(id=" ".join(paths), type="path_scan", results=results)
 
-        return output_handler.process_scan(scan)[1]
+        return output_handler.process_scan(scan)
     except Exception as error:
         return handle_exception(error, config.verbose)
 
@@ -260,5 +260,5 @@ def scan_commit_range(
 
         return_code = output_handler.process_scan(
             ScanCollection(id=scan_id, type="commit-range", scans=scans)
-        )[1]
+        )
     return return_code

ggshield/docker.py

@@ -127,7 +127,7 @@ def docker_name_cmd(ctx: click.Context, name: str) -> int:
                 banlisted_detectors=config.banlisted_detectors,
             )
 
-            return output_handler.process_scan(scan)[1]
+            return output_handler.process_scan(scan)
         except Exception as error:
             return handle_exception(error, config.verbose)
 
@@ -161,6 +161,6 @@ def docker_archive_cmd(
             banlisted_detectors=config.banlisted_detectors,
         )
 
-        return output_handler.process_scan(scan)[1]
+        return output_handler.process_scan(scan)
     except Exception as error:
         return handle_exception(error, config.verbose)

ggshield/hook_cmd.py

@@ -5,7 +5,7 @@
 import click
 
 from ggshield.dev_scan import scan_commit_range
-from ggshield.output import TextHandler
+from ggshield.output import TextOutputHandler
 from ggshield.scan import Commit, ScanCollection
 from ggshield.utils import EMPTY_SHA, EMPTY_TREE, SupportedScanMode, handle_exception
 
@@ -22,7 +22,7 @@ def precommit_cmd(
     scan as a pre-commit git hook.
     """
     config = ctx.obj["config"]
-    output_handler = TextHandler(
+    output_handler = TextOutputHandler(
         show_secrets=config.show_secrets, verbose=config.verbose, output=None
     )
     try:
@@ -39,7 +39,7 @@ def precommit_cmd(
 
         return output_handler.process_scan(
             ScanCollection(id="cached", type="pre-commit", results=results)
-        )[1]
+        )
     except Exception as error:
         return handle_exception(error, config.verbose)
 

ggshield/output/__init__.py

@@ -1,6 +1,6 @@
-from .json import JSONHandler
+from .json import JSONOutputHandler
 from .output_handler import OutputHandler
-from .text import TextHandler
+from .text import TextOutputHandler
 
 
-__all__ = ["TextHandler", "OutputHandler", "JSONHandler"]
+__all__ = ["TextOutputHandler", "OutputHandler", "JSONOutputHandler"]

ggshield/output/json/__init__.py

@@ -1,4 +1,4 @@
-from .json_output import JSONHandler
+from .json_output_handler import JSONOutputHandler
 
 
-__all__ = ["JSONHandler"]
+__all__ = ["JSONOutputHandler"]

ggshield/output/json/json_output.py ggshield/output/json/json_output_handler.py

@@ -1,6 +1,5 @@
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, cast
 
-import click
 from pygitguardian.client import VERSIONS
 from pygitguardian.models import Match, PolicyBreak
 
@@ -13,26 +12,29 @@
 from ggshield.utils import Filemode, find_match_indices, get_lines_from_content
 
 
-class JSONHandler(OutputHandler):
-    def process_scan(
+class JSONOutputHandler(OutputHandler):
+    def _process_scan_impl(self, scan: ScanCollection) -> str:
+        scan_dict = self.create_scan_dict(scan, top=True)
+        text = JSONScanCollectionSchema().dumps(scan_dict)
+        # dumps() return type is not defined, so cast `text`, otherwise mypy complains
+        return cast(str, text)
+
+    def create_scan_dict(
         self, scan: ScanCollection, top: bool = True
-    ) -> Tuple[Dict[str, Any], int]:
+    ) -> Dict[str, Any]:
         scan_dict: Dict[str, Any] = {
             "id": scan.id,
             "type": scan.type,
             "total_incidents": 0,
             "total_occurrences": 0,
         }
-        return_code = 0
-
         if scan.extra_info:
             scan_dict["extra_info"] = scan.extra_info
 
         if top and scan.results:
             scan_dict["secrets_engine_version"] = VERSIONS.secrets_engine_version
 
         if scan.results:
-            return_code = 1
             for result in scan.results:
                 result_dict = self.process_result(result)
                 scan_dict.setdefault("results", []).append(result_dict)
@@ -41,22 +43,12 @@ def process_scan(
 
         if scan.scans:
             for inner_scan in scan.scans_with_results:
-                inner_scan_dict, inner_return_code = self.process_scan(
-                    inner_scan, top=False
-                )
+                inner_scan_dict = self.create_scan_dict(inner_scan, top=False)
                 scan_dict.setdefault("scans", []).append(inner_scan_dict)
                 scan_dict["total_incidents"] += inner_scan_dict["total_incidents"]
                 scan_dict["total_occurrences"] += inner_scan_dict["total_occurrences"]
-                return_code = max(return_code, inner_return_code)
-
-        if top:
-            if self.output:
-                with open(self.output, "w+") as f:
-                    f.write(JSONScanCollectionSchema().dumps(scan_dict))
-            else:
-                click.echo(JSONScanCollectionSchema().dumps(scan_dict))
 
-        return scan_dict, return_code
+        return scan_dict
 
     def process_result(self, result: Result) -> Dict[str, Any]:
         result_dict: Dict[str, Any] = {
@@ -110,7 +102,9 @@ def flattened_policy_break(
             flattened_dict["validity"] = policy_breaks[0].validity
 
         for policy_break in policy_breaks:
-            matches = JSONHandler.make_matches(policy_break.matches, lines, is_patch)
+            matches = JSONOutputHandler.make_matches(
+                policy_break.matches, lines, is_patch
+            )
             flattened_dict["occurrences"].extend(matches)
 
         return flattened_dict

ggshield/output/output_handler.py

@@ -1,9 +1,12 @@
-from typing import Any, Optional, Tuple
+from abc import ABC, abstractmethod
+from typing import Optional
 
-from ggshield.scan import Result, ScanCollection
+import click
 
+from ggshield.scan import ScanCollection
 
-class OutputHandler:
+
+class OutputHandler(ABC):
     show_secrets: bool = False
     verbose: bool = False
     output: Optional[str] = None
@@ -13,15 +16,41 @@ def __init__(
         show_secrets: bool,
         verbose: bool,
         output: Optional[str] = None,
-        *args: Any,
-        **kwargs: Any,
     ):
         self.show_secrets = show_secrets
         self.verbose = verbose
         self.output = output
 
-    def process_scan(self, scan: ScanCollection, top: bool = True) -> Tuple[Any, int]:
-        pass
+    def process_scan(self, scan: ScanCollection) -> int:
+        """Process a scan collection, write the report to :attr:`self.output`
+
+        :param scan: The scan collection to process
+        :return: The exit code
+        """
+        text = self._process_scan_impl(scan)
+        if self.output:
+            with open(self.output, "w+") as f:
+                f.write(text)
+        else:
+            click.echo(text)
+        return OutputHandler._get_exit_code(scan)
+
+    @abstractmethod
+    def _process_scan_impl(self, scan: ScanCollection) -> str:
+        """Implementation of scan processing,
+        called by :meth:`OutputHandler.process_scan`
+
+        Must return a string for the report.
+
+        :param scan: The scan collection to process
+        :return: The content
+        """
+        raise NotImplementedError()
 
-    def process_result(self, result: Result) -> Any:
-        pass
+    @staticmethod
+    def _get_exit_code(scan: ScanCollection) -> int:
+        if scan.results:
+            return 1
+        if scan.scans and any(x.results for x in scan.scans):
+            return 1
+        return 0

ggshield/output/text/__init__.py

@@ -1,4 +1,4 @@
-from .text_output import TextHandler
+from .text_output_handler import TextOutputHandler
 
 
-__all__ = ["TextHandler"]
+__all__ = ["TextOutputHandler"]

ggshield/output/text/text_output.py ggshield/output/text/text_output_handler.py

@@ -1,5 +1,5 @@
 from io import StringIO
-from typing import ClassVar, List, Tuple
+from typing import ClassVar, List
 
 import click
 from pygitguardian.models import Match
@@ -34,11 +34,10 @@ def get_offset(padding: int, is_patch: bool = False) -> int:
     return len(LINE_DISPLAY["file"].format("0" * padding))
 
 
-class TextHandler(OutputHandler):
+class TextOutputHandler(OutputHandler):
     nb_lines: ClassVar[int] = 3
 
-    def process_scan(self, scan: ScanCollection, top: bool = True) -> Tuple[str, int]:
-        return_code = 0
+    def _process_scan_impl(self, scan: ScanCollection, top: bool = True) -> str:
         scan_buf = StringIO()
         if scan.optional_header and (scan.results or self.verbose):
             scan_buf.write(scan.optional_header)
@@ -47,7 +46,6 @@ def process_scan(self, scan: ScanCollection, top: bool = True) -> Tuple[str, int
             scan_buf.write(secrets_engine_version())
 
         if scan.results:
-            return_code = 1
             for result in scan.results:
                 scan_buf.write(self.process_result(result))
         else:
@@ -61,21 +59,10 @@ def process_scan(self, scan: ScanCollection, top: bool = True) -> Tuple[str, int
 
         if scan.scans:
             for sub_scan in scan.scans:
-                inner_scan_str, inner_return_code = self.process_scan(
-                    sub_scan, top=False
-                )
+                inner_scan_str = self._process_scan_impl(sub_scan, top=False)
                 scan_buf.write(inner_scan_str)
-                return_code = max(return_code, inner_return_code)
-
-        scan_str = scan_buf.getvalue()
-        if top:
-            if self.output:
-                with open(self.output, "w+") as f:
-                    click.echo(scan_str, file=f)
-            else:
-                click.echo(scan_str)
 
-        return scan_str, return_code
+        return scan_buf.getvalue()
 
     def process_result(self, result: Result) -> str:
         """
@@ -111,7 +98,7 @@ def process_result(self, result: Result) -> str:
         for issue_n, (ignore_sha, policy_breaks) in enumerate(sha_dict.items(), 1):
             result_buf.write(policy_break_header(issue_n, policy_breaks, ignore_sha))
             for policy_break in policy_breaks:
-                policy_break.matches = TextHandler.make_matches(
+                policy_break.matches = TextOutputHandler.make_matches(
                     policy_break.matches, lines, is_patch
                 )
 

tests/output/test_json_output.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from ggshield.output import JSONHandler
+from ggshield.output import JSONOutputHandler, OutputHandler
 from ggshield.output.json.schemas import JSONScanCollectionSchema
 from ggshield.scan import Commit, ScanCollection
 from ggshield.utils import Filemode, SupportedScanMode
@@ -53,7 +53,7 @@
 def test_json_output(client, cache, name, input_patch, expected, snapshot):
     c = Commit()
     c._patch = input_patch
-    handler = JSONHandler(verbose=True, show_secrets=False)
+    handler = JSONOutputHandler(verbose=True, show_secrets=False)
 
     with my_vcr.use_cassette(name):
         results = c.scan(
@@ -66,10 +66,9 @@ def test_json_output(client, cache, name, input_patch, expected, snapshot):
             banlisted_detectors=None,
         )
 
-        flat_results, exit_code = handler.process_scan(
-            scan=ScanCollection(id="path", type="test", results=results), top=True
-        )
+        scan = ScanCollection(id="path", type="test", results=results)
+        json_flat_results = handler._process_scan_impl(scan)
+        exit_code = OutputHandler._get_exit_code(scan)
 
         assert exit_code == expected
-        json_flat_results = JSONScanCollectionSchema().dumps(flat_results)
         snapshot.assert_match(JSONScanCollectionSchema().loads(json_flat_results))

tests/output/test_message.py

@@ -7,7 +7,7 @@
     format_line_count_break,
     no_leak_message,
 )
-from ggshield.output.text.text_output import get_offset, get_padding
+from ggshield.output.text.text_output_handler import get_offset, get_padding
 from ggshield.text_utils import Line
 
 

tests/output/test_text_output.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from ggshield.output import TextHandler
+from ggshield.output import TextOutputHandler
 from ggshield.scan import Result
 from ggshield.scan.scannable import ScanCollection
 from ggshield.utils import Filemode
@@ -90,9 +90,9 @@
     ],
 )
 def test_leak_message(result_input, snapshot, show_secrets, verbose):
-    output_handler = TextHandler(show_secrets=show_secrets, verbose=verbose)
+    output_handler = TextOutputHandler(show_secrets=show_secrets, verbose=verbose)
     new_result = deepcopy(result_input)
-    output, exit_code = output_handler.process_scan(
+    output = output_handler._process_scan_impl(
         ScanCollection(
             id="scan",
             type="test",
@@ -101,5 +101,4 @@ def test_leak_message(result_input, snapshot, show_secrets, verbose):
         )
     )
 
-    assert exit_code == 1
     snapshot.assert_match(output)