refactor: show dependency analysis with rich ui

Demolus13 · Demolus13 · commit 3ac708479d5b · 2025-10-21T10:46:40.000+05:30
Signed-off-by: Demolus13 &lt;parth.govale@oracle.com&gt;
diff --git a/src/macaron/console.py b/src/macaron/console.py
@@ -17,6 +17,174 @@
 from rich.table import Table
 
 
+class Dependency:
+    """A class to manage the display of dependency analysis in the console."""
+
+    def __init__(self) -> None:
+        """Initialize the Dependency instance with default values and tables."""
+        self.description_table = Table(show_header=False, box=None)
+        self.description_table_content: dict[str, str | Status] = {
+            "Package URL:": Status("[green]Processing[/]"),
+            "Local Cloned Path:": Status("[green]Processing[/]"),
+            "Remote Path:": Status("[green]Processing[/]"),
+            "Branch:": Status("[green]Processing[/]"),
+            "Commit Hash:": Status("[green]Processing[/]"),
+            "Commit Date:": Status("[green]Processing[/]"),
+            "CI Services:": Status("[green]Processing[/]"),
+            "Build Tools:": Status("[green]Processing[/]"),
+        }
+        self.progress = Progress(
+            TextColumn(" RUNNING ANALYSIS"),
+            BarColumn(bar_width=None, complete_style="green"),
+            MofNCompleteColumn(),
+        )
+        self.task_id: TaskID
+        self.progress_table = Table(show_header=False, box=None)
+        self.checks: dict[str, str] = {}
+        self.failed_checks_table = Table(show_header=False, box=None)
+        self.summary_table = Table(show_header=False, box=None)
+
+    def add_description_table_content(self, key: str, value: str | Status) -> None:
+        """
+        Add or update a key-value pair in the description table.
+
+        Parameters
+        ----------
+        key : str
+            The key to be added or updated.
+        value : str or Status
+            The value associated with the key.
+        """
+        self.description_table_content[key] = value
+        description_table = Table(show_header=False, box=None)
+        description_table.add_column("Details", justify="left")
+        description_table.add_column("Value", justify="left")
+        for field, content in self.description_table_content.items():
+            description_table.add_row(field, content)
+
+        self.description_table = description_table
+
+    def no_of_checks(self, value: int) -> None:
+        """
+        Initialize the progress bar with the total number of checks.
+
+        Parameters
+        ----------
+        value : int
+            The total number of checks to be performed.
+        """
+        self.task_id = self.progress.add_task("analyzing", total=value)
+
+    def remove_progress_bar(self) -> None:
+        """Remove the progress bar from the display."""
+        self.progress.remove_task(self.task_id)
+
+    def update_checks(self, check_id: str, status: str = "RUNNING") -> None:
+        """
+        Update the status of a specific check and refresh the progress table.
+
+        Parameters
+        ----------
+        check_id : str
+            The identifier of the check to be updated.
+        status : str, optional
+            The new status of the check, by default "RUNNING"
+        """
+        self.checks[check_id] = status
+
+        progress_table = Table(show_header=False, box=None)
+        progress_table.add_column("Status", justify="left")
+        progress_table.add_column("Check", justify="left")
+
+        for check_name, check_status in self.checks.items():
+            if check_status == "RUNNING":
+                progress_table.add_row(Status("[bold green]RUNNING[/]"), check_name)
+        self.progress_table = progress_table
+
+        if self.task_id is not None and status != "RUNNING":
+            self.progress.update(self.task_id, advance=1)
+
+    def update_checks_summary(self, checks_summary: dict, total_checks: int) -> None:
+        """
+        Update the summary tables with the results of the checks.
+
+        Parameters
+        ----------
+        checks_summary : dict
+            Dictionary containing lists of checks categorized by their results.
+        total_checks : int
+            The total number of checks.
+        """
+        failed_checks_table = Table(show_header=False, box=None)
+        failed_checks_table.add_column("Status", justify="left")
+        failed_checks_table.add_column("Check ID", justify="left")
+        failed_checks_table.add_column("Description", justify="left")
+
+        failed_checks = checks_summary["FAILED"]
+        for check in failed_checks:
+            failed_checks_table.add_row(
+                "[bold red]FAILED[/]",
+                check.check.check_id,
+                check.check.check_description,
+            )
+
+        self.failed_checks_table = failed_checks_table
+
+        summary_table = Table(show_header=False, box=None)
+        summary_table.add_column("Check Result Type", justify="left")
+        summary_table.add_column("Count", justify="left")
+        summary_table.add_row("Total Checks", str(total_checks), style="white")
+
+        for check_result_type, checks in checks_summary.items():
+            if check_result_type == "PASSED":
+                summary_table.add_row("PASSED", str(len(checks)), style="green")
+            if check_result_type == "FAILED":
+                summary_table.add_row("FAILED", str(len(checks)), style="red")
+            if check_result_type == "SKIPPED":
+                summary_table.add_row("SKIPPED", str(len(checks)), style="yellow")
+            if check_result_type == "DISABLED":
+                summary_table.add_row("DISABLED", str(len(checks)), style="bright_blue")
+            if check_result_type == "UNKNOWN":
+                summary_table.add_row("UNKNOWN", str(len(checks)), style="white")
+
+        self.summary_table = summary_table
+
+    def make_layout(self) -> list[RenderableType]:
+        """
+        Create the layout for the live console display.
+
+        Returns
+        -------
+        list[RenderableType]
+            A list of rich RenderableType objects containing the layout for the live console display.
+        """
+        layout: list[RenderableType] = []
+        if self.description_table.row_count > 0:
+            layout = layout + [
+                "",
+                self.description_table,
+            ]
+        if self.progress_table.row_count > 0:
+            layout = layout + ["", self.progress, "", self.progress_table]
+        if self.failed_checks_table.row_count > 0:
+            layout = layout + [
+                "",
+                Rule(" SUMMARY", align="left"),
+                "",
+                self.failed_checks_table,
+            ]
+            if self.summary_table.row_count > 0:
+                layout = layout + ["", self.summary_table]
+        elif self.summary_table.row_count > 0:
+            layout = layout + [
+                "",
+                Rule(" SUMMARY", align="left"),
+                "",
+                self.summary_table,
+            ]
+        return layout
+
+
 class RichConsoleHandler(RichHandler):
     """A rich console handler for logging with rich formatting and live updates."""
 
