Merge branch 'main' of github.com:sola-st/DynaPyt

requirements.txt

@@ -15,4 +15,5 @@ pytest-forked==1.4.0
 pytest-xdist==3.3.1
 typing-inspect==0.7.1
 typing_extensions==4.1.1
-zipp==3.6.0
+zipp==3.6.0
+filelock

src/dynapyt/run_analysis.py

@@ -2,66 +2,50 @@
 import argparse
 import importlib
 from os.path import abspath
+from shutil import rmtree
 import sys
-import signal
+from pathlib import Path
+from . import runtime as _rt
 
 
-def run_analysis(entry: str, analyses: List[str], name: str = None):
-    my_analyses = []
-    try:
-        for analysis in analyses:
-            module = importlib.import_module(".".join(analysis.split(".")[:-1]))
-            class_ = getattr(module, analysis.split(".")[-1])
-            my_analyses.append(class_())
-    except TypeError as e:
-        print(f"--module was used but no value specified {e}")
-    except ImportError as e:
-        print(f"module could not be imported {e}")
-
-    rt.set_analysis(my_analyses)
-
-    def end_execution():
-        try:
-            for my_analysis in my_analyses:
-                func = getattr(my_analysis, "end_execution")
-                func()
-        except AttributeError:
-            pass
+def run_analysis(
+    entry: str, analyses: List[str], name: str = None, coverage: bool = False
+):
+    if coverage:
+        Path("/tmp/dynapyt_coverage").mkdir(exist_ok=True)
+    else:
+        rmtree("/tmp/dynapyt_coverage", ignore_errors=True)
 
-    # allow dynapyt to exit gracefully
-    signal.signal(signal.SIGINT, end_execution)
-    signal.signal(signal.SIGTERM, end_execution)
+    if Path("/tmp/dynapyt_analyses.txt").exists():
+        Path("/tmp/dynapyt_analyses.txt").unlink()
+    with open("/tmp/dynapyt_analyses.txt", "w") as f:
+        f.write("\n".join(analyses))
 
-    if not name is None:
-        for my_analysis in my_analyses:
-            getattr(my_analysis, "add_metadata", lambda: None)({"name": name})
+    _rt.set_analysis(analyses)
 
-    try:
-        for my_analysis in my_analyses:
-            func = getattr(my_analysis, "begin_execution")
+    for analysis in _rt.analyses:
+        func = getattr(analysis, "begin_execution", None)
+        if func is not None:
             func()
-    except AttributeError:
-        pass
     if entry.endswith(".py"):
         sys.argv = [entry]
         entry_full_path = abspath(entry)
         globals_dict = globals().copy()
         globals_dict["__file__"] = entry_full_path
         exec(open(entry_full_path).read(), globals_dict)
     else:
-        importlib.import_module(entry, "*")
-    end_execution()
+        importlib.import_module(entry)
+    _rt.end_execution()
 
 
 parser = argparse.ArgumentParser()
 parser.add_argument("--entry", help="Entry file for execution")
 parser.add_argument("--analysis", help="Analysis class name(s)", nargs="+")
 parser.add_argument("--name", help="Associates a given name with current run")
-
-import dynapyt.runtime as rt
+parser.add_argument("--coverage", help="Enables coverage", action="store_true")
 
 if __name__ == "__main__":
     args = parser.parse_args()
     name = args.name
     analyses = args.analysis
-    run_analysis(args.entry, analyses, name)
+    run_analysis(args.entry, analyses, name, args.coverage)

src/dynapyt/run_instrumentation.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Set
 import argparse
 from os import walk
 from os import path
@@ -9,7 +9,12 @@
 from multiprocessing import Pool
 
 
-def instrument_dir(directory: str, analysis: List[str], use_external_dir: bool = False):
+def instrument_dir(
+    directory: str,
+    analysis: List[str],
+    use_external_dir: bool = False,
+    exclude: Set[str] = set(),
+):
     start_time = time.time()
     start = directory
     all_cmds = []
