Refactor revert (#267)

* Revert "Fix regression on Jacoco coverage & Corbertura (#264)"

This reverts commit f46980f.

* Revert "Removing bad reference to self.last_coverage_percentages causing necessary failure logs (#262)"

This reverts commit 1b1a9ad.

* Revert "Refactored coverage processor in to class hierarchy (#230)"

This reverts commit 3496069.

* Incrementing version.

Author: EmbeddedDevops1

cover_agent/CoverAgent.py

@@ -199,19 +199,19 @@ def run_test_gen(self, failed_test_runs: List, language: str, test_framework: st
 
             # Check if the desired coverage has been reached
             failed_test_runs, language, test_framework, coverage_report = self.test_validator.get_coverage()
-            if self.test_validator.get_current_coverage() >= (self.test_validator.desired_coverage / 100):
+            if self.test_validator.current_coverage >= (self.test_validator.desired_coverage / 100):
                 break
 
         # Log the final coverage
-        if self.test_validator.get_current_coverage() >= (self.test_validator.desired_coverage / 100):
+        if self.test_validator.current_coverage >= (self.test_validator.desired_coverage / 100):
             self.logger.info(
-                f"Reached above target coverage of {self.test_validator.desired_coverage}% (Current Coverage: {round(self.test_validator.get_current_coverage() * 100, 2)}%) in {iteration_count} iterations."
+                f"Reached above target coverage of {self.test_validator.desired_coverage}% (Current Coverage: {round(self.test_validator.current_coverage * 100, 2)}%) in {iteration_count} iterations."
             )
         elif iteration_count == self.args.max_iterations:
             if self.args.diff_coverage:
-                failure_message = f"Reached maximum iteration limit without achieving desired diff coverage. Current Coverage: {round(self.test_validator.get_current_coverage() * 100, 2)}%"
+                failure_message = f"Reached maximum iteration limit without achieving desired diff coverage. Current Coverage: {round(self.test_validator.current_coverage * 100, 2)}%"
             else:
-                failure_message = f"Reached maximum iteration limit without achieving desired coverage. Current Coverage: {round(self.test_validator.get_current_coverage() * 100, 2)}%"
+                failure_message = f"Reached maximum iteration limit without achieving desired coverage. Current Coverage: {round(self.test_validator.current_coverage * 100, 2)}%"
             if self.args.strict_coverage:
                 # User requested strict coverage (similar to "--cov-fail-under in pytest-cov"). Fail with exist code 2.
                 self.logger.error(failure_message)
@@ -237,11 +237,11 @@ def run_test_gen(self, failed_test_runs: List, language: str, test_framework: st
     def log_coverage(self):
         if self.args.diff_coverage:
             self.logger.info(
-                f"Current Diff Coverage: {round(self.test_validator.get_current_coverage() * 100, 2)}%"
+                f"Current Diff Coverage: {round(self.test_validator.current_coverage * 100, 2)}%"
             )
         else:
             self.logger.info(
-                f"Current Coverage: {round(self.test_validator.get_current_coverage() * 100, 2)}%"
+                f"Current Coverage: {round(self.test_validator.current_coverage * 100, 2)}%"
             )
         self.logger.info(f"Desired Coverage: {self.test_validator.desired_coverage}%")
 

cover_agent/CoverageProcessor.py

@@ -6,6 +6,7 @@
 import re
 
 from cover_agent.AICaller import AICaller
+from cover_agent.CoverageProcessor import CoverageProcessor
 from cover_agent.CustomLogger import CustomLogger
 from cover_agent.FilePreprocessor import FilePreprocessor
 from cover_agent.PromptBuilder import PromptBuilder

cover_agent/UnitTestGenerator.py

@@ -3,17 +3,19 @@
 import json
 import logging
 import os
+import re
 
 from diff_cover.diff_cover_tool import main as diff_cover_main
 
 from cover_agent.AICaller import AICaller
+from cover_agent.CoverageProcessor import CoverageProcessor
 from cover_agent.CustomLogger import CustomLogger
 from cover_agent.FilePreprocessor import FilePreprocessor
 from cover_agent.PromptBuilder import PromptBuilder
 from cover_agent.Runner import Runner
 from cover_agent.settings.config_loader import get_settings
 from cover_agent.utils import load_yaml
-from cover_agent.coverage.processor import process_coverage, CoverageReport, CoverageData
+
 
 class UnitTestValidator:
     def __init__(
@@ -106,6 +108,15 @@ def __init__(
         with open(self.source_file_path, "r") as f:
             self.source_code = f.read()
 
+        # initialize the coverage processor
+        self.coverage_processor = CoverageProcessor(
+            file_path=self.code_coverage_report_path,
+            src_file_path=self.source_file_path,
+            coverage_type=self.coverage_type,
+            use_report_coverage_feature_flag=self.use_report_coverage_feature_flag,
+            diff_coverage_report_path=self.diff_cover_report_path,
+        )
+
     def get_coverage(self):
         """
         Run code coverage and build the prompt to be used for generating tests.
@@ -115,9 +126,6 @@ def get_coverage(self):
         """
         # Run coverage and build the prompt
         self.run_coverage()
-        # Run diff coverage if enabled
-        if self.diff_coverage:
-            self.generate_diff_coverage_report()
         return self.failed_test_runs, self.language, self.testing_framework, self.code_coverage_report
 
     def get_code_language(self, source_file_path: str) -> str:
@@ -288,9 +296,14 @@ def run_coverage(self):
         ), f'Fatal: Error running test command. Are you sure the command is correct? "{self.test_command}"\nExit code {exit_code}. \nStdout: \n{stdout} \nStderr: \n{stderr}'
 
         try:
-            self.current_coverage_report = self.post_process_coverage_report(time_of_test_command)
+            # Process the extracted coverage metrics
+            coverage, coverage_percentages = self.post_process_coverage_report(
+                time_of_test_command
+            )
+            self.current_coverage = coverage
+            self.last_coverage_percentages = coverage_percentages.copy()
             self.logger.info(
-                f"Initial coverage: {round(self.current_coverage_report.total_coverage * 100, 2)}%"
+                f"Initial coverage: {round(self.current_coverage * 100, 2)}%"
             )
 
         except AssertionError as error:
@@ -449,6 +462,7 @@ def validate_test(self, generated_test: dict):
                     if exit_code != 0:
                         break
 
+
                 # Step 3: Check for pass/fail from the Runner object
                 if exit_code != 0:
                     # Test failed, roll back the test file to its original content
@@ -491,9 +505,11 @@ def validate_test(self, generated_test: dict):
 
                 # If test passed, check for coverage increase
                 try:
-                    new_coverage_report = self.post_process_coverage_report(time_of_test_command)
+                    new_percentage_covered, new_coverage_percentages = self.post_process_coverage_report(
+                        time_of_test_command
+                    )
 
-                    if self.current_coverage_report is not None and new_coverage_report.total_coverage <= self.current_coverage_report.total_coverage:
+                    if new_percentage_covered <= self.current_coverage:
                         # Coverage has not increased, rollback the test by removing it from the test file
                         with open(self.test_file_path, "w") as test_file:
                             test_file.write(original_content)
@@ -565,20 +581,20 @@ def validate_test(self, generated_test: dict):
                     additional_imports_lines
                 )  # this is important, otherwise the next test will be inserted at the wrong line
 
-                for key in new_coverage_report.file_coverage:
-                    new_v: CoverageData = new_coverage_report.file_coverage[key]
-                    old_v: CoverageData = self.current_coverage_report.file_coverage[key]
-                    if new_v.coverage > old_v.coverage and key == self.source_file_path.split("/")[-1]:
+                for key in new_coverage_percentages:
+                    if new_coverage_percentages[key] > self.last_coverage_percentages[key] and key == self.source_file_path.split("/")[-1]:
                         self.logger.info(
-                            f"Coverage for provided source file: {key} increased to {round(new_v.coverage * 100, 2)}"
+                            f"Coverage for provided source file: {key} increased from {round(self.last_coverage_percentages[key] * 100, 2)} to {round(new_coverage_percentages[key] * 100, 2)}"
                         )
-                    elif new_v.coverage > old_v.coverage:
+                    elif new_coverage_percentages[key] > self.last_coverage_percentages[key]:
                         self.logger.info(
-                            f"Coverage for non-source file: {key} increased to {round(new_v.coverage * 100, 2)}"
+                            f"Coverage for non-source file: {key} increased from {round(self.last_coverage_percentages[key] * 100, 2)} to {round(new_coverage_percentages[key] * 100, 2)}"
                         )
-                
+                self.current_coverage = new_percentage_covered
+                self.last_coverage_percentages = new_coverage_percentages.copy()
+
                 self.logger.info(
-                    f"Test passed and coverage increased. Current coverage: {round(new_coverage_report.total_coverage * 100, 2)}%"
+                    f"Test passed and coverage increased. Current coverage: {round(new_percentage_covered * 100, 2)}%"
                 )
                 return {
                     "status": "PASS",
@@ -675,20 +691,59 @@ def extract_error_message(self, fail_details):
             logging.error(f"Error extracting error message: {e}")
             return ""
 
-    def post_process_coverage_report(self, time_of_test_command: int):
-        report: CoverageReport = process_coverage(
-            tool_type=self.coverage_type,
-            time_of_test_command=time_of_test_command,
-            report_path=self.code_coverage_report_path,
-            src_file_path=self.source_file_path,
-            is_global_coverage_enabled=self.use_report_coverage_feature_flag,
-            file_pattern=None,
-            diff_coverage_report_path=self.diff_cover_report_path,
-        )
-        self.logger.info(
-            f"coverage: Percentage {round(report.total_coverage * 100, 2)}%"
-        )
-        return report
+    def post_process_coverage_report(self, time_of_test_command):
+        coverage_percentages = {}
+        if self.use_report_coverage_feature_flag:
+            self.logger.info(
+                "Using the report coverage feature flag to process the coverage report"
+            )
+            file_coverage_dict = self.coverage_processor.process_coverage_report(
+                time_of_test_command=time_of_test_command
+            )
+            total_lines_covered = 0
+            total_lines_missed = 0
+            total_lines = 0
+            for key in file_coverage_dict:
+                lines_covered, lines_missed, percentage_covered = (
+                    file_coverage_dict[key]
+                )
+                total_lines_covered += len(lines_covered)
+                total_lines_missed += len(lines_missed)
+                total_lines += len(lines_covered) + len(lines_missed)
+                if key == self.source_file_path:
+                    self.last_source_file_coverage = percentage_covered
+                if key not in coverage_percentages:
+                    coverage_percentages[key] = 0
+                coverage_percentages[key] = percentage_covered
+            try:
+                percentage_covered = total_lines_covered / total_lines
+            except ZeroDivisionError:
+                self.logger.error(f"ZeroDivisionError: Attempting to perform total_lines_covered / total_lines: {total_lines_covered} / {total_lines}.")
+                percentage_covered = 0
+
+            self.logger.info(
+                f"Total lines covered: {total_lines_covered}, Total lines missed: {total_lines_missed}, Total lines: {total_lines}"
+            )
+            self.logger.info(
+                f"coverage: Percentage {round(percentage_covered * 100, 2)}%"
+            )
+        elif self.diff_coverage:
+            self.generate_diff_coverage_report()
+            lines_covered, lines_missed, percentage_covered = (
+                self.coverage_processor.process_coverage_report(
+                    time_of_test_command=time_of_test_command
+                )
+            )
+            self.code_coverage_report = f"Lines covered: {lines_covered}\nLines missed: {lines_missed}\nPercentage covered: {round(percentage_covered * 100, 2)}%"
+        else:
+            lines_covered, lines_missed, percentage_covered = (
+                self.coverage_processor.process_coverage_report(
+                    time_of_test_command=time_of_test_command
+                )
+            )
+            self.code_coverage_report = f"Lines covered: {lines_covered}\nLines missed: {lines_missed}\nPercentage covered: {round(percentage_covered * 100, 2)}%"
+        return percentage_covered, coverage_percentages
+
 
     def generate_diff_coverage_report(self):
         """
@@ -719,4 +774,4 @@ def generate_diff_coverage_report(self):
             self.logger.error(f"Error running diff-cover: {e}")
 
     def get_current_coverage(self):
-            return self.current_coverage_report.total_coverage
+            return self.current_coverage_report.total_coverage