refactor: use gre_ast to determine language

Author: jungs1

src/mutahunter/core/analyzer.py

@@ -1,9 +1,11 @@
-import shlex
+import os
 import subprocess
 import xml.etree.ElementTree as ET
 from importlib import resources
+from shlex import split
 from typing import Any, Dict, List
 
+from grep_ast import filename_to_lang
 from tree_sitter_languages import get_language, get_parser
 
 
@@ -17,7 +19,6 @@ def __init__(self, config: Dict[str, Any]) -> None:
         """
         super().__init__()
         self.config = config
-        self.tree_sitter_parser = get_parser(self.config["language"])
 
         if self.config["coverage_type"] == "cobertura":
             self.file_lines_executed = self.parse_coverage_report_cobertura()
@@ -47,9 +48,9 @@ def parse_coverage_report_lcov(self) -> Dict[str, List[int]]:
                     result[current_file] = []
                 elif line.startswith("DA:") and current_file:
                     parts = line.strip().split(":")[1].split(",")
-                    line_number = int(parts[0])
                     hits = int(parts[1])
                     if hits > 0:
+                        line_number = int(parts[0])
                         result[current_file].append(line_number)
                 elif line.startswith("end_of_record"):
                     current_file = None
@@ -117,28 +118,28 @@ def dry_run(self) -> None:
             Exception: If any tests fail during the dry run.
         """
         test_command = self.config["test_command"]
-        result = subprocess.run(shlex.split(test_command))
+        result = subprocess.run(split(test_command), cwd=os.getcwd())
         if result.returncode != 0:
             raise Exception(
                 "Tests failed. Please ensure all tests pass before running mutation testing."
             )
 
     def get_covered_function_blocks(
-        self, executed_lines: List[int], filename: str
+        self, executed_lines: List[int], source_file_path: str
     ) -> List[Any]:
         """
-        Retrieves covered function blocks based on executed lines and filename.
+        Retrieves covered function blocks based on executed lines and source_file_path.
 
         Args:
             executed_lines (List[int]): List of executed line numbers.
-            filename (str): The name of the file being analyzed.
+            source_file_path (str): The name of the file being analyzed.
 
         Returns:
             List[Any]: A list of covered function blocks.
         """
         covered_function_blocks = []
         covered_function_block_executed_lines = []
-        function_blocks = self.get_function_blocks(filename=filename)
+        function_blocks = self.get_function_blocks(source_file_path=source_file_path)
         for function_block in function_blocks:
             start_point = function_block.start_point
             end_point = function_block.end_point
@@ -152,16 +153,15 @@ def get_covered_function_blocks(
             if any(
                 line in executed_lines for line in range(start_line + 1, end_line + 1)
             ):  # start_line + 1 to exclude the function definition line
-                function_executed_lines = []
-                for line in range(start_line, end_line + 1):
-                    function_executed_lines.append(line - start_line + 1)
-
+                function_executed_lines = [
+                    line - start_line + 1 for line in range(start_line, end_line + 1)
+                ]
                 covered_function_blocks.append(function_block)
                 covered_function_block_executed_lines.append(function_executed_lines)
 
         return covered_function_blocks, covered_function_block_executed_lines
 
-    def get_function_blocks(self, filename: str) -> List[Any]:
+    def get_function_blocks(self, source_file_path: str) -> List[Any]:
         """
         Retrieves function blocks from a given file.
 
@@ -171,12 +171,13 @@ def get_function_blocks(self, filename: str) -> List[Any]:
         Returns:
             List[Any]: A list of function block nodes.
         """
-        with open(filename, "rb") as f:
+        with open(source_file_path, "rb") as f:
             source_code = f.read()
-        function_blocks = self.find_function_blocks_nodes(source_code=source_code)
-        return function_blocks
+        return self.find_function_blocks_nodes(
+            source_file_path=source_file_path, source_code=source_code
+        )
 
-    def check_syntax(self, source_code: str) -> bool:
+    def check_syntax(self, source_file_path: str, source_code: str) -> bool:
         """
         Checks the syntax of the provided source code.
 
@@ -186,11 +187,16 @@ def check_syntax(self, source_code: str) -> bool:
         Returns:
             bool: True if the syntax is correct, False otherwise.
         """
-        tree = self.tree_sitter_parser.parse(bytes(source_code, "utf8"))
+
+        lang = filename_to_lang(source_file_path)
+        parser = get_parser(lang)
+        tree = parser.parse(bytes(source_code, "utf8"))
         root_node = tree.root_node
         return not root_node.has_error
 
-    def find_function_blocks_nodes(self, source_code: bytes) -> List[Any]:
+    def find_function_blocks_nodes(
+        self, source_file_path: str, source_code: bytes
+    ) -> List[Any]:
         """
         Finds function block nodes in the provided source code.
 
@@ -200,10 +206,11 @@ def find_function_blocks_nodes(self, source_code: bytes) -> List[Any]:
         Returns:
             List[Any]: A list of function block nodes.
         """
-        tree = self.tree_sitter_parser.parse(source_code)
-        function_blocks = []
-        lang = self.config["language"]
+        lang = filename_to_lang(source_file_path)
+        parser = get_parser(lang)
         language = get_language(lang)
+
+        tree = parser.parse(source_code)
         # Load the tags queries
         try:
             scm_fname = resources.files(__package__).joinpath(
@@ -219,12 +226,8 @@ def find_function_blocks_nodes(self, source_code: bytes) -> List[Any]:
         captures = query.captures(tree.root_node)
 
         captures = list(captures)
-        for node, tag in captures:
-            if tag == "definition.function" or tag == "definition.method":
-                function_blocks.append(node)
-                # start_byte = node.start_byte
-                # end_byte = node.end_byte
-                # func_code = source_code[start_byte:end_byte].decode("utf-8")
-                # print(f"Function block code: {func_code}")
-
-        return function_blocks
+        return [
+            node
+            for node, tag in captures
+            if tag in ["definition.function", "definition.method"]
+        ]

src/mutahunter/core/hunter.py

@@ -28,46 +28,11 @@ def __init__(self, config: Dict[str, Any]) -> None:
                 - only_mutate_file_paths (List[str]): List of specific files to mutate.
         """
         self.config: Dict[str, Any] = config
-        self.config["language"] = self.determine_language(config["test_file_path"])
         self.mutants: List[Mutant] = []
         self.mutant_report = MutantReport(config=self.config)
         self.analyzer = Analyzer(self.config)
         self.test_runner = TestRunner()
 
-    def determine_language(self, filename: str) -> str:
-        """
-        Determines the programming language based on the file extension. For Tree-Sitter language detection.
-
-        Args:
-            filename (str): The filename to determine the language from.
-
-        Returns:
-            str: The programming language corresponding to the file extension.
-
-        Raises:
-            ValueError: If the file extension is not supported.
-        """
-        ext = filename.split(".")[-1]
-        language_mappings = {
-            "py": "python",
-            "java": "java",
-            "js": "javascript",
-            "ts": "typescript",
-            "c": "c",
-            "cpp": "cpp",
-            "rs": "rust",
-            "go": "go",
-            "php": "php",
-            "rb": "ruby",
-            "swift": "swift",
-            "kt": "kotlin",
-            "tsx": "tsx",
-            "ts": "typescript",
-        }
-        if ext not in language_mappings:
-            raise ValueError(f"Unsupported file extension: {ext}")
-        return language_mappings[ext]
-
     def run(self) -> None:
         """
         Executes the mutation testing process from start to finish.
@@ -109,19 +74,14 @@ def should_skip_file(self, filename: str) -> bool:
                     logger.error(f"File {file_path} does not exist.")
                     raise FileNotFoundError(f"File {file_path} does not exist.")
             # NOTE: Only mutate the files specified in the config.
-            for file_path in self.config["only_mutate_file_paths"]:
-                if file_path == filename:
-                    return False
+            return all(
+                file_path != filename
+                for file_path in self.config["only_mutate_file_paths"]
+            )
+        if filename in self.config["exclude_files"]:
             return True
-        else:
-            if filename in self.config["exclude_files"]:
-                return True
-            if any(
-                keyword in filename
-                for keyword in ["test/", "tests/", "test_", "_test", ".test"]
-            ):
-                return True
-        return False
+        test_keywords = ["test/", "tests/", "test_", "_test", ".test"]
+        return any(keyword in filename for keyword in test_keywords)
 
     def generate_mutations(self) -> Generator[Dict[str, Any], None, None]:
         """
@@ -131,47 +91,41 @@ def generate_mutations(self) -> Generator[Dict[str, Any], None, None]:
             Generator[Dict[str, Any], None, None]: Dictionary containing mutation details.
         """
         all_covered_files = self.analyzer.file_lines_executed.keys()
-        for filename in tqdm(all_covered_files):
-            if self.should_skip_file(filename):
+        for covered_file_path in tqdm(all_covered_files):
+            if self.should_skip_file(covered_file_path):
                 continue
             covered_function_blocks, covered_function_block_executed_lines = (
                 self.analyzer.get_covered_function_blocks(
-                    executed_lines=self.analyzer.file_lines_executed[filename],
-                    filename=filename,
+                    executed_lines=self.analyzer.file_lines_executed[covered_file_path],
+                    source_file_path=covered_file_path,
                 )
             )
             logger.info(
-                f"Total function blocks to mutate: {len(covered_function_blocks)}"
+                f"Total function blocks to mutate for {covered_file_path}: {len(covered_function_blocks)}"
             )
             if not covered_function_blocks:
                 continue
 
-            with open(filename, "rb") as f:
-                source_code = f.read()
-
             for function_block, executed_lines in zip(
                 covered_function_blocks,
                 covered_function_block_executed_lines,
             ):
                 start_byte = function_block.start_byte
                 end_byte = function_block.end_byte
-                function_block_source_code = source_code[start_byte:end_byte].decode(
-                    "utf-8"
-                )
 
                 mutant_generator = MutantGenerator(
                     config=self.config,
                     executed_lines=executed_lines,
                     cov_files=list(all_covered_files),
                     test_file_path=self.config["test_file_path"],
-                    filename=filename,
-                    function_block_source_code=function_block_source_code,
-                    language=self.config["language"],
+                    source_file_path=covered_file_path,
+                    start_byte=start_byte,
+                    end_byte=end_byte,
                 )
 
-                for path, hunk, content in mutant_generator.generate():
+                for _, hunk, content in mutant_generator.generate():
                     yield {
-                        "source_path": filename,
+                        "source_path": covered_file_path,
                         "start_byte": start_byte,
                         "end_byte": end_byte,
                         "hunk": hunk,
@@ -188,7 +142,7 @@ def run_mutation_testing(self) -> None:
                 mutant_id = str(len(self.mutants) + 1)
                 mutant_path = self.prepare_mutant_file(
                     mutant_id=mutant_id,
-                    source_path=mutant_data["source_path"],
+                    source_file_path=mutant_data["source_path"],
                     start_byte=mutant_data["start_byte"],
                     end_byte=mutant_data["end_byte"],
                     mutant_code=mutant_data["mutant_code_snippet"],
@@ -215,7 +169,7 @@ def run_mutation_testing(self) -> None:
     def prepare_mutant_file(
         self,
         mutant_id: str,
-        source_path: str,
+        source_file_path: str,
         start_byte: int,
         end_byte: int,
         mutant_code: str,
@@ -236,12 +190,12 @@ def prepare_mutant_file(
         Raises:
             Exception: If the mutant code has syntax errors.
         """
-        mutant_file_name = f"{mutant_id}_{os.path.basename(source_path)}"
+        mutant_file_name = f"{mutant_id}_{os.path.basename(source_file_path)}"
         mutant_path = os.path.join(
             os.getcwd(), f"logs/_latest/mutants/{mutant_file_name}"
         )
 
-        with open(source_path, "rb") as f:
+        with open(source_file_path, "rb") as f:
             source_code = f.read()
 
         modified_byte_code = (
@@ -250,12 +204,15 @@ def prepare_mutant_file(
             + source_code[end_byte:]
         )
 
-        if self.analyzer.check_syntax(modified_byte_code.decode("utf-8")):
+        if self.analyzer.check_syntax(
+            source_file_path=source_file_path,
+            source_code=modified_byte_code.decode("utf-8"),
+        ):
             with open(mutant_path, "wb") as f:
                 f.write(modified_byte_code)
             return mutant_path
         else:
-            raise Exception("Mutant code has syntax errors.")
+            raise SyntaxError("Mutant code has syntax errors.")
 
     def run_test(self, params: Dict[str, str]) -> Any:
         """