Merge pull request #130 from scanoss/fix/mdaloia/SP-2870-scanoss-py-fs-does-not-write-output-when-to-file

[SP-2870] fix: cyclonedx format output not writing to file

Author: matiasdaloia

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [1.27.1] - 2025-07-09
+### Fixed
+- Fixed when running `folder-scan` with `--format cyclonedx` the output was not writing to file
+- Fixed when running `container-scan` with `--format cyclonedx` the output was not writing to file
+
 ## [1.27.0] - 2025-06-30
 ### Added
 - Add directory hash calculation to folder hasher
@@ -577,3 +582,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [1.26.2]: https://github.com/scanoss/scanoss.py/compare/v1.26.1...v1.26.2
 [1.26.3]: https://github.com/scanoss/scanoss.py/compare/v1.26.2...v1.26.3
 [1.27.0]: https://github.com/scanoss/scanoss.py/compare/v1.26.3...v1.27.0
+[1.27.1]: https://github.com/scanoss/scanoss.py/compare/v1.27.0...v1.27.1

src/scanoss/__init__.py

@@ -22,4 +22,4 @@
   THE SOFTWARE.
 """
 
-__version__ = '1.27.0'
+__version__ = '1.27.1'

src/scanoss/cyclonedx.py

@@ -22,14 +22,13 @@
   THE SOFTWARE.
 """
 
+import datetime
 import json
 import os.path
 import sys
 import uuid
-import datetime
 
 from . import __version__
-
 from .scanossbase import ScanossBase
 from .spdxlite import SpdxLite
 
@@ -49,7 +48,7 @@ def __init__(self, debug: bool = False, output_file: str = None):
         self.debug = debug
         self._spdx = SpdxLite(debug=debug)
 
-    def parse(self, data: json):
+    def parse(self, data: json):  # noqa: PLR0912, PLR0915
         """
         Parse the given input (raw/plain) JSON string and return CycloneDX summary
         :param data: json - JSON object
@@ -58,7 +57,7 @@ def parse(self, data: json):
         if not data:
             self.print_stderr('ERROR: No JSON data provided to parse.')
             return None, None
-        self.print_debug(f'Processing raw results into CycloneDX format...')
+        self.print_debug('Processing raw results into CycloneDX format...')
         cdx = {}
         vdx = {}
         for f in data:
@@ -171,17 +170,22 @@ def produce_from_file(self, json_file: str, output_file: str = None) -> bool:
             success = self.produce_from_str(f.read(), output_file)
         return success
 
-    def produce_from_json(self, data: json, output_file: str = None) -> bool:
+    def produce_from_json(self, data: json, output_file: str = None) -> tuple[bool, json]:  # noqa: PLR0912
         """
-        Produce the CycloneDX output from the input data
-        :param data: JSON object
-        :param output_file: Output file (optional)
-        :return: True if successful, False otherwise
+        Produce the CycloneDX output from the raw scan results input data
+
+        Args:
+            data (json): JSON object
+            output_file (str, optional): Output file (optional). Defaults to None.
+
+        Returns:
+            bool: True if successful, False otherwise
+            json: The CycloneDX output
         """
         cdx, vdx = self.parse(data)
         if not cdx:
             self.print_stderr('ERROR: No CycloneDX data returned for the JSON string provided.')
-            return False
+            return False, None
         self._spdx.load_license_data()  # Load SPDX license name data for later reference
         #
         # Using CDX version 1.4: https://cyclonedx.org/docs/1.4/json/
@@ -264,7 +268,7 @@ def produce_from_json(self, data: json, output_file: str = None) -> bool:
         if output_file:
             file.close()
 
-        return True
+        return True, data
 
     def produce_from_str(self, json_str: str, output_file: str = None) -> bool:
         """

src/scanoss/scanner.py