@@ -67,6 +235,8 @@ def __init__(self, *args: Any, verbose: bool = False, **kwargs: Any) -> None:
             "Dependencies Report": "Not Generated",
             "JSON Report": "Not Generated",
         }
+        self.if_dependency: bool = False
+        self.dependency_analysis_list: list[Dependency] = []
         self.components_violates_table = Table(box=None)
         self.components_satisfy_table = Table(box=None)
         self.policy_summary_table = Table(show_header=False, box=None)
@@ -138,6 +308,11 @@ def add_description_table_content(self, key: str, value: str | Status) -> None:
         value : str or Status
             The value associated with the key.
         """
+        if self.if_dependency and self.dependency_analysis_list:
+            dependency = self.dependency_analysis_list[-1]
+            dependency.add_description_table_content(key, value)
+            return
+
         self.description_table_content[key] = value
         description_table = Table(show_header=False, box=None)
         description_table.add_column("Details", justify="left")
@@ -156,8 +331,20 @@ def no_of_checks(self, value: int) -> None:
         value : int
             The total number of checks to be performed.
         """
+        if self.if_dependency and self.dependency_analysis_list:
+            dependency = self.dependency_analysis_list[-1]
+            dependency.no_of_checks(value)
+            return
         self.task_id = self.progress.add_task("analyzing", total=value)
 
+    def remove_progress_bar(self) -> None:
+        """Remove the progress bar from the display."""
+        if self.if_dependency and self.dependency_analysis_list:
+            dependency = self.dependency_analysis_list[-1]
+            dependency.remove_progress_bar()
+            return
+        self.progress.remove_task(self.task_id)
+
     def update_checks(self, check_id: str, status: str = "RUNNING") -> None:
         """
         Update the status of a specific check and refresh the progress table.
@@ -169,6 +356,10 @@ def update_checks(self, check_id: str, status: str = "RUNNING") -> None:
         status : str, optional
             The new status of the check, by default "RUNNING"
         """
+        if self.if_dependency and self.dependency_analysis_list:
+            dependency = self.dependency_analysis_list[-1]
+            dependency.update_checks(check_id, status)
+            return
         self.checks[check_id] = status
 
         progress_table = Table(show_header=False, box=None)
