feat: add surviving mutant analysis

Author: jungs1

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = 'mutahunter'
 description = "LLM Mutation Testing for any programming language"
 requires-python = ">= 3.11"
-version = "1.1.9"
+version = "1.2.0"
 dependencies = [
     "tree-sitter==0.21.3",
     'tree_sitter_languages==1.10.2',
@@ -18,6 +18,7 @@ dependencies = [
     'litellm',
     "numpy",
     "scipy",
+    'md2pdf',
 ]
 keywords = ["mutahunter", 'test', "testing", "LLM", 'mutant']
 readme = "README.md"

src/mutahunter/core/controller.py

@@ -5,7 +5,9 @@
 from uuid import uuid4
 
 from tqdm import tqdm
-
+from md2pdf.core import md2pdf
+from jinja2 import Template
+import json
 from mutahunter.core.analyzer import Analyzer
 from mutahunter.core.coverage_processor import CoverageProcessor
 from mutahunter.core.db import MutationDatabase
@@ -23,7 +25,10 @@
 from mutahunter.core.io import FileOperationHandler
 from mutahunter.core.llm_mutation_engine import LLMMutationEngine
 from mutahunter.core.logger import logger
-from mutahunter.core.prompts.mutant_generator import MUTANT_ANALYSIS
+from mutahunter.core.prompts.mutant_generator import (
+    SYSTEM_PROMPT_MUTANT_ANALYSUS,
+    USER_PROMPT_MUTANT_ANALYSIS,
+)
 from mutahunter.core.report import MutantReport
 from mutahunter.core.router import LLMRouter
 from mutahunter.core.runner import MutantTestRunner
@@ -52,8 +57,11 @@ def __init__(
         self.mutant_report = mutant_report
         self.file_handler = file_handler
 
+        self.current_run_id = None
+
     def run(self) -> None:
         start = time.time()
+        self.current_run_id = self.db.start_new_run(self.config.test_command)
         try:
             self.run_coverage_analysis()
         except CoverageAnalysisError as e:
@@ -67,7 +75,7 @@ def run(self) -> None:
             self.generate_report()
         except ReportGenerationError as e:
             logger.error(f"Report generation failed: {str(e)}")
-        # self._run_mutant_analysis()
+        self.run_mutant_analysis()
         logger.info(f"Mutation Testing Ended. Took {round(time.time() - start)}s")
 
     def run_coverage_analysis(self) -> None:
@@ -174,7 +182,7 @@ def process_mutations(
                 mutant_data["error_msg"] = str(e)
 
             # Write complete mutant data to database
-            self.db.add_mutant(file_version_id, mutant_data)
+            self.db.add_mutant(self.current_run_id, file_version_id, mutant_data)
 
     def test_mutant(
         self,
@@ -217,38 +225,37 @@ def generate_report(self) -> None:
             self.mutant_report.generate_report(
                 total_cost=self.router.total_cost,
                 line_rate=self.coverage_processor.line_coverage_rate,
+                run_id=self.current_run_id,
             )
         except Exception as e:
             raise ReportGenerationError(f"Failed to generate report: {str(e)}")
 
-    # def _run_mutant_analysis(self) -> None:
-    #     """
-    #     Runs mutant analysis on the generated mutants.
-    #     """
-    #     survived_mutants = [m for m in self.mutants if m.status == "SURVIVED"]
-
-    #     source_file_paths = []
-    #     for mutant in survived_mutants:
-    #         if mutant.source_path not in source_file_paths:
-    #             source_file_paths.append(mutant.source_path)
-
-    #     src_code_list = []
-    #     for file_path in source_file_paths:
-    #         with open(file_path, "r", encoding="utf-8") as f:
-    #             src_code = f.read()
-    #         src_code_list.append(
-    #             f"## Source File: {file_path}\n```{filename_to_lang(file_path)}\n{src_code}\n```"
-    #         )
-
-    #     prompt = {
-    #         "system": "",
-    #         "user": Template(MUTANT_ANALYSIS).render(
-    #             source_code="\n".join(src_code_list),
-    #             surviving_mutants=survived_mutants,
-    #         ),
-    #     }
-    #     mode_response, _, _ = self.router.generate_response(
-    #         prompt=prompt, streaming=False
-    #     )
-    #     with open("logs/_latest/mutant_analysis.md", "w") as f:
-    #         f.write(mode_response)
+    def run_mutant_analysis(self) -> None:
+        """
+        Runs mutant analysis on the generated mutants.
+        """
+        mutants = self.db.get_survived_mutants_by_run_id(run_id=self.current_run_id)
+        mutants_by_files = {}
+        for mutant in mutants:
+            if mutant["file_path"] not in mutants_by_files:
+                mutants_by_files[mutant["file_path"]] = []
+            else:
+                mutants_by_files[mutant["file_path"]].append(mutant)
+
+        for k, v in mutants_by_files.items():
+            with open(k, "r", encoding="utf-8") as f:
+                src_code = f.read()
+            prompt = {
+                "system": Template(SYSTEM_PROMPT_MUTANT_ANALYSUS).render(),
+                "user": Template(USER_PROMPT_MUTANT_ANALYSIS).render(
+                    source_code=src_code,
+                    surviving_mutants=json.dumps(v, indent=2),
+                ),
+            }
+            mode_response, _, _ = self.router.generate_response(
+                prompt=prompt, streaming=True
+            )
+            md2pdf(
+                pdf_file_path=f"logs/_latest/llm/audit_{str(uuid4())[:4]}.pdf",
+                md_content=mode_response,
+            )

src/mutahunter/core/db.py

@@ -3,6 +3,7 @@
 import sqlite3
 from contextlib import contextmanager
 from typing import Any, Dict, List, Optional, Tuple
+from datetime import datetime
 
 
 class DatabaseError(Exception):
@@ -44,9 +45,20 @@ def create_tables(self):
                     UNIQUE (source_file_id, version_hash)
                 );
 
+                CREATE TABLE IF NOT EXISTS Runs (
+                    id INTEGER PRIMARY KEY,
+                    command_line TEXT NOT NULL,
+                    start_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    end_time TIMESTAMP,
+                    execution_time REAL,
+                    mutation_score REAL,
+                    line_coverage REAL
+                );
+
                 CREATE TABLE IF NOT EXISTS Mutants (
                     id INTEGER PRIMARY KEY,
                     file_version_id INTEGER,
+                    run_id INTEGER,
                     status TEXT,
                     type TEXT,
                     line_number INTEGER,
@@ -57,11 +69,39 @@ def create_tables(self):
                     error_msg TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                     FOREIGN KEY (file_version_id) REFERENCES FileVersions(id)
+                    FOREIGN KEY (run_id) REFERENCES Runs(id)
                 );
             """
             )
             conn.commit()
 
+    def start_new_run(self, command_line: str) -> int:
+        """
+        Start a new run by inserting a record into the Runs table.
+
+        Args:
+            command_line (str): The command line used to start the mutation testing run.
+
+        Returns:
+            int: The ID of the newly created run.
+        """
+        with self.get_connection() as conn:
+            try:
+                cursor = conn.cursor()
+                current_time = datetime.now().isoformat()
+                cursor.execute(
+                    """
+                    INSERT INTO Runs (command_line, start_time)
+                    VALUES (?, ?)
+                """,
+                    (command_line, current_time),
+                )
+                conn.commit()
+                return cursor.lastrowid
+            except sqlite3.Error as e:
+                conn.rollback()
+                raise DatabaseError(f"Failed to start new run: {str(e)}")
+
     def get_file_version(self, file_path: str) -> Tuple[int, int, bool]:
         """
         Get or create a file version for the given file path.
@@ -114,18 +154,19 @@ def get_file_version(self, file_path: str) -> Tuple[int, int, bool]:
                 conn.rollback()
                 raise DatabaseError(f"Error processing file version: {str(e)}")
 
-    def add_mutant(self, file_version_id: int, mutant_data: dict):
+    def add_mutant(self, run_id: int, file_version_id: int, mutant_data: dict):
         with self.get_connection() as conn:
             cursor = conn.cursor()
             try:
                 cursor.execute(
                     """
                     INSERT INTO Mutants (
-                        file_version_id, status, type, line_number, 
+                        run_id, file_version_id, status, type, line_number, 
                         original_code, mutated_code, description, mutant_path, error_msg
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                     (
+                        run_id,
                         file_version_id,
                         mutant_data["status"],
                         mutant_data["type"],
@@ -223,6 +264,44 @@ def update_mutant_status(self, mutant_id: int, status: str):
                 conn.rollback()
                 raise DatabaseError(f"Error updating mutant status: {str(e)}")
 
+    def get_survived_mutants_by_run_id(self, run_id: int) -> List[Dict[str, Any]]:
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute(
+                    """
+                    SELECT 
+                        m.id,
+                        sf.file_path,
+                        m.line_number,
+                        m.mutant_path,
+                        m.original_code,
+                        m.mutated_code,
+                        m.description,
+                        m.type
+                    FROM Mutants m
+                    JOIN FileVersions fv ON m.file_version_id = fv.id
+                    JOIN SourceFiles sf ON fv.source_file_id = sf.id
+                    WHERE m.run_id = ? AND m.status = 'SURVIVED'
+                    """,
+                    (run_id,),
+                )
+                return [
+                    {
+                        "id": row[0],
+                        "file_path": row[1],
+                        "line_number": row[2],
+                        "mutant_path": row[3],
+                        "original_code": row[4],
+                        "mutated_code": row[5],
+                        "description": row[6],
+                        "type": row[7],
+                    }
+                    for row in cursor.fetchall()
+                ]
+            except sqlite3.Error as e:
+                raise DatabaseError(f"Error fetching survived mutants: {str(e)}")
+
     def get_survived_mutants(self, source_file_path):
         with self.get_connection() as conn:
             cursor = conn.cursor()
@@ -324,7 +403,50 @@ def get_file_mutations(self, file_name: str) -> List[Dict[str, Any]]:
             except sqlite3.Error as e:
                 raise DatabaseError(f"Error fetching file mutations: {str(e)}")
 
-    def get_file_data(self) -> List[Dict[str, Any]]:
+    def get_mutant_summary(self, run_id) -> Dict[str, int]:
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute(
+                    """
+                    SELECT 
+                        COUNT(*) as total_mutants,
+                        SUM(CASE WHEN status = 'KILLED' THEN 1 ELSE 0 END) as killed_mutants,
+                        SUM(CASE WHEN status = 'SURVIVED' THEN 1 ELSE 0 END) as survived_mutants,
+                        SUM(CASE WHEN status = 'TIMEOUT' THEN 1 ELSE 0 END) as timeout_mutants,
+                        SUM(CASE WHEN status = 'COMPILE_ERROR' THEN 1 ELSE 0 END) as compile_error_mutants,
+                        SUM(CASE WHEN status = 'SYNTAX_ERROR' THEN 1 ELSE 0 END) as syntax_error_mutants,
+                        SUM(CASE WHEN status = 'UNEXPECTED_TEST_ERROR' THEN 1 ELSE 0 END) as unexpected_test_error_mutants
+                    FROM Mutants
+                    WHERE run_id = ?
+                """,
+                    (run_id,),
+                )
+                result = cursor.fetchone()
+                if result[0] == 0:
+                    return None
+
+                else:
+                    valid_mutants = (
+                        result[0] - result[3] - result[4] - result[5] - result[6]
+                    )
+                    mutation_coverage = (
+                        result[1] / valid_mutants if valid_mutants > 0 else 0.0
+                    )
+                return {
+                    "total_mutants": result[0],
+                    "killed_mutants": result[1],
+                    "survived_mutants": result[2],
+                    "timeout_mutants": result[3],
+                    "compile_error_mutants": result[4],
+                    "syntax_error_mutants": result[5],
+                    "unexpected_test_error_mutants": result[6],
+                    "mutation_coverage": mutation_coverage,
+                }
+            except sqlite3.Error as e:
+                raise DatabaseError(f"Error fetching mutant summary: {str(e)}")
+
+    def get_file_data(self, run_id: int) -> List[Dict[str, Any]]:
         with self.get_connection() as conn:
             cursor = conn.cursor()
             try:
@@ -335,12 +457,18 @@ def get_file_data(self) -> List[Dict[str, Any]]:
                         sf.file_path,
                         COUNT(m.id) as total_mutants,
                         SUM(CASE WHEN m.status = 'KILLED' THEN 1 ELSE 0 END) as killed_mutants,
-                        SUM(CASE WHEN m.status = 'SURVIVED' THEN 1 ELSE 0 END) as survived_mutants
+                        SUM(CASE WHEN m.status = 'SURVIVED' THEN 1 ELSE 0 END) as survived_mutants,
+                        SUM(CASE WHEN m.status = 'TIMEOUT' THEN 1 ELSE 0 END) as timeout_mutants,
+                        SUM(CASE WHEN m.status = 'COMPILE_ERROR' THEN 1 ELSE 0 END) as compile_error_mutants,
+                        SUM(CASE WHEN m.status = 'SYNTAX_ERROR' THEN 1 ELSE 0 END) as syntax_error_mutants,
+                        SUM(CASE WHEN m.status = 'UNEXPECTED_TEST_ERROR' THEN 1 ELSE 0 END) as unexpected_test_error_mutants
                     FROM SourceFiles sf
                     JOIN FileVersions fv ON sf.id = fv.source_file_id
                     JOIN Mutants m ON fv.id = m.file_version_id
+                    WHERE m.run_id = ?
                     GROUP BY sf.file_path
-                """
+                    """,
+                    (run_id,),
                 )
                 return [
                     {
@@ -350,41 +478,18 @@ def get_file_data(self) -> List[Dict[str, Any]]:
                         "mutationCoverage": (
                             f"{(row[3] / row[2] * 100):.2f}" if row[2] > 0 else "0.00"
                         ),
+                        "killedMutants": row[3],
                         "survivedMutants": row[4],
+                        "timeoutMutants": row[5],
+                        "compileErrorMutants": row[6],
+                        "syntaxErrorMutants": row[7],
+                        "unexpectedTestErrorMutants": row[8],
                     }
                     for row in cursor.fetchall()
                 ]
             except sqlite3.Error as e:
                 raise DatabaseError(f"Error fetching file data: {str(e)}")
 
-    def get_mutant_summary(self) -> Dict[str, int]:
-        with self.get_connection() as conn:
-            cursor = conn.cursor()
-            try:
-                cursor.execute(
-                    """
-                    SELECT 
-                        COUNT(*) as total_mutants,
-                        SUM(CASE WHEN status = 'KILLED' THEN 1 ELSE 0 END) as killed_mutants,
-                        SUM(CASE WHEN status = 'SURVIVED' THEN 1 ELSE 0 END) as survived_mutants,
-                        SUM(CASE WHEN status = 'TIMEOUT' THEN 1 ELSE 0 END) as timeout_mutants,
-                        SUM(CASE WHEN status = 'COMPILE_ERROR' THEN 1 ELSE 0 END) as compile_error_mutants
-                    FROM Mutants
-                """
-                )
-                result = cursor.fetchone()
-                if result[0] == 0:
-                    return None
-                return {
-                    "total_mutants": result[0],
-                    "killed_mutants": result[1],
-                    "survived_mutants": result[2],
-                    "timeout_mutants": result[3],
-                    "compile_error_mutants": result[4],
-                }
-            except sqlite3.Error as e:
-                raise DatabaseError(f"Error fetching mutant summary: {str(e)}")
-
     def remove_mutants_by_file_version_id(self, file_version_id: int):
         with self.get_connection() as conn:
             cursor = conn.cursor()