@@ -22,40 +22,40 @@
   THE SOFTWARE.
 """
 
+import datetime
 import json
 import os
-from pathlib import Path
 import sys
-import datetime
+from pathlib import Path
 from typing import Any, Dict, List, Optional
-import importlib_resources
 
+import importlib_resources
 from progress.bar import Bar
 from progress.spinner import Spinner
 from pypac.parser import PACFile
 
 from scanoss.file_filters import FileFilters
 
-from .scanossapi import ScanossApi
-from .cyclonedx import CycloneDx
-from .spdxlite import SpdxLite
+from . import __version__
 from .csvoutput import CsvOutput
-from .threadedscanning import ThreadedScanning
+from .cyclonedx import CycloneDx
 from .scancodedeps import ScancodeDeps
-from .threadeddependencies import ThreadedDependencies, SCOPE
-from .scanossgrpc import ScanossGrpc
-from .scantype import ScanType
-from .scanossbase import ScanossBase
 from .scanoss_settings import ScanossSettings
+from .scanossapi import ScanossApi
+from .scanossbase import ScanossBase
+from .scanossgrpc import ScanossGrpc
 from .scanpostprocessor import ScanPostProcessor
-from . import __version__
+from .scantype import ScanType
+from .spdxlite import SpdxLite
+from .threadeddependencies import SCOPE, ThreadedDependencies
+from .threadedscanning import ThreadedScanning
 
 FAST_WINNOWING = False
 try:
     from scanoss_winnowing.winnowing import Winnowing
 
     FAST_WINNOWING = True
-except ModuleNotFoundError or ImportError:
+except (ModuleNotFoundError, ImportError):
     FAST_WINNOWING = False
     from .winnowing import Winnowing
 
@@ -284,7 +284,7 @@ def is_dependency_scan(self):
             return True
         return False
 
-    def scan_folder_with_options(
+    def scan_folder_with_options(  # noqa: PLR0913
         self,
         scan_dir: str,
         deps_file: str = None,
@@ -332,7 +332,7 @@ def scan_folder_with_options(
                 success = False
         return success
 
-    def scan_folder(self, scan_dir: str) -> bool:
+    def scan_folder(self, scan_dir: str) -> bool:  # noqa: PLR0912, PLR0915
         """
         Scan the specified folder producing fingerprints, send to the SCANOSS API and return results
 
@@ -400,7 +400,7 @@ def scan_folder(self, scan_dir: str) -> bool:
                 scan_block += wfp
                 scan_size = len(scan_block.encode('utf-8'))
                 wfp_file_count += 1
-                # If the scan request block (group of WFPs) or larger than the POST size or we have reached the file limit, add it to the queue
+                # If the scan request block (group of WFPs) or larger than the POST size or we have reached the file limit, add it to the queue  # noqa: E501
                 if wfp_file_count > self.post_file_count or scan_size >= self.max_post_size:
                     self.threaded_scan.queue_add(scan_block)
                     queue_size += 1
@@ -484,7 +484,7 @@ def __finish_scan_threaded(self, file_map: Optional[Dict[Any, Any]] = None) -> b
             self.__log_result(json.dumps(results, indent=2, sort_keys=True))
         elif self.output_format == 'cyclonedx':
             cdx = CycloneDx(self.debug, self.scan_output)
-            success = cdx.produce_from_json(results)
+            success, _ = cdx.produce_from_json(results)
         elif self.output_format == 'spdxlite':
             spdxlite = SpdxLite(self.debug, self.scan_output)
             success = spdxlite.produce_from_json(results)
