CM-25543 - Implement proper handling of errors in printers (#151)

Author: MarshalX

cycode/cli/code_scanner.py

@@ -90,7 +90,7 @@ def scan_repository(context: click.Context, path: str, branch: str) -> None:
         perform_pre_scan_documents_actions(context, scan_type, documents_to_scan, is_git_diff=False)
 
         logger.debug('Found all relevant files for scanning %s', {'path': path, 'branch': branch})
-        return scan_documents(
+        scan_documents(
             context, documents_to_scan, is_git_diff=False, scan_parameters=get_scan_parameters(context, path)
         )
     except Exception as e:
@@ -420,6 +420,16 @@ def scan_documents(
 ) -> None:
     progress_bar = context.obj['progress_bar']
 
+    if not documents_to_scan:
+        progress_bar.stop()
+        ConsolePrinter(context).print_error(
+            CliError(
+                code='no_relevant_files',
+                message='Error: The scan could not be completed - relevant files to scan are not found.',
+            )
+        )
+        return
+
     scan_batch_thread_func = _get_scan_documents_thread_func(context, is_git_diff, is_commit_range, scan_parameters)
     errors, local_scan_results = run_parallel_batched_scan(
         scan_batch_thread_func, documents_to_scan, progress_bar=progress_bar
@@ -430,25 +440,7 @@ def scan_documents(
     progress_bar.stop()
 
     set_issue_detected_by_scan_results(context, local_scan_results)
-    print_results(context, local_scan_results)
-
-    if not errors:
-        return
-
-    if context.obj['output'] == 'json':
-        # TODO(MarshalX): we can't just print JSON formatted errors here
-        #  because we should return only one root json structure per scan
-        #  could be added later to "print_results" function if we wish to display detailed errors in UI
-        return
-
-    click.secho(
-        'Unfortunately, Cycode was unable to complete the full scan. '
-        'Please note that not all results may be available:',
-        fg='red',
-    )
-    for scan_id, error in errors.items():
-        click.echo(f'- {scan_id}: ', nl=False)
-        ConsolePrinter(context).print_error(error)
+    print_results(context, local_scan_results, errors)
 
 
 def scan_commit_range_documents(
@@ -506,6 +498,7 @@ def scan_commit_range_documents(
         progress_bar.update(ProgressBarSection.GENERATE_REPORT)
         progress_bar.stop()
 
+        # errors will be handled with try-except block; printing will not occur on errors
         print_results(context, [local_scan_result])
 
         scan_completed = True
@@ -693,9 +686,11 @@ def print_debug_scan_details(scan_details_response: 'ScanDetailsResponse') -> No
         logger.debug(f'Scan message: {scan_details_response.message}')
 
 
-def print_results(context: click.Context, local_scan_results: List[LocalScanResult]) -> None:
+def print_results(
+    context: click.Context, local_scan_results: List[LocalScanResult], errors: Optional[Dict[str, 'CliError']] = None
+) -> None:
     printer = ConsolePrinter(context)
-    printer.print_scan_results(local_scan_results)
+    printer.print_scan_results(local_scan_results, errors)
 
 
 def get_document_detections(

cycode/cli/printers/console_printer.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, ClassVar, Dict, List
+from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional
 
 import click
 
@@ -33,9 +33,11 @@ def __init__(self, context: click.Context) -> None:
         if self._printer_class is None:
             raise CycodeError(f'"{self.output_type}" output type is not supported.')
 
-    def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> None:
+    def print_scan_results(
+        self, local_scan_results: List['LocalScanResult'], errors: Optional[Dict[str, 'CliError']] = None
+    ) -> None:
         printer = self._get_scan_printer()
-        printer.print_scan_results(local_scan_results)
+        printer.print_scan_results(local_scan_results, errors)
 
     def _get_scan_printer(self) -> 'PrinterBase':
         printer_class = self._printer_class

cycode/cli/printers/json_printer.py

@@ -1,5 +1,5 @@
 import json
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import click
 
@@ -15,27 +15,35 @@ class JsonPrinter(PrinterBase):
     def print_result(self, result: CliResult) -> None:
         result = {'result': result.success, 'message': result.message}
 
-        click.secho(self.get_data_json(result))
+        click.echo(self.get_data_json(result))
 
     def print_error(self, error: CliError) -> None:
         result = {'error': error.code, 'message': error.message}
 
-        click.secho(self.get_data_json(result))
+        click.echo(self.get_data_json(result))
 
-    def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> None:
+    def print_scan_results(
+        self, local_scan_results: List['LocalScanResult'], errors: Optional[Dict[str, 'CliError']] = None
+    ) -> None:
         detections = []
         for local_scan_result in local_scan_results:
             for document_detections in local_scan_result.document_detections:
                 detections.extend(document_detections.detections)
 
         detections_dict = DetectionSchema(many=True).dump(detections)
 
-        click.secho(self._get_json_scan_result(detections_dict))
+        inlined_errors = []
+        if errors:
+            # FIXME(MarshalX): we don't care about scan IDs in JSON output due to clumsy JSON root structure
+            inlined_errors = [err._asdict() for err in errors.values()]
 
-    def _get_json_scan_result(self, detections: dict) -> str:
+        click.echo(self._get_json_scan_result(detections_dict, inlined_errors))
+
+    def _get_json_scan_result(self, detections: dict, errors: List[dict]) -> str:
         result = {
             'scan_id': 'DEPRECATED',  # FIXME(MarshalX): we need change JSON struct to support multiple scan results
             'detections': detections,
+            'errors': errors,
         }
 
         return self.get_data_json(result)

cycode/cli/printers/printer_base.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import click
 
@@ -18,7 +18,9 @@ def __init__(self, context: click.Context) -> None:
         self.context = context
 
     @abstractmethod
-    def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> None:
+    def print_scan_results(
+        self, local_scan_results: List['LocalScanResult'], errors: Optional[Dict[str, 'CliError']] = None
+    ) -> None:
         pass
 
     @abstractmethod

cycode/cli/printers/tables/table_printer_base.py

@@ -1,5 +1,5 @@
 import abc
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import click
 
@@ -24,13 +24,27 @@ def print_result(self, result: CliResult) -> None:
     def print_error(self, error: CliError) -> None:
         TextPrinter(self.context).print_error(error)
 
-    def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> None:
-        if all(result.issue_detected == 0 for result in local_scan_results):
+    def print_scan_results(
+        self, local_scan_results: List['LocalScanResult'], errors: Optional[Dict[str, 'CliError']] = None
+    ) -> None:
+        if not errors and all(result.issue_detected == 0 for result in local_scan_results):
             click.secho('Good job! No issues were found!!! 👏👏👏', fg=self.GREEN_COLOR_NAME)
             return
 
         self._print_results(local_scan_results)
 
+        if not errors:
+            return
+
+        click.secho(
+            'Unfortunately, Cycode was unable to complete the full scan. '
+            'Please note that not all results may be available:',
+            fg='red',
+        )
+        for scan_id, error in errors.items():
+            click.echo(f'- {scan_id}: ', nl=False)
+            self.print_error(error)
+
     def _is_git_repository(self) -> bool:
         return self.context.obj.get('remote_url') is not None
 
@@ -40,4 +54,5 @@ def _print_results(self, local_scan_results: List['LocalScanResult']) -> None:
 
     @staticmethod
     def _print_table(table: 'Table') -> None:
-        click.echo(table.get_table().draw())
+        if table.get_rows():
+            click.echo(table.get_table().draw())

cycode/cli/printers/text_printer.py

@@ -1,5 +1,5 @@
 import math
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import click
 
@@ -30,8 +30,10 @@ def print_result(self, result: CliResult) -> None:
     def print_error(self, error: CliError) -> None:
         click.secho(error.message, fg=self.RED_COLOR_NAME)
 
-    def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> None:
-        if all(result.issue_detected == 0 for result in local_scan_results):
+    def print_scan_results(
+        self, local_scan_results: List['LocalScanResult'], errors: Optional[Dict[str, 'CliError']] = None
+    ) -> None:
+        if not errors and all(result.issue_detected == 0 for result in local_scan_results):
             click.secho('Good job! No issues were found!!! 👏👏👏', fg=self.GREEN_COLOR_NAME)
             return
 
@@ -41,6 +43,18 @@ def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> Non
                     document_detections, local_scan_result.scan_id, local_scan_result.report_url
                 )
 
+        if not errors:
+            return
+
+        click.secho(
+            'Unfortunately, Cycode was unable to complete the full scan. '
+            'Please note that not all results may be available:',
+            fg='red',
+        )
+        for scan_id, error in errors.items():
+            click.echo(f'- {scan_id}: ', nl=False)
+            self.print_error(error)
+
     def _print_document_detections(
         self, document_detections: DocumentDetections, scan_id: str, report_url: Optional[str]
     ) -> None:

cycode/cli/utils/progress_bar.py

@@ -146,6 +146,8 @@ def set_section_length(self, section: 'ProgressBarSection', length: int) -> None
 
         if length == 0:
             self._skip_section(section)
+        else:
+            self._maybe_update_current_section()
 
     def _skip_section(self, section: 'ProgressBarSection') -> None:
         self._progress_bar.update(_get_section_length(section))