chore:SP-3607 refactor inspect module

Author: agustingroh

Makefile

@@ -50,6 +50,13 @@ publish_test:  ## Publish the Python package to TestPyPI
 	@echo "Publishing package to TestPyPI..."
 	twine upload --repository testpypi dist/*
 
+linter: ## Run ruff linter with docker
+	docker run --rm -v $(PWD):/src -w /src ghcr.io/astral-sh/ruff:0.14.2 check $$(git diff --name-only --diff-filter=ACMR | grep '\.py$$' | xargs)
+
+fix-linter: ## Run ruff linter with docker
+	docker run --rm -v $(PWD):/src -w /src ghcr.io/astral-sh/ruff:0.14.2 check $$(git diff --name-only --diff-filter=ACMR | grep '\.py$$' | xargs) --fix
+
+
 publish:  ## Publish Python package to PyPI
 	@echo "Publishing package to PyPI..."
 	twine upload dist/*

src/scanoss/cli.py

@@ -35,12 +35,6 @@
 from scanoss.cryptography import Cryptography, create_cryptography_config_from_args
 from scanoss.delta import Delta
 from scanoss.export.dependency_track import DependencyTrackExporter
-from scanoss.inspection.dependency_track.project_violation import (
-    DependencyTrackProjectViolationPolicyCheck,
-)
-from scanoss.inspection.raw.component_summary import ComponentSummary
-from scanoss.inspection.raw.license_summary import LicenseSummary
-from scanoss.inspection.raw.match_summary import MatchSummary
 from scanoss.scanners.container_scanner import (
     DEFAULT_SYFT_COMMAND,
     DEFAULT_SYFT_TIMEOUT,
@@ -75,8 +69,14 @@
 from .cyclonedx import CycloneDx
 from .filecount import FileCount
 from .gitlabqualityreport import GitLabQualityReport
-from .inspection.raw.copyleft import Copyleft
-from .inspection.raw.undeclared_component import UndeclaredComponent
+from .inspection.policy_check.dependency_track.project_violation import (
+    DependencyTrackProjectViolationPolicyCheck,
+)
+from .inspection.policy_check.scanoss.copyleft import Copyleft
+from .inspection.policy_check.scanoss.undeclared_component import UndeclaredComponent
+from .inspection.summary.component_summary import ComponentSummary
+from .inspection.summary.license_summary import LicenseSummary
+from .inspection.summary.match_summary import MatchSummary
 from .results import Results
 from .scancodedeps import ScancodeDeps
 from .scanner import FAST_WINNOWING, Scanner
@@ -1757,7 +1757,6 @@ def inspect_copyleft(parser, args):
             exclude=args.exclude,  # Licenses to ignore
             explicit=args.explicit,  # Explicit license list
         )
-
         # Execute inspection and exit with appropriate status code
         status, _ = i_copyleft.run()
         sys.exit(status)

src/scanoss/inspection/raw/__init__.py renamed to src/scanoss/inspection/policy_check/__init__.py

@@ -26,9 +26,9 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional, TypedDict
 
-from ...services.dependency_track_service import DependencyTrackService
-from ..policy_check import PolicyCheck, PolicyStatus
-from ..utils.markdown_utils import generate_jira_table, generate_table
+from ....services.dependency_track_service import DependencyTrackService
+from ...utils.markdown_utils import generate_jira_table, generate_table
+from ..policy_check import PolicyCheck, PolicyOutput, PolicyStatus
 
 # Constants
 PROCESSING_RETRY_DELAY = 5  # seconds
@@ -171,7 +171,7 @@ def __init__(  # noqa: PLR0913
         self.url = url.strip().rstrip('/') if url else None
         self.dep_track_service = DependencyTrackService(self.api_key, self.url, debug=debug, trace=trace, quiet=quiet)
 
-    def _json(self, project_violations: list[PolicyViolationDict]) -> Dict[str, Any]:
+    def _json(self, project_violations: list[PolicyViolationDict]) -> PolicyOutput:
         """
         Format project violations as JSON.
         
@@ -181,12 +181,12 @@ def _json(self, project_violations: list[PolicyViolationDict]) -> Dict[str, Any]
         Returns:
             Dictionary containing JSON formatted results and summary
         """
-        return {
-            "details": json.dumps(project_violations, indent=2),
-            "summary": f'{len(project_violations)} policy violations were found.\n',
-        }
+        return PolicyOutput(
+            details= json.dumps(project_violations, indent=2),
+            summary= f'{len(project_violations)} policy violations were found.\n',
+        )
 
-    def _markdown(self, project_violations: list[PolicyViolationDict]) -> Dict[str, Any]:
+    def _markdown(self, project_violations: list[PolicyViolationDict]) -> PolicyOutput:
         """
         Format Dependency Track violations to Markdown format.
         
@@ -198,7 +198,7 @@ def _markdown(self, project_violations: list[PolicyViolationDict]) -> Dict[str,
         """
         return self._md_summary_generator(project_violations, generate_table)
 
-    def _jira_markdown(self, data: list[PolicyViolationDict]) -> Dict[str, Any]:
+    def _jira_markdown(self, data: list[PolicyViolationDict]) -> PolicyOutput:
         """
         Format project violations for Jira Markdown.
         
@@ -357,8 +357,7 @@ def _set_project_id(self) -> None:
             self.print_stderr(f'Error: Failed to get project uuid from: {dt_project}')
             raise ValueError(f'Error: Project {self.project_name}@{self.project_version} does not have a valid UUID')
 
-    @staticmethod
-    def _sort_project_violations(violations: List[PolicyViolationDict]) -> List[PolicyViolationDict]:
+    def _sort_project_violations(self,violations: List[PolicyViolationDict]) -> List[PolicyViolationDict]:
         """
         Sort project violations by priority.
         
@@ -377,7 +376,7 @@ def _sort_project_violations(violations: List[PolicyViolationDict]) -> List[Poli
             key=lambda x: -type_priority.get(x.get('type', 'OTHER'), 1)
         )
 
-    def _md_summary_generator(self, project_violations: list[PolicyViolationDict], table_generator):
+    def _md_summary_generator(self, project_violations: list[PolicyViolationDict], table_generator) -> PolicyOutput:
         """
         Generates a Markdown summary of project policy violations.
 
@@ -396,10 +395,10 @@ def _md_summary_generator(self, project_violations: list[PolicyViolationDict], t
         """
         if project_violations is None:
             self.print_stderr('Warning: No project violations found. Returning empty results.')
-            return {
-                "details": "h3. Dependency Track Project Violations\n\nNo policy violations found.\n",
-                "summary": "0 policy violations were found.\n",
-            }
+            return PolicyOutput(
+                details= "h3. Dependency Track Project Violations\n\nNo policy violations found.\n",
+                summary= "0 policy violations were found.\n",
+            )
         headers = ['State', 'Risk Type', 'Policy Name', 'Component', 'Date']
         c_cols = [0, 1]
         rows: List[List[str]] = []
@@ -424,11 +423,11 @@ def _md_summary_generator(self, project_violations: list[PolicyViolationDict], t
             ]
             rows.append(row)
         # End for loop
-        return {
-            "details": f'### Dependency Track Project Violations\n{table_generator(headers, rows, c_cols)}\n\n'
+        return PolicyOutput(
+            details= f'### Dependency Track Project Violations\n{table_generator(headers, rows, c_cols)}\n\n'
                        f'View project in Dependency Track [here]({self.url}/projects/{self.project_id}).\n',
-            "summary": f'{len(project_violations)} policy violations were found.\n'
-        }
+            summary= f'{len(project_violations)} policy violations were found.\n'
+        )
 
     def run(self) -> int:
         """
@@ -470,10 +469,11 @@ def run(self) -> int:
             self.print_stderr('Error: Invalid format specified.')
             return PolicyStatus.ERROR.value
         # Format and output data - handle empty results gracefully
-        data = formatter(self._sort_project_violations(dt_project_violations))
-        self.print_to_file_or_stdout(data['details'], self.output)
-        self.print_to_file_or_stderr(data['summary'], self.status)
+        policy_output = formatter(self._sort_project_violations(dt_project_violations))
+        self.print_to_file_or_stdout(policy_output.details, self.output)
+        self.print_to_file_or_stderr(policy_output.summary, self.status)
         # Return appropriate status based on violation count
         if len(dt_project_violations) > 0:
             return PolicyStatus.POLICY_FAIL.value
         return PolicyStatus.POLICY_SUCCESS.value
+

src/scanoss/inspection/policy_check/dependency_track/__init__.py

@@ -22,12 +22,12 @@
   THE SOFTWARE.
 """
 
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Any, Callable, Dict, Generic, List, TypeVar
+from typing import Callable, Dict, Generic, List, NamedTuple, TypeVar
 
-from ..scanossbase import ScanossBase
-from .utils.license_utils import LicenseUtil
+from ...scanossbase import ScanossBase
+from ..utils.license_utils import LicenseUtil
 
 
 class PolicyStatus(Enum):
@@ -46,9 +46,13 @@ class PolicyStatus(Enum):
 # End of PolicyStatus Class
 #
 
+class PolicyOutput(NamedTuple):
+    details: str
+    summary: str
+
 T = TypeVar('T')
 
-class PolicyCheck(ScanossBase, Generic[T]):
+class PolicyCheck(ScanossBase, Generic[T], ABC):
     """
     A base class for implementing various software policy checks.
 
@@ -80,7 +84,7 @@ def __init__(  # noqa: PLR0913
         self.output = output
 
     @abstractmethod
-    def run(self):
+    def run(self)-> tuple[int,PolicyOutput]:
         """
         Execute the policy check process.
 
@@ -91,14 +95,14 @@ def run(self):
         3. Formatting the results
         4. Saving the output to files if required
 
-        :return: A tuple containing:
+        :return: A named tuple containing two elements:
                  - First element: PolicyStatus enum value (SUCCESS, FAIL, or ERROR)
-                 - Second element: Dictionary containing the inspection results
+                 - Second element: PolicyOutput A tuple containing the policy results.
         """
         pass
 
     @abstractmethod
-    def _json(self, data: list[T]) -> Dict[str, Any]:
+    def _json(self, data: list[T]) -> PolicyOutput:
         """
         Format the policy checks results as JSON.
         This method should be implemented by subclasses to create a Markdown representation
@@ -112,7 +116,7 @@ def _json(self, data: list[T]) -> Dict[str, Any]:
         pass
 
     @abstractmethod
-    def _markdown(self, data: list[T]) -> Dict[str, Any]:
+    def _markdown(self, data: list[T]) -> PolicyOutput:
         """
         Generate Markdown output for the policy check results.
 
@@ -125,7 +129,7 @@ def _markdown(self, data: list[T]) -> Dict[str, Any]:
         pass
 
     @abstractmethod
-    def _jira_markdown(self, data: list[T]) -> Dict[str, Any]:
+    def _jira_markdown(self, data: list[T]) -> PolicyOutput:
         """
         Generate Markdown output for the policy check results.
 
@@ -137,15 +141,15 @@ def _jira_markdown(self, data: list[T]) -> Dict[str, Any]:
         """
         pass
 
-    def _get_formatter(self) -> Callable[[List[dict]], Dict[str, Any]] or None:
+    def _get_formatter(self) -> Callable[[List[dict]], PolicyOutput]:
         """
         Get the appropriate formatter function based on the specified format.
 
         :return: Formatter function (either _json or _markdown)
         """
         valid_format = self._is_valid_format()
         if not valid_format:
-            return None
+            raise ValueError('Invalid format specified')
         # a map of which format function to return
         function_map = {
             'json': self._json,
@@ -205,14 +209,14 @@ def _generate_formatter_report(self, components: list[Dict]):
         if formatter is None:
             return PolicyStatus.ERROR.value, {}
         # Format the results
-        data = formatter(components)
+        policy_output = formatter(components)
         ## Save outputs if required
-        self.print_to_file_or_stdout(data['details'], self.output)
-        self.print_to_file_or_stderr(data['summary'], self.status)
+        self.print_to_file_or_stdout(policy_output.details, self.output)
+        self.print_to_file_or_stderr(policy_output.summary, self.status)
         # Check to see if we have policy violations
         if len(components) > 0:
-            return PolicyStatus.POLICY_FAIL.value, data
-        return PolicyStatus.POLICY_SUCCESS.value, data
+            return PolicyStatus.POLICY_FAIL.value, policy_output
+        return PolicyStatus.POLICY_SUCCESS.value, policy_output
 #
 # End of PolicyCheck Class
 #