@@ -509,7 +509,7 @@ def _merge_scan_results(
             for response in scan_responses:
                 if response is not None:
                     if file_map:
-                        response = self._deobfuscate_filenames(response, file_map)
+                        response = self._deobfuscate_filenames(response, file_map)  # noqa: PLW2901
                     results.update(response)
 
         dep_files = dep_responses.get('files', None) if dep_responses else None
@@ -532,7 +532,7 @@ def _deobfuscate_filenames(self, response: dict, file_map: dict) -> dict:
                 deobfuscated[key] = value
         return deobfuscated
 
-    def scan_file_with_options(
+    def scan_file_with_options(  # noqa: PLR0913
         self,
         file: str,
         deps_file: str = None,
@@ -603,7 +603,7 @@ def scan_file(self, file: str) -> bool:
             success = False
         return success
 
-    def scan_files(self, files: []) -> bool:
+    def scan_files(self, files: []) -> bool:  # noqa: PLR0912, PLR0915
         """
         Scan the specified list of files, producing fingerprints, send to the SCANOSS API and return results
         Please note that by providing an explicit list you bypass any exclusions that may be defined on the scanner
@@ -657,7 +657,7 @@ def scan_files(self, files: []) -> bool:
             file_count += 1
             if self.threaded_scan:
                 wfp_size = len(wfp.encode('utf-8'))
-                # If the WFP is bigger than the max post size and we already have something stored in the scan block, add it to the queue
+                # If the WFP is bigger than the max post size and we already have something stored in the scan block, add it to the queue  # noqa: E501
                 if scan_block != '' and (wfp_size + scan_size) >= self.max_post_size:
                     self.threaded_scan.queue_add(scan_block)
                     queue_size += 1
@@ -666,7 +666,7 @@ def scan_files(self, files: []) -> bool:
                 scan_block += wfp
                 scan_size = len(scan_block.encode('utf-8'))
                 wfp_file_count += 1
-                # If the scan request block (group of WFPs) or larger than the POST size or we have reached the file limit, add it to the queue
+                # If the scan request block (group of WFPs) or larger than the POST size or we have reached the file limit, add it to the queue  # noqa: E501
                 if wfp_file_count > self.post_file_count or scan_size >= self.max_post_size:
                     self.threaded_scan.queue_add(scan_block)
                     queue_size += 1
@@ -755,7 +755,7 @@ def scan_contents(self, filename: str, contents: bytes) -> bool:
                 success = False
         return success
 
-    def scan_wfp_file(self, file: str = None) -> bool:
+    def scan_wfp_file(self, file: str = None) -> bool:  # noqa: PLR0912, PLR0915
         """
         Scan the contents of the specified WFP file (in the current process)
         :param file: Scan the contents of the specified WFP file (in the current process)

src/scanoss/scanners/container_scanner.py

@@ -436,10 +436,12 @@ def _format_cyclonedx_output(self) -> str:
         scan_results = {}
         for f in self.scanner.decorated_scan_results['files']:
             scan_results[f['file']] = [f]
-        if not cdx.produce_from_json(scan_results, self.output_file):
+        success, cdx_output = cdx.produce_from_json(scan_results)
+        if not success:
             error_msg = 'Failed to produce CycloneDX output'
             self.base.print_stderr(error_msg)
-            raise ValueError(error_msg)
+            return None
+        return json.dumps(cdx_output, indent=2)
 
     def _format_spdxlite_output(self) -> str:
         """

src/scanoss/scanners/scanner_hfh.py

@@ -162,7 +162,7 @@ def _format_plain_output(self) -> str:
             else str(self.scanner.scan_results)
         )
 
-    def _format_cyclonedx_output(self) -> str:
+    def _format_cyclonedx_output(self) -> str:  # noqa: PLR0911
         if not self.scanner.scan_results:
             return ''
         try:
@@ -196,14 +196,16 @@ def _format_cyclonedx_output(self) -> str:
 
             decorated_scan_results = self.scanner.client.get_dependencies(get_dependencies_json_request)
 
-            cdx = CycloneDx(self.base.debug, self.output_file)
+            cdx = CycloneDx(self.base.debug)
             scan_results = {}
             for f in decorated_scan_results['files']:
                 scan_results[f['file']] = [f]
-            if not cdx.produce_from_json(scan_results, self.output_file):
+            success, cdx_output = cdx.produce_from_json(scan_results)
+            if not success:
                 error_msg = 'ERROR: Failed to produce CycloneDX output'
                 self.base.print_stderr(error_msg)
-                raise ValueError(error_msg)
+                return None
+            return json.dumps(cdx_output, indent=2)
         except Exception as e:
             self.base.print_stderr(f'ERROR: Failed to get license information: {e}')
             return None