Refactoing pit-run script

1 files changed, 245 insertions, 98 deletions

diff --git a/pit-run b/pit-run
index c51eb3c..683a1a0 100755
--- a/pit-run
+++ b/pit-run
@@ -2,143 +2,290 @@
 """
 Script add PIT mutation plugin to maven files and run mutations.
 """
-import argparse
+
 import os
 import os.path
 import shutil
 import subprocess
 import sys
-import xml.etree.ElementTree as ET
-
-PIT_GROUP_ID = 'org.pitest'
-PIT_ARTIFACT_ID = 'pitest-maven'
-PIT_VERSION = '1.14.4'
-PIT_JUNIT5_ARTIFACT_ID = 'pitest-junit5-plugin'
-PIT_JUNIT5_VERSION = '1.2.0'
-NAMESPACES = {'': 'http://maven.apache.org/POM/4.0.0'}
-for prefix, url in NAMESPACES.items():
-    ET.register_namespace(prefix, url)
-
-MUTATORS = [
-    'STRONGER'
+from argparse import ArgumentParser, Namespace
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Iterator, Optional, Self, Any
+from xml.etree.ElementTree import SubElement, Element, parse, register_namespace
+
+PIT_GROUP_ID = "org.pitest"
+PIT_ARTIFACT_ID = "pitest-maven"
+PIT_VERSION = "1.14.4"
+PIT_PLUGINS = [
+    (PIT_GROUP_ID, "pitest-junit5-plugin", "1.2.0"),
 ]
-OPEN_COMMAND = 'x-www-browser' # or xdg-open
+MUTATORS = ["STRONGER"]
+BROWSER_COMMAND = "x-www-browser"  # or xdg-open
+NAMESPACES = {"": "http://maven.apache.org/POM/4.0.0"}
+
+
+def create_args() -> Namespace:
+    parser = ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "-pl",
+        "--projects",
+        help="Comma-delimited list of specified reactor projects to build instead of all projects (maven options)",
+    )
+    parser.add_argument(
+        "-id", "--artifact-id", help="artifactId of project in which PIT should run"
+    )
+    parser.add_argument(
+        "-am",
+        "--also-make",
+        action="store_true",
+        help="If project list is specified, also build projects required by the list (maven options)",
+    )
+    parser.add_argument(
+        "-q",
+        "--quiet",
+        action="store_true",
+        help="Quiet output - only show errors (maven options)",
+    )
+    parser.add_argument(
+        "-w", "--open-browser", action="store_true", help="Open result in browser"
+    )
+    parser.add_argument("globs", nargs="+")
+    return parser.parse_args()
+
+
+@dataclass(frozen=True)
+class XmlElement:
+    underlying_element: Element
+
+    @property
+    def text(self) -> Optional[str]:
+        return self.underlying_element.text
+
+    def find(self, path: str) -> Optional[Self]:
+        element = self.underlying_element.find(path, NAMESPACES)
+        return None if element is None else XmlElement(element)
+
+    def get_or_create(self, tag: str) -> Self:
+        element = self.find(tag)
+        return (
+            XmlElement(SubElement(self.underlying_element, tag))
+            if element is None
+            else element
+        )
+
+    def add_text_child(self, tag: str, text: Any) -> None:
+        child = xml_text_element(tag, text)
+        self.add_child(child)
+
+    def add_child(self, child: Self) -> None:
+        self.underlying_element.append(child.underlying_element)
+
+
+def xml_element(tag: str) -> XmlElement:
+    element = Element(tag)
+    return XmlElement(element)
+
+
+def xml_text_element(tag: str, text: Any) -> XmlElement:
+    element = Element(tag)
+    element.text = str(text)
+    return XmlElement(element)
+
+
+def create_skip_configuration() -> XmlElement:
+    configuration = xml_element("configuration")
+    configuration.add_child(xml_text_element("skip", "True"))
+    return configuration
+
+
+def create_configuration(globs: list[str]) -> XmlElement:
+    configuration = xml_element("configuration")
+    configuration.add_child(xml_text_element("skip", "False"))
+
+    target_classes = xml_element("targetClasses")
+    target_tests = xml_element("targetTests")
+    for glob in globs:
+        parameter = xml_text_element("param", glob)
+        target_classes.add_child(parameter)
+        target_tests.add_child(parameter)
+
+    mutators = xml_element("mutators")
+    for mutator in MUTATORS:
+        mutators.add_text_child("mutator", mutator)
+
+    output_formats = xml_element("outputFormats")
+    output_formats.add_text_child("outputFormat", "HTML")
+    output_formats.add_text_child("outputFormat", "XML")
 
+    configuration.add_child(target_classes)
+    configuration.add_child(target_tests)
+    configuration.add_child(mutators)
+    configuration.add_child(output_formats)
+    return configuration
 