@@ -194,6 +385,10 @@ def update_checks_summary(self, checks_summary: dict, total_checks: int) -> None
         total_checks : int
             The total number of checks.
         """
+        if self.if_dependency and self.dependency_analysis_list:
+            dependency = self.dependency_analysis_list[-1]
+            dependency.update_checks_summary(checks_summary, total_checks)
+            return
         failed_checks_table = Table(show_header=False, box=None)
         failed_checks_table.add_column("Status", justify="left")
         failed_checks_table.add_column("Check ID", justify="left")
@@ -249,6 +444,20 @@ def update_report_table(self, report_type: str, report_path: str) -> None:
 
         self.report_table = report_table
 
+    def is_dependency(self, value: bool) -> None:
+        """
+        Update the flag indicating whether the analyzed package is a dependency.
+
+        Parameters
+        ----------
+        value : bool
+            True if the package is a dependency, False otherwise.
+        """
+        self.if_dependency = value
+        if self.if_dependency:
+            dependency = Dependency()
+            self.dependency_analysis_list.append(dependency)
+
     def generate_policy_summary_table(self) -> None:
         """Generate the policy summary table based on the current policy summary data."""
         policy_summary_table = Table(show_header=False, box=None)
@@ -398,7 +607,11 @@ def make_layout(self) -> Group:
             layout = layout + [error_log_panel]
         if self.command == "analyze":
             if self.description_table.row_count > 0:
-                layout = layout + [Rule(" DESCRIPTION", align="left"), "", self.description_table]
+                layout = layout + [
+                    Rule(" DESCRIPTION", align="left"),
+                    "",
+                    self.description_table,
+                ]
             if self.progress_table.row_count > 0:
                 layout = layout + ["", self.progress, "", self.progress_table]
             if self.failed_checks_table.row_count > 0:
@@ -425,6 +638,17 @@ def make_layout(self) -> Group:
                     layout = layout + [
                         self.report_table,
                     ]
+            if self.if_dependency and self.dependency_analysis_list:
+                for idx, dependency in enumerate(self.dependency_analysis_list, start=1):
+                    dependency_layout = dependency.make_layout()
+                    layout = (
+                        layout
+                        + [
+                            "",
+                            Rule(f" DEPENDENCY {idx}", align="left"),
+                        ]
+                        + dependency_layout
+                    )
         elif self.command == "verify-policy":
             if self.policy_summary_table.row_count > 0:
                 if self.components_satisfy_table.row_count > 0:
diff --git a/src/macaron/output_reporter/results.py b/src/macaron/output_reporter/results.py
@@ -9,7 +9,6 @@
 from typing import Generic, TypedDict, TypeVar
 
 from macaron.config.target_config import Configuration
-from macaron.console import access_handler
 from macaron.output_reporter.scm import SCMStatus
 from macaron.slsa_analyzer.analyze_context import AnalyzeContext
 from macaron.slsa_analyzer.checks.check_result import CheckResultType
@@ -200,7 +199,6 @@ def __init__(self, root_record: Record) -> None:
         self.record_mapping: dict[str, Record] = {}
         if root_record.context:
             self.record_mapping[root_record.record_id] = root_record
-        self.rich_handler = access_handler.get_handler()
 
     def get_records(self) -> Iterable[Record]:
         """Get the generator for all records in the report.
@@ -299,7 +297,6 @@ def __str__(self) -> str:
         """Return the string representation of the Report instance."""
         ctx_list = list(self.get_ctxs())
         main_ctx: AnalyzeContext = ctx_list.pop(0)
-        self.rich_handler = access_handler.get_handler()
 
         output = "".join(
             [
@@ -309,7 +306,6 @@ def __str__(self) -> str:
                 "\nSLSA REQUIREMENT RESULTS:\n",
             ]
         )
-        self.rich_handler.update_checks_summary(main_ctx.get_check_summary(), len(main_ctx.check_results))
 
         slsa_req_mesg: dict[SLSALevels, list[str]] = {level: [] for level in SLSALevels if level != SLSALevels.LEVEL0}
         for req_name, req_status in main_ctx.ctx_data.items():
diff --git a/src/macaron/slsa_analyzer/analyzer.py b/src/macaron/slsa_analyzer/analyzer.py
@@ -240,6 +240,10 @@ def run(
 
                 # Create a report instance with the record of the main repo.
                 report = Report(main_record)
+                self.rich_handler.update_checks_summary(
+                    main_record.context.get_check_summary(),
+                    len(main_record.context.check_results),
+                )
 
                 duplicated_scm_records: list[Record] = []
 
@@ -257,8 +261,14 @@ def run(
                             report.add_dep_record(dep_record)
                             duplicated_scm_records.append(dep_record)
                             continue
+                        self.rich_handler.is_dependency(True)
                         dep_record = self.run_single(config, analysis, report.record_mapping)
                         report.add_dep_record(dep_record)
+                        if dep_record.context:
+                            self.rich_handler.update_checks_summary(
+                                dep_record.context.get_check_summary(),
+                                len(dep_record.context.check_results),
+                            )
                 else:
                     logger.info("Found no dependencies to analyze.")
 
diff --git a/src/macaron/slsa_analyzer/registry.py b/src/macaron/slsa_analyzer/registry.py
@@ -519,6 +519,8 @@ def scan(self, target: AnalyzeContext) -> dict[str, CheckResult]:
                 logger.info("Check %s has failed.", check_id)
                 return results
 
+        self.rich_handler.remove_progress_bar()
+
         return results
 
     def prepare(self) -> bool:
