fix pylint for evaluation

Author: EYH0602

.github/workflows/pylint.yml

@@ -5,9 +5,7 @@ on: [pull_request]
 jobs:
   build:
     runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        python-version: ["3.10.12"]
+
     steps:
       - uses: actions/checkout@v2
         with:

evaluation/ablation_n_fuzz.py

@@ -1,6 +1,4 @@
 import matplotlib.pyplot as plt
-import pandas as pd
-import numpy as np
 import fire
 
 plt.style.use("_mpl-gallery")
@@ -10,6 +8,7 @@
 
 
 def barplot(ns, xs, name):
+    """make a bar plot for n_fuzz experiment"""
     plt.figure(figsize=(12, 6))
     plt.bar(ns, xs, color="skyblue", width=10)
     plt.plot(ns, xs, color="red", marker="o")
@@ -22,6 +21,7 @@ def barplot(ns, xs, name):
 
 
 def main():
+    """script to make RQ4 plots"""
     ns = [20 * i for i in range(7)]
     cov = [17.64, 20.94, 18.69, 19.39, 19.31, 21.69, 21.50]
     acc = [34.94, 35.85, 36.93, 37.01, 37.93, 37.44, 39.21]

evaluation/compile.py

@@ -1,10 +1,11 @@
 from typing import Iterable
-import fire
 import os
-from tree_sitter.binding import Node
+
+from tree_sitter import Node
+from funcy import mapcat
+
 from UniTSyn.frontend.parser import RUST_LANGUAGE
 from UniTSyn.frontend.parser.ast_util import ASTUtil
-from funcy import mapcat
 
 
 def flatten_use_delc(use_delc_code: str) -> list[str]:
@@ -84,7 +85,7 @@ def collect_rs_files(root: str):
                 yield os.path.join(dirpath, filename)
 
 
-def construct_use_delcs(workspace_dir: str, type: str) -> set[str]:
+def construct_use_delcs(workspace_dir: str, test_type: str) -> set[str]:
     """construct a set of unique use_list for a project from all use declarations in
     a subdirectory to
 
@@ -93,12 +94,12 @@ def construct_use_delcs(workspace_dir: str, type: str) -> set[str]:
 
     Args:
         workspace_dir (str): path to project's workdir
-        type (str): tests or fuzz to collect use_delcs.
+        test_type (str): tests or fuzz to collect use_delcs.
 
     Returns:
         set[str]: set of use declarations to write to generated test files
     """
-    subdir = os.path.join(workspace_dir, type)
+    subdir = os.path.join(workspace_dir, test_type)
 
     def get_use_list_from_file(fpath: str) -> Iterable[str]:
         with open(fpath) as f:
@@ -129,15 +130,3 @@ def write_tests_to_workspace(workspace_dir: str, tests: list[str], test_type: st
         p = os.path.join(workspace_dir, "tests", f"generated_{test_type}_{i}.rs")
         with open(p, "w") as f:
             f.write(use_delc + "\n\n" + "#[test]\n" + test)
-
-
-def main():
-    workspace_dir = os.path.abspath(
-        "data/repos/marshallpierce-rust-base64/marshallpierce-rust-base64-4ef33cc"
-    )
-
-    print(construct_use_delcs(workspace_dir, "tests"))
-
-
-if __name__ == "__main__":
-    fire.Fire(main)

evaluation/coverage.py

@@ -2,25 +2,27 @@
 
 import json
 from typing import Iterable
-import fire
 import os
 import subprocess
-from returns.maybe import Maybe, Some, Nothing
+
+from funcy import mapcat
+from returns.maybe import Nothing
 from returns.result import Result, Success, Failure
 from UniTSyn.frontend.parser import RUST_LANGUAGE
 from UniTSyn.frontend.parser.ast_util import ASTUtil
 from UniTSyn.frontend.rust.rust_util import get_test_functions
 from UniTSyn.frontend.rust.collect_all import collect_test_files
 from evaluation.util import CovError
-from funcy import mapcat
 
 
 def clean_workspace(workspace_dir: str):
+    """clean-up a workspace by removing coverage script and target dir"""
     subprocess.run(["rm", "rust_test_coverage.sh"], cwd=workspace_dir)
     subprocess.run(["rm", "-r", "target"], cwd=workspace_dir)
 
 
 def init_workspace(workspace_dir: str):
+    """initialize a workspace by copying coverage script"""
     fuzz_aug_home = os.environ["FUZZ_AUG_HOME"]
     cov_script_path = f"{fuzz_aug_home}/evaluation/rust_test_coverage.sh"
     subprocess.run(["cp", cov_script_path, workspace_dir])
@@ -29,6 +31,7 @@ def init_workspace(workspace_dir: str):
 def get_coverage(
     workspace_dir: str, test_target: str, clean_run: bool = False, timeout: int = 60
 ) -> Result[float, CovError]:
+    """get coverage of a specific test target in the project"""
     if clean_run:
         clean_workspace(workspace_dir)
 
@@ -105,15 +108,3 @@ def get_testcase_coverages(workspace_dir: str) -> dict[str, float]:
         cov = get_coverage(workspace_dir, test_name).unwrap()
         coverages[test_name] = cov
     return coverages
-
-
-def main():
-    workspace_dir = os.path.abspath(
-        "data/repos/marshallpierce-rust-base64/marshallpierce-rust-base64-4ef33cc"
-    )
-
-    print(get_testcase_coverages(workspace_dir))
-
-
-if __name__ == "__main__":
-    fire.Fire(main)

evaluation/humaneval.py

@@ -1,47 +1,51 @@
 import json
+import os
+import re
+import tempfile
+import logging
+from multiprocessing import cpu_count
+from typing import Iterable
+
 import fire
 from funcy import func_partial
 from funcy_chain import Chain
-from functools import reduce
 from dacite import from_dict
-from returns.result import Result, Success, Failure
-import os
-import re
-import tempfile
+from returns.result import Success, Failure
 from tqdm import tqdm
 import pandas as pd
-import logging
-from multiprocessing import cpu_count
 from pathos.multiprocessing import ProcessingPool
-from returns.result import Result, Success, Failure
-from typing import Iterable
-from evaluation.util import CovError, HumanEvalTask
+
+from evaluation.util import HumanEvalTask
 from evaluation.result_analysis import to_record
 
 
 def init_humaneval_x_workspace(tmpdir: str = "tmp"):
+    """setup a temporary workspace for compiling and running humaneval-x programs"""
     fuzz_aug_home = os.environ["FUZZ_AUG_HOME"]
     cargo_files = f"{fuzz_aug_home}/evaluation/cargo/Cargo.*"
     os.system(f"cp {cargo_files} {tmpdir}/")
     os.mkdir(f"{tmpdir}/src")
 
 
 def evaluate_program(program: str, fn_name: str, timeout: int):
-    from evaluation.coverage import get_coverage
+    """write, compile, and run a HumanEval-X program, return coverage result"""
+    from evaluation.coverage import (  # pylint: disable=import-outside-toplevel
+        get_coverage,
+    )
+
+    with tempfile.TemporaryDirectory() as tmpdir_path:
 
-    tmpdir = tempfile.TemporaryDirectory()
-    tmpdir_path = tmpdir.name
+        init_humaneval_x_workspace(tmpdir=tmpdir_path)
+        test_target = f"test_{fn_name}"
+        with open(f"{tmpdir_path}/src/main.rs", "w") as fp:
+            fp.write(program)
+        cov = get_coverage(tmpdir_path, test_target, timeout=timeout)
 
-    init_humaneval_x_workspace(tmpdir=tmpdir_path)
-    test_target = f"test_{fn_name}"
-    with open(f"{tmpdir_path}/src/main.rs", "w") as fp:
-        fp.write(program)
-    cov = get_coverage(tmpdir_path, test_target, timeout=timeout)
-    tmpdir.cleanup()
-    return cov
+        return cov
 
 
 def evaluate_assertions(solution: HumanEvalTask, timeout: int = 60):
+    """evaluate individual assertions in the generated test function"""
     assert solution.fn_name is not None
     return [
         evaluate_program(p, solution.fn_name, timeout)
@@ -92,17 +96,20 @@ def extract_assertions(test_function: str) -> list[str]:
 
 
 def concat_assertions_to_test(solution, assertions):
+    """concatenate multiple extracted assertions into a single test function"""
     return solution.assertion_to_program("\n".join(assertions))
 
 
 def generated_assertions(sol: HumanEvalTask) -> list[str]:
+    """extract assertions from the generated test function"""
     assert sol.generated_test is not None
     return extract_assertions(sol.test_prompt_header + sol.generated_test)
 
 
 def evaluate_whole_function_coverage(solution: HumanEvalTask, timeout: int = 60):
     """
-    Evaluate the whole function coverage using the combined test function of correct assertions, which is similar to evaluate_assertion
+    Evaluate the whole function coverage using the combined test function of correct assertions,
+    which is similar to evaluate_assertion
     """
     assert solution.fn_name is not None
 
@@ -120,8 +127,8 @@ def main(
     """evaluate coverage on HumanEval-X
 
     Args:
-        input_human_eval_x_path (str, optional): input path. Defaults to "data/humaneval_rust.jsonl".
-        output_result_file (str, optional): write to write results. Defaults to "humaneval_rust_coverage.jsonl".
+        input_human_eval_x_path (str, optional): input path.
+        output_result_file (str, optional): write to write results.
         nproc (int, optional): number of proc to use. Defaults to cpu_count().
         timeout (int, optional): allowance time for each exec, in seconds. Defaults to 60.
         whole_function_coverage(bool, optional): whether or not test whole function coverage