-def create_args():
-    parser = argparse.ArgumentParser()
-    parser.add_argument('-pl', '--projects')
-    parser.add_argument('-pn', '--project-name')
-    parser.add_argument('-am', '--also-make', action='store_true')
-    parser.add_argument('-q', '--quiet', action='store_true')
-    parser.add_argument('globs', nargs='+')
-    return parser
 
+def create_plugin(root: XmlElement, configuration: XmlElement) -> None:
+    plugin = xml_element("plugin")
 
-def get_pom_files():
-    for root, dirs, files in os.walk('.'):
+    plugin.add_text_child("groupId", PIT_GROUP_ID)
+    plugin.add_text_child("artifactId", PIT_ARTIFACT_ID)
+    plugin.add_text_child("version", PIT_VERSION)
+
+    plugin.add_child(configuration)
+
+    dependencies = xml_element("dependencies")
+    for group_id, artifact_id, version in PIT_PLUGINS:
+        dependency = xml_element("dependency")
+        dependency.add_text_child("groupId", group_id)
+        dependency.add_text_child("artifactId", artifact_id)
+        dependency.add_text_child("version", version)
+
+        dependencies.add_child(dependency)
+
+    plugin.add_child(dependencies)
+
+    root.get_or_create("build").get_or_create("plugins").add_child(plugin)
+
+
+class PomFile:
+    def __init__(self, filename: str) -> None:
+        self.tree = parse(filename)
+        self.root = XmlElement(self.tree.getroot())
+
+    def is_skip_pit_for(self, expected_artifact_id: Optional[str]) -> bool:
+        if expected_artifact_id is None:
+            return False
+
+        artifact_element = self.root.find("./artifactId")
+        return artifact_element is None or artifact_element.text != expected_artifact_id
+
+    def configure_pit_plugin(
+        self, project_artifact_id: Optional[str], globs: list[str]
+    ) -> None:
+        if self.is_skip_pit_for(project_artifact_id):
+            configuration = create_skip_configuration()
+        else:
+            configuration = create_configuration(globs)
+        create_plugin(self.root, configuration)
+
+    def write(self, filename: str) -> None:
+        self.tree.write(filename)
+
+
+def get_pom_files() -> Iterator[str]:
+    for root, _, files in os.walk("."):
         for file in files:
-            if file == 'pom.xml':
+            if file == "pom.xml":
                 yield os.path.join(root, file)
 
 
-def make_file_backup(file):
-    dst = '%s.bak.%d' % (file, os.getpid())
-    shutil.copyfile(file, dst)
-    return file, dst
-
+def make_file_backup(filename: str) -> str:
+    backup_filename = f"{filename}.bak.{os.getpid()}"
+    shutil.copyfile(filename, backup_filename)
+    return backup_filename
 
-def update_pom(file, project, globs):
-    tree = ET.parse(file)
-    root = tree.getroot()
 
-    artifact_id = root.find('./artifactId', NAMESPACES).text
+def create_pit_plugin_configuration(
+    globs: list[str], skip_pit_in_project: bool
+) -> XmlElement:
+    configuration = xml_element("configuration")
+    configuration.add_child(xml_text_element("skip", skip_pit_in_project))
 
-    build = root.find('./build', NAMESPACES)
-    if build is None:
-        build = ET.SubElement(root, 'build')
-    plugins = build.find('./plugins', NAMESPACES)
-    if plugins is None:
-        plugins = ET.SubElement(build, 'plugins')
+    if skip_pit_in_project:
+        return configuration
 
-    plugin = ET.SubElement(plugins, 'plugin')
-    ET.SubElement(plugin, 'groupId').text = PIT_GROUP_ID
-    ET.SubElement(plugin, 'artifactId').text = PIT_ARTIFACT_ID
-    ET.SubElement(plugin, 'version').text = PIT_VERSION
+    target_classes = xml_element("targetClasses")
+    target_tests = xml_element("targetTests")
+    for glob in globs:
+        parameter = xml_text_element("param", glob)
+        target_classes.add_child(parameter)
+        target_tests.add_child(parameter)
 
-    dependencies = ET.SubElement(plugin, 'dependencies')
-    dependency = ET.SubElement(dependencies, 'dependency')
-    ET.SubElement(dependency, 'groupId').text = PIT_GROUP_ID
-    ET.SubElement(dependency, 'artifactId').text = PIT_JUNIT5_ARTIFACT_ID
-    ET.SubElement(dependency, 'version').text = PIT_JUNIT5_VERSION
+    mutators = xml_element("mutators")
+    for mutator in MUTATORS:
+        mutators.add_text_child("mutator", mutator)
 
-    configuration = ET.SubElement(plugin, 'configuration')
+    output_formats = xml_element("outputFormats")
+    output_formats.add_text_child("outputFormat", "HTML")
+    output_formats.add_text_child("outputFormat", "XML")
 
