run-itest.py

import re
import os
import json
import glob
import shutil
import xmltodict
import subprocess
import pandas as pd
from tqdm import tqdm


def coevolve(idx, project_data, output_dir, df, itest_jar, junit_jar, exli_repo_path):

    # Preparation
    project_sha = project_data["given_sha"][:7]
    project_branch = project_data["default_branch"]
    project_folder =  f"{idx:03d}-" + project_data["project_name"].replace("/", "_") + "-" + project_sha + "-" + project_data["line_number_in_given_sha"]
    project_path = os.path.join(output_dir, project_folder)

    # Create the project folder structure
    os.makedirs(project_path, exist_ok=False) # Should be unique
    os.makedirs(os.path.join(project_path, "reports"), exist_ok=False)

    # Clone the project
    code_path = os.path.join(project_path, "code")
    subprocess.run(["git", "clone", project_data["project_url"], code_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    subprocess.run(f"git checkout {project_branch}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # Download required XML files
    exli_project_name = project_data["project_name"].replace("/", "_") + "-" + project_sha
    xml_dir_name = os.path.join(exli_repo_path, "R2-tests", exli_project_name, ".inlinegen", "serialized-data")
    project_xml_dir = os.path.join(code_path, "src", "main", "java", ".inlinegen", "serialized-data")
    if os.path.exists(xml_dir_name):
        shutil.copytree(xml_dir_name, project_xml_dir)

    # Other I-Test required variables
    class_path = project_data["class_path"]
    dep_file_path = os.path.join(project_path, "deps.txt")
    app_src_path = f"{code_path}/src/main/java:{code_path}/src/test/java"
    inlinetest_package_dir = os.path.dirname(class_path)
    select_package = inlinetest_package_dir.replace("./src/main/java/","").replace("/",".")

    # Create commit files
    os.chdir(code_path)
    first_commit_sha = subprocess.run(f'git log -L {project_data["line_number_in_given_sha"]},+1:{class_path} --reverse --pretty=format:"%H" | head -n 1',
        stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, shell=True, text=True).stdout.strip()
    subprocess.run([f"git log --format=%H {first_commit_sha}..{project_sha} | tac > ../commits.txt"], 
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, shell=True) 
    subprocess.run([f"git log --format=%H {project_sha}..HEAD | tac >> ../commits.txt"], 
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, shell=True) 
    os.chdir(output_dir)

    # Commit loop
    for commit_sha, change_sha, line_number, sha_type in get_list_of_changes(project_data):
        commit_sha = commit_sha.strip()
        os.chdir(code_path)

        # Checkout the commit 
        subprocess.run(f"git checkout {commit_sha}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

        if sha_type == "DELETE" and not os.path.exists(class_path):
            # Save results for tables
            df["project_name"].append(project_folder)
            df["SHA_change"].append(change_sha)
            df["SHA_eval"].append(commit_sha)
            df["type"].append(sha_type)
            df["parse_result"].append("deleted")
            df["compile_result"].append("deleted")
            df["running_result"].append("deleted") 
            df["reports_result"].append("deleted")
            df["target_statement"].append("deleted")
            df["STATUS"].append("deleted") 
            continue

        # Add I-test 
        inline_test_code = project_data["inline_test_code"]
        found_target_statement = add_inline_test(class_path, line_number, inline_test_code, project_folder, commit_sha, output_dir)

        # Run I-test
        parse_result, compile_result, running_result = run_inline_tests(project_path, code_path, dep_file_path, app_src_path, 
            class_path, select_package, commit_sha, inlinetest_package_dir, itest_jar, junit_jar, project_folder, output_dir)

        # Save results for tables
        sha_reports_path = os.path.join(project_path, "reports", commit_sha)
        parse_results(project_folder, sha_reports_path, df, sha_type, 
            found_target_statement, parse_result, compile_result, running_result, output_dir, change_sha)

        # Clean repo
        subprocess.run('git clean -df "*.java"', shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        subprocess.run('git clean -df "*.class"', shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    
def get_list_of_changes(project_data):
    changes = []
    
    # Given SHA
    change_sha = project_data["given_sha"]
    commit_sha = change_sha #TODO change_sha
    line_number = project_data["line_number_in_given_sha"]
    sha_type = "GIVEN"
    changes.append((commit_sha, change_sha, int(line_number), sha_type))
    
    for version in project_data["changed_versions"]:
        
        # Before SHA
        if "before_change_stmt" in version:
            change_sha = version["changed_commit"][:7] + "^"
            commit_sha = change_sha #TODO change_sha
            line_number = version["line_number_in_changed_sha"]
            sha_type = "BEFORE"

        # After SHA
        elif "after_change_stmt" in version:
            change_sha = version["changed_commit"][:7]
            commit_sha = change_sha #TODO change_sha
            line_number = version["line_number_in_changed_sha"]
            sha_type = "AFTER"

        # Delete SHA
        elif "delete_stmt" in version:
            change_sha = version["delete_commit"][:7]
            commit_sha = change_sha
            line_number = version["line_number_in_deleted_sha"]
            sha_type = "DELETE"

        changes.append((commit_sha, change_sha, int(line_number), sha_type))

    return changes

def add_inline_test(class_path, line_number, inline_test_code, project_folder, commit_sha, output_dir):

    # Read the contents of the Java file
    with open(class_path, 'r') as file:
        java_code = file.read()

    # Find line number
    lines = java_code.split("\n")
    previous = "\n".join(lines[:line_number-1])
    posterior = "\n".join(lines[line_number-1:])

    # Is an "If" statement?
    is_if_statement = re.search(r"^\s*(}?\s*else)?\s*if\s*\(", posterior.strip())
    if is_if_statement:
        # Does it have curly braces?
        curly_pos = posterior.find("{")
        if posterior[curly_pos-1] == '"' or posterior[curly_pos-1] == "'":
            curly_pos = posterior.find("{", curly_pos+1)
        curly_braces = (
            curly_pos != -1 and 
            curly_pos < posterior.find(";")
        )
        if not curly_braces:
            # Find where the statement begins
            begin = posterior.find("(")
            stack = []
            begin_idx = -1
            for idx, char in enumerate(posterior):
                if idx < begin:
                    continue
                if char == "(":
                    stack.append(idx)
                elif char == ")":
                    stack.pop()
                if not stack:
                    begin_idx = idx + 1
                    break

            if begin_idx == -1:
                raise Exception("Could not find the beginning of the statement")

            # Find where the statement ends
            end_idx = posterior.find(";") + 1
            if end_idx == -1:
                raise Exception("Could not find the end of the statement")

            # Add the curly braces 
            posterior = posterior[:begin_idx] + "{" + posterior[begin_idx:end_idx] + "}" + posterior[end_idx:]

        # Search for the next ocurrence of "{"
        itest_idx = posterior.find("{")
        target_statement = posterior[:itest_idx+1]
        posterior = posterior[itest_idx+1:]

        # Insert the inline test code after the target statement
        modified_code = "\n".join([previous, target_statement, inline_test_code, posterior])

    else:
        # Search for the next ocurrence of ";"
        itest_idx = posterior.find(";")
        target_statement = posterior[:itest_idx+1]
        posterior = posterior[itest_idx+1:]

        # Insert the inline test code after the target statement
        modified_code = "\n".join([previous, target_statement, inline_test_code, posterior])

    # Add imports for inline tests
    package_statement = re.search(r"package [^;]+;", modified_code)
    import_statement = "\nimport org.inlinetest.Here;\nimport static org.inlinetest.Here.group;\n"
    modified_code = modified_code[:package_statement.end()] + import_statement + modified_code[package_statement.end():]

    # Write the modified code back to the Java file and also to logs
    file_name = project_folder + "-" + commit_sha + '.txt'
    write_file(class_path, modified_code)
    write_file(os.path.join(output_dir, "results", "code-files", file_name), modified_code)

    return target_statement

def run_inline_tests(
    project_path,
    code_path,
    dep_file_path,
    app_src_path,
    class_path,
    select_package,
    commit_sha,
    inlinetest_package_dir,
    itest_jar,
    junit_jar,
    project_folder,
    output_dir,
):
    subprocess.run(["mvn", "dependency:build-classpath", f"-Dmdep.outputFile={dep_file_path}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # Parse inline tests
    parse_result = subprocess.run([
        "java", "-cp", itest_jar, "org.inlinetest.InlineTestRunnerSourceCode",
        "--input_file=" + class_path, "--assertion_style=junit",
        "--output_dir=" + inlinetest_package_dir, "--multiple_test_classes=true",
        "--dep_file_path=" + dep_file_path, "--app_src_path=" + app_src_path
    ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) 
    if parse_result.returncode != 0:
        return parse_result.stderr, "PARSE ERROR", "PARSE ERROR"
    else:
        parse_result = ""

    # Note that this does not remove the new generated file, only removes the added inline test in the original file
    subprocess.run("git reset --hard HEAD", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    os.chdir("src/main/java")

    # Save test files to logs
    directory_path = os.path.dirname(class_path.replace("./src/main/java/", ""))
    file_list = os.listdir(directory_path)
    pattern = os.path.basename(class_path).replace(".java", r"_(\d+)Test\.java")
    matching_file = next((file_name for file_name in file_list if re.search(pattern, file_name)), "")
    test_file_path = os.path.join(directory_path, matching_file)
    with open(test_file_path, 'r') as file:
        test_file_content = file.read()
    file_name = project_folder + "-" + commit_sha + '.txt'
    write_file(os.path.join(output_dir, "results", "test-files", file_name), test_file_content) 

    # Compile inline tests
    with open(os.path.join(project_path, "commits.txt"), 'r') as file: 
        commits_list = file.read().strip().split("\n")
        commits_list = [commit.strip()[:7] for commit in commits_list]
    if "^" in commit_sha:
        commit_idx = commits_list.index(commit_sha.replace("^", "")) - 1
    else:
        commit_idx = commits_list.index(commit_sha)
    compile_succesful = False
    while commit_idx < len(commits_list) and not compile_succesful:
        subprocess.run(f"git checkout {commits_list[commit_idx]}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        compile_result = subprocess.run(
            ["javac", "-cp", f".:{itest_jar}:{junit_jar}:{open(dep_file_path).read().strip()}", test_file_path], 
            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True
        )
        if compile_result.returncode == 0:
            compile_succesful = True
            compile_result = ""
        commit_idx += 1
    if not compile_succesful:
        return parse_result, compile_result.stderr, "COMPILE ERROR"

    if commits_list[commit_idx-1] != commit_sha:
        print(f"WARNING: (eval) {commits_list[commit_idx-1]} != {commit_sha} (change) at {project_folder}")
        
    # Run inline tests
    running_result = subprocess.run([
        "java", "-jar", junit_jar, "-cp",
        f".:{itest_jar}:{open(dep_file_path).read().strip()}",
        "--select-package", select_package,
        "--reports-dir", os.path.join(project_path, "reports", commit_sha) 
    ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) 
    if running_result.returncode != 0:
        return parse_result, compile_result, running_result.stderr
    else:
        running_result = ""

    os.chdir(code_path)

    return parse_result, compile_result, running_result

def parse_results(project_folder, sha_reports_path, df, sha_type, found_target_statement, parse_result, compile_result, running_result, output_dir, change_sha):

    # Assume tests were not parsed or compiled correctly
    target_stmt_linenumber = None
    class_name = None
    inline_test_linenumber = None
    reports_result = "failed"

    # If everything went well, it will change accordingly
    xml_path = os.path.join(sha_reports_path, "TEST-junit-jupiter.xml")
    if os.path.exists(xml_path):
        with open(xml_path) as xml_file:
            report = xmltodict.parse(xml_file.read())

        # Tests were parsed and compiled correctly
        if "testsuite" in report and "testcase" in report["testsuite"]:
            test_cases = report["testsuite"]["testcase"]
            if not isinstance(test_cases, list):
                test_cases = [test_cases]
            for test_case in test_cases:
                target_stmt_linenumber = test_case["@classname"].split(".")[-1].split("_")[-1].replace("Test", "")
                class_name = test_case["@classname"].replace(f"_{target_stmt_linenumber}Test", "")
                inline_test_linenumber = (test_case["@name"].replace("testLine", "").replace("()", ""))
                reports_result = "failed" if "failure" in test_case or "error" in test_case else "passed"

    # Save error-logs to files (assumes we're on output directory)
    file_name = project_folder + "-" + sha_reports_path.split("/")[-1] + '.txt'
    if parse_result:
        write_file(os.path.join(output_dir, "results", "parse-errors", file_name), parse_result)
    elif compile_result:
        write_file(os.path.join(output_dir, "results", "compile-errors", file_name), compile_result)
    elif running_result:
        write_file(os.path.join(output_dir, "results", "run-errors", file_name), running_result)

    # Save results for tables
    status = not parse_result and not compile_result and not running_result and reports_result == "passed"
    df["project_name"].append(project_folder)
    df["SHA_change"].append(change_sha)
    df["SHA_eval"].append(sha_reports_path.split("/")[-1])
    df["type"].append(sha_type)
    df["parse_result"].append("passed" if not parse_result else "failed")
    df["compile_result"].append("passed" if not compile_result else "failed")
    df["running_result"].append("passed" if not running_result else "failed")     
    df["reports_result"].append(reports_result)
    df["target_statement"].append(found_target_statement.replace("\n", "\\n").replace("\t", "\\t"))
    df["STATUS"].append("passed" if status else "failed")

def write_file(file_path, content):
    with open(file_path, 'w') as file:
        file.write(content)

def main(output_dir, data_file, itest_jar, junit_jar):
    # Make the path absolute
    output_dir = os.path.abspath(output_dir)

    # Open the JSON file
    with open(data_file, 'r') as file: 

        # Load the JSON data
        data = json.load(file)
        
    # Create the output directory
    os.makedirs(output_dir, exist_ok=False)
    os.makedirs(os.path.join(output_dir, "results"), exist_ok=False)
    os.makedirs(os.path.join(output_dir, "results", "parse-errors"), exist_ok=False) 
    os.makedirs(os.path.join(output_dir, "results", "compile-errors"), exist_ok=False) 
    os.makedirs(os.path.join(output_dir, "results", "run-errors"), exist_ok=False) 
    os.makedirs(os.path.join(output_dir, "results", "code-files"), exist_ok=False)
    os.makedirs(os.path.join(output_dir, "results", "test-files"), exist_ok=False)

    # Load ExLi data
    exli_repo_path = os.path.join(output_dir, "exli")
    subprocess.run(["git", "clone", "git@github.com:EngineeringSoftware/exli.git", exli_repo_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # Tabulate results
    df = {
        "project_name": [],
        "SHA_change": [],
        "SHA_eval": [],
        "type": [],  # before, given, after, delete
        "parse_result": [],
        "compile_result": [],
        "running_result": [],
        "reports_result": [],
        "target_statement": [],
        "STATUS": [],
    }

    # Iterate over each entry in the JSON data
    for idx, entry in enumerate(tqdm(data)):

        if entry["total_versions"] == "0":
            print(f"Skipping entry No. {idx}")
            continue

        try:
            os.chdir(output_dir)
            coevolve(idx, entry, output_dir, df, itest_jar, junit_jar, exli_repo_path)
        except Exception as e:
            print(f"Failed at entry No. {idx}:" + str(e))

        if idx % 10 == 0:
            pd.DataFrame(df).to_excel(os.path.join(output_dir, "results", "results.xlsx"), index=False)
    
    pd.DataFrame(df).to_excel(os.path.join(output_dir, "results", "results.xlsx"), index=False)


main(
    output_dir = "../projects-20", 
    data_file = "changed-result-given-first-20.json",
    itest_jar = os.path.expanduser("~/.m2/repository/org/inlinetest/inlinetest/1.0/inlinetest-1.0.jar"),
    junit_jar = os.path.expanduser("~/.m2/repository/junit/junit/1.9.3/junit-platform-console-standalone-1.9.3.jar"),
)