pit-run

#!/usr/bin/env python3
"""
Script add PIT mutation plugin to maven files and run mutations.
"""

import os
import os.path
import shutil
import subprocess
import sys
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.16.1"
PIT_PLUGINS = [
    (PIT_GROUP_ID, "pitest-junit5-plugin", "1.2.1"),
]
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_plugin(root: XmlElement, configuration: XmlElement) -> None:
    plugin = xml_element("plugin")

    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":
                yield os.path.join(root, file)


def make_file_backup(filename: str) -> str:
    backup_filename = f"{filename}.bak.{os.getpid()}"
    shutil.copyfile(filename, backup_filename)
    return backup_filename


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))

    if skip_pit_in_project:
        return configuration

    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 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")
    if args.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)


def print_details(filename: str) -> None:
    subprocess.call(
        ["pandoc", "--to", "plain", filename], stdout=sys.stdout, stderr=sys.stderr
    )


def open_in_browser(filename: str) -> None:
    subprocess.call([BROWSER_COMMAND, os.path.abspath(filename)])


@contextmanager
def pit_in_poms(project_artifact_id: str | None, globs: list[str]) -> Iterator[None]:
    backups: list[tuple[str, str]] = []
    try:
        for filename in get_pom_files():
            backup_filename = f"{filename}.bak.{os.getpid()}"
            shutil.copyfile(filename, backup_filename)
            backups.append((filename, backup_filename))

            pom_file = PomFile(filename)
            pom_file.configure_pit_plugin(project_artifact_id, globs)
            pom_file.write(filename)

        yield None
    finally:
        for target_filename, backup_filename in backups:
            shutil.move(backup_filename, target_filename)


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__":
    main()