-    if artifact_id == project or project is None:
-        ET.SubElement(configuration, 'skip').text = 'False'
-        target_classes = ET.SubElement(configuration, 'targetClasses')
-        target_tests = ET.SubElement(configuration, 'targetTests')
-        for glob in globs:
-            ET.SubElement(target_classes, 'param').text = glob
-            ET.SubElement(target_tests, 'param').text = glob
-        mutators = ET.SubElement(configuration, 'mutators')
-        for mutator in MUTATORS:
-            ET.SubElement(mutators, 'mutator').text = mutator
-        formats = ET.SubElement(configuration, 'outputFormats')
-        ET.SubElement(formats, 'outputFormat').text = 'HTML'
-        ET.SubElement(formats, 'outputFormat').text = 'XML'
-    else:
-        ET.SubElement(configuration, 'skip').text = 'True'
+    configuration.add_child(target_classes)
+    configuration.add_child(target_tests)
+    configuration.add_child(mutators)
+    configuration.add_child(output_formats)
 
-    tree.write(file)
+    return configuration
 
 
-def run_mutation_coverage(args):
-    command = ['mvn', '--batch-mode', 'test-compile', 'org.pitest:pitest-maven:mutationCoverage']
+def run_mutation_coverage(args: Namespace) -> None:
+    command = [
+        "mvn",
+        "--batch-mode",
+        "test-compile",
+        "org.pitest:pitest-maven:mutationCoverage",
+    ]
     if args.also_make:
-        command.append('--also-make')
+        command.append("--also-make")
     if args.projects:
-        command.append('--projects')
+        command.append("--projects")
         command.append(args.projects)
     if args.quiet:
-        command.append('--quiet')
-    subprocess.call(' '.join(command), shell=True, stdout=sys.stdout, stderr=sys.stderr)
+        command.append("--quiet")
+    subprocess.call(" ".join(command), shell=True, stdout=sys.stdout, stderr=sys.stderr)
 
 
-def print_details(filename):
-    subprocess.call(['pandoc', '--to', 'plain', filename], stdout=sys.stdout, stderr=sys.stderr)
+def print_details(filename: str) -> None:
+    subprocess.call(
+        ["pandoc", "--to", "plain", filename], stdout=sys.stdout, stderr=sys.stderr
+    )
 
 
-def open_result(project):
-    target = os.path.join(project, 'target', 'pit-reports')
-    dirs = os.listdir(target)
-    if dirs:
-        last_dir = sorted(dirs)[-1]
-        index_file = os.path.join(target, last_dir, 'index.html')
-        if os.path.isfile(index_file):
-            print_details(index_file)
-            print('Report:', os.path.abspath(index_file))
-            subprocess.call([OPEN_COMMAND, os.path.abspath(index_file)])
+def open_in_browser(filename: str) -> None:
+    subprocess.call([BROWSER_COMMAND, os.path.abspath(filename)])
 
 
-def update_pom_files(args):
-    backup_files = []
+@contextmanager
+def pit_in_poms(project_artifact_id: str | None, globs: list[str]) -> Iterator[None]:
+    backups: list[tuple[str, str]] = []
     try:
-        for file in get_pom_files():
-            backup_files.append(make_file_backup(file))
-            update_pom(file, args.project_name or args.projects, args.globs)
+        for filename in get_pom_files():
+            backup_filename = f"{filename}.bak.{os.getpid()}"
+            shutil.copyfile(filename, backup_filename)
+            backups.append((filename, backup_filename))
 
-        run_mutation_coverage(args)
-        open_result(args.projects)
+            pom_file = PomFile(filename)
+            pom_file.configure_pit_plugin(project_artifact_id, globs)
+            pom_file.write(filename)
+
+        yield None
     finally:
-        for orig, backup in backup_files:
-            shutil.move(backup, orig)
+        for target_filename, backup_filename in backups:
+            shutil.move(backup_filename, target_filename)
 
 
-def main():
-    args = create_args().parse_args()
-    update_pom_files(args)
+def find_report_filename_for(projects: str) -> Optional[str]:
+    filename = os.path.join(projects, "target", "pit-reports", "index.html")
+    if os.path.isfile(filename):
+        return filename
+    return None
+
+
+def main() -> None:
+    args = create_args()
+
+    for prefix, uri in NAMESPACES.items():
+        register_namespace(prefix, uri)
+    with pit_in_poms(args.artifact_id or args.projects, args.globs):
+        run_mutation_coverage(args)
+
+        report_filename = find_report_filename_for(args.projects)
+        if report_filename:
+            print_details(report_filename)
+            print("Report:", os.path.abspath(report_filename))
+            if args.open_browser:
+                open_in_browser(report_filename)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()