@@ -22,9 +27,12 @@ def instrument_dir(directory: str, analysis: List[str], use_external_dir: bool =
         start = str(external_path)
 
     for dir_path, dir_names, file_names in walk(start):
+        for name in dir_names:
+            if path.join(dir_path, name) in exclude:
+                dir_names.remove(name)
         for name in file_names:
-            if name.endswith(".py"):
-                file_path = path.join(dir_path, name)
+            file_path = path.join(dir_path, name)
+            if name.endswith(".py") and file_path not in exclude:
                 cmd_list = [
                     "python",
                     "-m",

src/dynapyt/runtime.py

@@ -1,22 +1,117 @@
 from typing import List, Tuple, Any
+from pathlib import Path
 from sys import exc_info
+import sys
+import atexit
+import signal
+import json
+import importlib
+from filelock import FileLock
 import libcst as cst
-from dynapyt.utils.hooks import snake, get_name
+from .utils.hooks import snake, get_name
+from .instrument.IIDs import IIDs
+from .instrument.filters import START, END, SEPERATOR
+from .utils.load_analysis import load_analyses
 
 analyses = None
+covered = None
+current_file = None
+
+
+def end_execution():
+    global covered
+    call_if_exists("end_execution")
+    if covered is not None:
+        with FileLock("/tmp/dynapyt_coverage/covered.jsonl.lock"):
+            if Path("/tmp/dynapyt_coverage/covered.jsonl").exists():
+                existing_coverage = {}
+                with open("/tmp/dynapyt_coverage/covered.jsonl", "r") as f:
+                    content = f.read().splitlines()
+                for c in content:
+                    tmp = json.loads(c)
+                    print(tmp, file=sys.stderr)
+                    existing_coverage.update(tmp)
+                Path("/tmp/dynapyt_coverage/covered.jsonl").unlink()
+            else:
+                existing_coverage = {}
+            for r_file, line_nums in covered.items():
+                if r_file not in existing_coverage:
+                    existing_coverage[r_file] = {}
+                for ln, anas in line_nums.items():
+                    if ln not in existing_coverage[r_file]:
+                        existing_coverage[r_file][ln] = {}
+                    for ana, count in anas.items():
+                        if ana not in existing_coverage[r_file][ln]:
+                            existing_coverage[r_file][ln][ana] = 0
+                        existing_coverage[r_file][ln][ana] += count
+            with open("/tmp/dynapyt_coverage/covered.jsonl", "w") as f:
+                for r_file, line_nums in existing_coverage.items():
+                    tmp = {r_file: line_nums}
+                    f.write(json.dumps(tmp) + "\n")
 
 
 def set_analysis(new_analyses: List[Any]):
-    global analyses
-    analyses = new_analyses
+    global analyses, covered
+    if analyses is None:
+        analyses = []
+        if Path("/tmp/dynapyt_coverage/").exists():
+            covered = {}
+        signal.signal(signal.SIGINT, end_execution)
+        signal.signal(signal.SIGTERM, end_execution)
+        atexit.register(end_execution)
+        analyses = load_analyses(new_analyses)
+
+
+def filtered(func, f, args):
+    docs = func.__doc__
+    if docs is None or START not in docs:
+        return False
+    if len(args) >= 2:
+        sub_args = args[2:]
+    else:
+        return False
+    while START in docs:
+        start = docs.find(START)
+        end = docs.find(END)
+        fltr = docs[start + len(START) : end].strip()
+        patterns = fltr.split(" -> ")[1].split(SEPERATOR)
+        if fltr.startswith("only ->") and any(
+            [getattr(arg, "__name__", repr(arg)) in patterns for arg in sub_args]
+        ):
+            return False
+        elif fltr.startswith("ignore ->") and any(
+            [getattr(arg, "__name__", repr(arg)) in patterns for arg in sub_args]
+        ):
+            return True
+        docs = docs[end + len(END) :].lstrip()
+    return False
 
 
 def call_if_exists(f, *args):
-    global analyses
+    global covered, analyses, current_file
     return_value = None
+    if analyses is None:
+        with open("/tmp/dynapyt_analyses.txt", "r") as af:
+            analysis_list = af.read().split("\n")
+        set_analysis(analysis_list)
     for analysis in analyses:
-        func = getattr(analysis, f, lambda *args: None)
-        return_value = func(*args)
+        func = getattr(analysis, f, None)
+        if func is not None and not filtered(func, f, args):
+            return_value = func(*args)
+            if covered is not None and len(args) >= 2:
+                r_file, iid = args[0], args[1]
+                if current_file is None or current_file.file_path != r_file:
+                    current_file = IIDs(r_file)
+                if r_file not in covered:
+                    covered[r_file] = {}
+                line_no = current_file.iid_to_location[
+                    iid
+                ].start_line  # This is not accurate for multiline statements like if, for, multiline calls, etc.
+                if line_no not in covered[r_file]:
+                    covered[r_file][line_no] = {analysis.__class__.__name__: 0}
+                if analysis.__class__.__name__ not in covered[r_file][line_no]:
+                    covered[r_file][line_no][analysis.__class__.__name__] = 0
+                covered[r_file][line_no][analysis.__class__.__name__] += 1
     return return_value
 
 

src/dynapyt/utils/hooks.py

@@ -11,6 +11,7 @@
 import importlib
 
 from ..instrument.filters import START, END, SEPERATOR, get_details
+from .load_analysis import load_analyses
 
 
 def snake(x):
@@ -63,10 +64,8 @@ def get_used_leaves(
 def get_hooks_from_analysis(classes: List[str]) -> Dict[str, Dict[str, List[str]]]:
     try:
         methods = {}
-        for cls in classes:
-            module = importlib.import_module(".".join(cls.split(".")[:-1]))
-            class_ = getattr(module, cls.split(".")[-1])
-            instance = class_()
+        analyses = load_analyses(classes)
+        for instance in analyses:
             methods.update(
                 {
                     func: get_details(getattr(instance, func))

src/dynapyt/utils/load_analysis.py

@@ -0,0 +1,24 @@
+import importlib
+from typing import List, Any
+from ..analyses.BaseAnalysis import BaseAnalysis
+
+
+def load_analyses(analyses: List[Any]) -> List[BaseAnalysis]:
+    res_analyses = []
+    for ana in analyses:
+        if isinstance(ana, str):
+            conf = None
+            if ":" in ana:
+                ana, conf = ana.split(":")
+            module_parts = ana.split(".")
+            module = importlib.import_module(".".join(module_parts[:-1]))
+            class_ = getattr(module, module_parts[-1])
+            if conf is not None:
+                res_analyses.append(class_(conf))
+            else:
+                res_analyses.append(class_())
+        elif isinstance(ana, BaseAnalysis):
+            res_analyses.append(ana)
+        else:
+            continue
+    return res_analyses

tests/run_single_test.py

@@ -69,6 +69,7 @@ def test_runner(directory_pair: Tuple[str, str], capsys):
     # analyze
     # class_ = getattr(module, "TestAnalysis")
     analysis_instances = [class_[1]() for class_ in analysis_classes]
+    rt.analyses = None
     rt.set_analysis(analysis_instances)
     captured = capsys.readouterr()  # clear stdout
     # print(f"Before analysis: {captured.out}")  # for debugging purposes