Formating prints according to CERT-7899

Quorum/apis/block_explorers/source_code.py

@@ -47,7 +47,7 @@ def _parse_source_code(self) -> None:
             self._parsed_contract = contract_ast[contract_name]['ast']
 
         except Exception as e:
-            pp.pretty_print(f"Error parsing source code for {self.file_name}: {e}\n"
+            pp.pprint(f"Error parsing source code for {self.file_name}: {e}\n"
                             f"Some of the checks will not apply to this contract!!!",
                             pp.Colors.FAILURE)
         finally:

Quorum/apis/git_api/git_manager.py

@@ -52,12 +52,12 @@ def _load_repos_from_file(self, gt_config: dict[str, any]) -> tuple[dict[str, st
     def __clone_or_update_for_repo(repo_name: str, repo_url: str, to_path: Path):
             repo_path = to_path / repo_name
             if repo_path.exists():
-                pp.pretty_print(f"Repository {repo_name} already exists at {repo_path}. Updating repo and submodules.", pp.Colors.INFO)
+                pp.pprint(f"Repository {repo_name} already exists at {repo_path}. Updating repo and submodules.", pp.Colors.INFO)
                 repo = Repo(repo_path)
                 repo.git.pull()
                 repo.git.submodule('update', '--init', '--recursive')
             else:
-                pp.pretty_print(f"Cloning {repo_name} from URL: {repo_url} to {repo_path}...", pp.Colors.INFO)
+                pp.pprint(f"Cloning {repo_name} from URL: {repo_url} to {repo_path}...", pp.Colors.INFO)
                 Repo.clone_from(repo_url, repo_path, multi_options=["--recurse-submodules"])
 
 
@@ -68,7 +68,7 @@ def clone_or_update(self) -> None:
         If the repository already exists locally, it will update the repository and its submodules.
         Otherwise, it will clone the repository and initialize submodules.
         """
-
+        pp.pprint('Cloning and updating preliminaries', pp.Colors.INFO)
         for repo_name, repo_url in self.repos.items():
            GitManager.__clone_or_update_for_repo(repo_name, repo_url, self.modules_path)
 

Quorum/auto_report/create_report.py

@@ -28,21 +28,21 @@ def parse_args() -> argparse.Namespace:
 
 
 def create_report(proposal_id: int, template: Path, generate_report_path: Path):
-    pprinter.pretty_print(f'Generating a report using template in {template}', pprinter.Colors.INFO)
+    pprinter.pprint(f'Generating a report using template in {template}', pprinter.Colors.INFO)
     env = Environment(loader=FileSystemLoader(template.parent))
     env.globals.update(zip=zip)
     template = env.get_template(template.name)
 
-    pprinter.pretty_print(f'Retrieving tag information for proposal {proposal_id}', pprinter.Colors.INFO)
+    pprinter.pprint(f'Retrieving tag information for proposal {proposal_id}', pprinter.Colors.INFO)
     tags = aave_tags.get_aave_tags(proposal_id)
-    pprinter.pretty_print(f'Tag information retrieved', pprinter.Colors.INFO)
+    pprinter.pprint(f'Tag information retrieved', pprinter.Colors.INFO)
 
     report = template.render(tags)
 
     with open(generate_report_path, 'w') as f:
         f.write(report)
 
-    pprinter.pretty_print(f'Created report at {generate_report_path}.', pprinter.Colors.SUCCESS)
+    pprinter.pprint(f'Created report at {generate_report_path}.', pprinter.Colors.SUCCESS)
 
 
 def main():

Quorum/check_proposal.py

@@ -45,7 +45,7 @@ def load_config(config_path: str) -> dict[str, Any] | None:
             config_data = json.load(file)
         return config_data
     except (FileNotFoundError, json.JSONDecodeError) as e:
-        pp.pretty_print(f"Failed to parse given config file {config_path}:\n{e}", pp.Colors.FAILURE)
+        pp.pprint(f"Failed to parse given config file {config_path}:\n{e}", pp.Colors.FAILURE)
 
 
 def proposals_check(customer: str, chain: Chain, proposal_addresses: list[str], providers: list[PriceFeedProviderBase]) -> None:
@@ -62,38 +62,43 @@ def proposals_check(customer: str, chain: Chain, proposal_addresses: list[str],
     """
     api = ChainAPI(chain)
 
-    pp.pretty_print(f"Processing customer {customer}, for chain: {chain}", pp.Colors.INFO)
     for proposal_address in proposal_addresses:
-        pp.pretty_print(f"Processing proposal {proposal_address}", pp.Colors.INFO)
+        pp.pprint(f'Analyzing payload {proposal_address}', pp.Colors.INFO)
 
         try:
             source_codes = api.get_source_code(proposal_address)
-        except ValueError as e:
+        except ValueError as _:
             error_message = (
-                f"Payload address {proposal_address} is not verified on {chain.name} explorer.\n"
-                "We do not recommend to approve this proposal until the code is approved!\n"
-                "Try contacting the proposer and ask them to verify the contract.\n"
-                "No further checks are being performed on this payload."
+                f'Payload address {proposal_address} is not verified on {chain.name} explorer.\n'
+                'We do not recommend to approve this proposal until the code is approved!\n'
+                'Try contacting the proposer and ask them to verify the contract.\n'
+                'No further checks are being performed on this payload.'
             )
-            pp.pretty_print(error_message, pp.Colors.FAILURE)
+            pp.pprint(error_message, pp.Colors.FAILURE)
             # Skip further checks for this proposal 
             continue
 
         # Diff check
+        pp.pprint('Check 1 - Comparing payload contract and imports with the source of truth', pp.Colors.INFO)
         missing_files = Checks.DiffCheck(customer, chain, proposal_address, source_codes).find_diffs()
-
+        
         # Review diff check
+        pp.pprint(f'Check 2 - Verifying missing files against customer review repo', pp.Colors.INFO)
         Checks.ReviewDiffCheck(customer, chain, proposal_address, missing_files).find_diffs()
 
         # Global variables check
+        pp.pprint('Check 3 - Global variables', pp.Colors.INFO)
         Checks.GlobalVariableCheck(customer, chain, proposal_address, missing_files).check_global_variables()
 
         # Feed price check
+        pp.pprint('Check 4 - Explicit addresses validation', pp.Colors.INFO)
         Checks.PriceFeedCheck(customer, chain, proposal_address, missing_files, providers).verify_price_feed()
 
         # New listing check
+        pp.pprint('Check 5 - First deposit for new listing', pp.Colors.INFO)
         Checks.NewListingCheck(customer, chain, proposal_address, missing_files).new_listing_check()
 
+
 def main() -> None:
     """
     Main function to execute tasks based on command line arguments or JSON configuration.
@@ -106,9 +111,12 @@ def main() -> None:
     if config_data:
         # Multi-task mode using JSON configuration
         for customer, chain_info in config_data.items():
+            pp.pprint('Run Preparation', pp.Colors.INFO)
             ground_truth_config = ConfigLoader.load_customer_config(customer)
             GitManager(customer, ground_truth_config).clone_or_update()
             price_feed_providers = ground_truth_config.get("price_feed_providers", [])
+            pp.pprint('Run Metadata', pp.Colors.INFO)
+            pp.pprint(f'Customer: {customer}\nChains and payloads:\n{chain_info}', pp.Colors.INFO)
             for chain, proposals in chain_info.items():
                 # Validate chain is supported by cast to Chain enum
                 chain = Chain(chain)
@@ -118,10 +126,12 @@ def main() -> None:
         # Single-task mode using command line arguments
         if not (customer and chain and proposal_address):
             raise ValueError("Customer, chain, and proposal_address must be specified if not using a config file.")
-
+        pp.pprint('Run Preparation', pp.Colors.INFO)
         ground_truth_config = ConfigLoader.load_customer_config(customer)
         GitManager(customer, ground_truth_config).clone_or_update()
-        price_feed_providers = ground_truth_config.get("price_feed_providers", [])    
+        price_feed_providers = ground_truth_config.get("price_feed_providers", [])
+        pp.pprint('Run Metadata', pp.Colors.INFO)
+        pp.pprint(f'Customer: {customer}\nChains and payloads:\n{chain}: {proposal_address}', pp.Colors.INFO)
         proposals_check(customer, chain, [proposal_address], price_feed_providers)
 
 

Quorum/checks/diff.py

@@ -123,21 +123,29 @@ def __print_diffs_results(self, missing_files: list[SourceCode], files_with_diff
             missing_files (list[SourceCode]): A list of SourceCode objects representing missing files.
             files_with_diffs (list[Compared]): A list of Compared objects representing files with differences.
         """
-        total_number_of_files = len(self.source_codes)
-        number_of_missing_files = len(missing_files)
-        number_of_files_with_diffs = len(files_with_diffs)
-
-        msg = f"Compared {total_number_of_files - number_of_missing_files}/{total_number_of_files} files for proposal {self.proposal_address}"
-        if number_of_missing_files == 0:
-            pp.pretty_print(msg, pp.Colors.SUCCESS)
-        else:
-            pp.pretty_print(msg, pp.Colors.WARNING)
-            for source_code in missing_files:
-                pp.pretty_print(f"Missing file: {source_code.file_name} in local repo", pp.Colors.WARNING)
-
-        if number_of_files_with_diffs == 0:
-            pp.pretty_print("No differences found.", pp.Colors.SUCCESS)
-        else:
-            pp.pretty_print(f"Found differences in {number_of_files_with_diffs} files", pp.Colors.FAILURE)
-            for compared_pair in files_with_diffs:
-                pp.pretty_print(f"Local: {compared_pair.local_file}\nProposal: {compared_pair.proposal_file}\nDiff: {compared_pair.diff}", pp.Colors.FAILURE)
+        num_total_files = len(self.source_codes)
+        num_missing_files = len(missing_files)
+        num_diff_files = len(files_with_diffs)
+        num_identical = num_total_files - num_missing_files - num_diff_files
+
+        # Identical files message.
+        pp.pprint(f'Files found identical: {num_identical}/{num_total_files}', pp.Colors.SUCCESS)
+
+        # Diffs files message.
+        if num_diff_files > 0:
+            diffs_msg = ('Proposal files found to deviate from their source of truth counterpart: '
+                        f'{num_diff_files}/{num_total_files}\n')
+            for i, compared_pair in enumerate(files_with_diffs, 1):
+                diffs_msg += (f'{i}. Proposal file: {compared_pair.proposal_file}\n'
+                              f'Source of truth file: {compared_pair.local_file}\n'
+                              f'Diff can be found here: {compared_pair.diff}\n')
+            pp.pprint(diffs_msg, pp.Colors.FAILURE)
+
+        # Missing files message.
+        if num_missing_files > 0:
+            missing_msg = ('Proposal files missing from source of truth: '
+                        f'{num_missing_files}/{num_total_files}\n')
+            for i, source_code in enumerate(missing_files, 1):
+                missing_msg += f'{i}. File: {source_code.file_name}\n'
+            pp.pprint(missing_msg, pp.Colors.WARNING)
+

Quorum/checks/global_variables.py

@@ -1,7 +1,5 @@
-import re
 from pathlib import Path
 
-
 from Quorum.checks.check import Check
 from Quorum.apis.block_explorers.source_code import SourceCode
 import Quorum.utils.pretty_printer as pp
@@ -75,11 +73,16 @@ def __process_results(self, source_code_to_violated_variables: dict[str, list[di
                                                                        to lists of violated variables.
         """
         if not source_code_to_violated_variables:
-            pp.pretty_print("All global variables are constant or immutable.", pp.Colors.SUCCESS)
-        else:
-            pp.pretty_print("Global variable checks failed:", pp.Colors.FAILURE)
-            for file_name, violated_variables in source_code_to_violated_variables.items():
-                pp.pretty_print(f"File {file_name} contains variables that are not constant or immutable"
-                                ,pp.Colors.FAILURE)
-                self._write_to_file(Path(file_name).stem.removesuffix(".sol"), violated_variables)
+            pp.pprint('All global variables are constant or immutable.', pp.Colors.SUCCESS)
+            return
+
+        msg = ("Some global variables aren't constant or immutable. A storage collision may occur!\n"
+               f'The following variables found to be storage variables: ')
+        i = 1
+        for file_name, violated_variables in source_code_to_violated_variables.items():
+            for var in violated_variables:
+                msg += f"{i}. File {file_name}: {var['name']}"
+                i += 1
+            self._write_to_file(Path(file_name).stem.removesuffix('.sol'), violated_variables)
+        pp.pprint(msg, pp.Colors.FAILURE)
 

Quorum/checks/new_listing.py

@@ -5,7 +5,6 @@
 
 
 class NewListingCheck(Check):
-
     def new_listing_check(self) -> None:
         """
         Checks if the proposal address is a new listing on the blockchain.
@@ -14,13 +13,14 @@ def new_listing_check(self) -> None:
         no new listings were found.
         """
         functions = self._get_functions_from_source_codes()
-        if functions.get("newListings", functions.get("newListingsCustom")):
-            pp.pretty_print(f"New listings detected for {self.proposal_address}", pp.Colors.WARNING)
+        if functions.get('newListings', functions.get('newListingsCustom')):
+            pp.pprint(f"New listings detected for {self.proposal_address}", pp.Colors.WARNING)
 
             # Check if Anthropic API key is configured
             if not config.ANTHROPIC_API_KEY:
-                pp.pretty_print(
-                    "First deposit check is skipped. If you have a LLM API key, you can add it to your environment variables to enable this check",
+                pp.pprint(
+                    'New listings were detected in payload but first deposit check is skipped.\n'
+                    'If you have a LLM API key, you can add it to your environment variables to enable this check',
                     pp.Colors.WARNING
                 )
                 return
@@ -29,20 +29,22 @@ def new_listing_check(self) -> None:
             proposal_code_str = '\n'.join(proposal_code)
             listings: ListingArray | None = FirstDepositChain().execute(proposal_code_str)
             if listings is None:
-                pp.pretty_print(f"Failed to retrieve new listings for {self.proposal_address}", pp.Colors.FAILURE)
+                pp.pprint('New listings were detected in payload but LLM failed to retrieve them.',
+                                pp.Colors.FAILURE)
                 return
-            for listing in listings.listings:
+
+            msg = f'{len(listings.listings)} new asset listings were detected:\n'
+            for i, listing in enumerate(listings.listings, 1):
                 if listing.approve_indicator and listing.supply_indicator:
-                    pp.pretty_print(
-                        f"New listing detected for {listing}", pp.Colors.SUCCESS
-                    )
+                    msg += f'{i}. New listing detected for {listing.asset_symbol}\n'
                 else:
-                    pp.pretty_print(f"New listing detected for {listing.asset_symbol} but no approval or supply detected", pp.Colors.FAILURE)
-            self._write_to_file("new_listings.json", listings.model_dump())
-
+                    msg += f'{i}. New listing detected for {listing.asset_symbol} but no approval or supply detected\n'
+            pp.pprint(msg, pp.Colors.INFO)
+            self._write_to_file('new_listings.json', listings.model_dump())
+
         else:
-            pp.pretty_print(f"No new listings detected for {self.proposal_address}", pp.Colors.INFO)
-    
+            pp.pprint(f'No new listings detected for {self.proposal_address}', pp.Colors.INFO)
+
     def _get_functions_from_source_codes(self) -> dict:
         """
         Retrieves functions from the source codes.