Fixed issues with windows

.github/workflows/tests.yml

@@ -5,10 +5,10 @@ on: [push]
 jobs:
   build:
 
-    runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ["3.7", "3.8", "3.9", "3.10"]
+        os: ["ubuntu-latest", "windows-latest"]
 
     steps:
       - uses: actions/checkout@v3

src/dynapyt/analyses/CallGraph.py

@@ -1,56 +1,68 @@
 from typing import Callable, Tuple, Dict
 import logging
-import libcst as cst
+import os
 import libcst.matchers as m
 from .BaseAnalysis import BaseAnalysis
 from ..utils.nodeLocator import get_parent_by_type
 import json
 from inspect import getmodule
 
+
 class CallGraph(BaseAnalysis):
     def __init__(self):
         super(CallGraph, self).__init__()
-        logging.basicConfig(filename="dynapyt.json", format='%(message)s', level=logging.INFO)
+        logging.basicConfig(
+            filename="dynapyt.json", format="%(message)s", level=logging.INFO
+        )
         self.graph = {}
 
-    '''
+    """
     DynaPyt hook for pre function call
-    '''
-    def pre_call(self, dyn_ast: str, iid: int, function: Callable, pos_args: Tuple, kw_args: Dict):
+    """
+
+    def pre_call(
+        self, dyn_ast: str, iid: int, function: Callable, pos_args: Tuple, kw_args: Dict
+    ):
         ast, iids = self._get_ast(dyn_ast)
         module = getmodule(function)
-        module = str(module).split(' ')[1] if module is not None else "''"
-        # calling function 
+        module = str(module).split(" ")[1] if module is not None else "''"
+        # calling function
         caller = get_parent_by_type(ast, iids.iid_to_location[iid], m.FunctionDef())
         # called function
         if hasattr(function, "__qualname__"):
-            '''
+            """
             module of the callee is added through the module if present
-            '''
-            callee = module[1:-1] + '.' + function.__qualname__ if module != "''" else function.__qualname__
+            """
+            callee = (
+                module[1:-1] + "." + function.__qualname__
+                if module != "''"
+                else function.__qualname__
+            )
         else:
-            '''
+            """
             this is done to capture functions whose function.__qualname__ is not defined,
             but the string gives an idea as to which function is called.
-            Captures MarkDecorator annotations, lambdas object calls, etc 
-            '''
+            Captures MarkDecorator annotations, lambdas object calls, etc
+            """
             temp = str(function)
             callee = temp
-        
-        #file name
-        key = dyn_ast.replace('.py.orig', '').replace('/','.')
+
+        # file name
+        key = dyn_ast.replace(".py.orig", "").replace(os.sep, ".")
         # format = "file"
-        
+
         if caller is None:
             f = key
         else:
             # if caller is a part of class, find the class name
-            caller_parent = get_parent_by_type(ast, iids.iid_to_location[iid], m.ClassDef())
+            caller_parent = get_parent_by_type(
+                ast, iids.iid_to_location[iid], m.ClassDef()
+            )
             if caller_parent is None:
-                f = key + '.' + caller.name.value
+                f = key + "." + caller.name.value
                 # format += ".func"
             else:
-                f = key + '.' + caller_parent.name.value + '.' + caller.name.value
+                f = key + "." + caller_parent.name.value + "." + caller.name.value
                 # format += ".class.func"
 
         # if caller already added
@@ -63,22 +75,23 @@ def pre_call(self, dyn_ast: str, iid: int, function: Callable, pos_args: Tuple,
         else:
             # self.graph[f] = [format, callee]
             self.graph[f] = [callee]
-    
-    '''
+
+    """
     DynaPyt hook for end of execution
-    '''
+    """
+
     def end_execution(self):
-        '''
+        """
         to avoid serialization failures in converting dict to json
-        '''
+        """
         try:
             logging.info(json.dumps(self.graph))
         except Exception:
             logging.info("{")
             for idx, key in enumerate(self.graph):
-                values = ["\"{}\"".format(x) for x in self.graph[key]]
+                values = ['"{}"'.format(x) for x in self.graph[key]]
                 if not idx == (len(self.graph.keys()) - 1):
-                    logging.info("\"{}\" : {}, ".format(key, values))
+                    logging.info('"{}" : {}, '.format(key, values))
                 else:
-                    logging.info("\"{}\" : {}".format(key, values))
+                    logging.info('"{}" : {}'.format(key, values))
             logging.info("}")

src/dynapyt/run_analysis.py

@@ -2,6 +2,7 @@
 import argparse
 import importlib
 from os.path import abspath
+from tempfile import gettempdir
 from shutil import rmtree
 import sys
 from pathlib import Path
@@ -11,14 +12,16 @@
 def run_analysis(
     entry: str, analyses: List[str], name: str = None, coverage: bool = False
 ):
+    coverage_dir = Path(gettempdir()) / "dynapyt_coverage"
     if coverage:
-        Path("/tmp/dynapyt_coverage").mkdir(exist_ok=True)
+        coverage_dir.mkdir(exist_ok=True)
     else:
-        rmtree("/tmp/dynapyt_coverage", ignore_errors=True)
+        rmtree(str(coverage_dir), ignore_errors=True)
 
-    if Path("/tmp/dynapyt_analyses.txt").exists():
-        Path("/tmp/dynapyt_analyses.txt").unlink()
-    with open("/tmp/dynapyt_analyses.txt", "w") as f:
+    analyses_file = Path(gettempdir()) / "dynapyt_analyses.txt"
+    if analyses_file.exists():
+        analyses_file.unlink()
+    with open(str(analyses_file), "w") as f:
         f.write("\n".join(analyses))
 
     _rt.set_analysis(analyses)

src/dynapyt/runtime.py

@@ -5,7 +5,7 @@
 import atexit
 import signal
 import json
-import importlib
+import tempfile
 from filelock import FileLock
 import libcst as cst
 from .utils.hooks import snake, get_name
@@ -26,16 +26,19 @@ def end_execution():
     end_execution_called = True
     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():
+        coverage_file = (
+            Path(tempfile.gettempdir()) / "dynapyt_coverage" / "covered.jsonl"
+        )
+        with FileLock(f"{str(coverage_file)}.lock"):
+            if coverage_file.exists():
                 existing_coverage = {}
-                with open("/tmp/dynapyt_coverage/covered.jsonl", "r") as f:
+                with open(str(coverage_file), "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()
+                coverage_file.unlink()
             else:
                 existing_coverage = {}
             for r_file, line_nums in covered.items():
@@ -48,7 +51,7 @@ def end_execution():
                         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:
+            with open(str(coverage_file), "w") as f:
                 for r_file, line_nums in existing_coverage.items():
                     tmp = {r_file: line_nums}
                     f.write(json.dumps(tmp) + "\n")
@@ -58,7 +61,8 @@ def set_analysis(new_analyses: List[Any]):
     global analyses, covered
     if analyses is None:
         analyses = []
-        if Path("/tmp/dynapyt_coverage/").exists():
+        coverage_dir = Path(tempfile.gettempdir()) / "dynapyt_coverage"
+        if coverage_dir.exists():
             covered = {}
         signal.signal(signal.SIGINT, end_execution)
         signal.signal(signal.SIGTERM, end_execution)
@@ -95,7 +99,8 @@ def call_if_exists(f, *args):
     global covered, analyses, current_file
     return_value = None
     if analyses is None:
-        with open("/tmp/dynapyt_analyses.txt", "r") as af:
+        analyses_file = Path(tempfile.gettempdir()) / "dynapyt_analyses.txt"
+        with open(str(analyses_file), "r") as af:
             analysis_list = af.read().split("\n")
         set_analysis(analysis_list)
     for analysis in analyses:

tests/regression/file_entry/expected.txt

@@ -1,2 +1,2 @@
 pre call of print
-<...>/program.py
+<...>program.py

tests/test_analysis/call_graph/analysis.py

@@ -1,5 +1,5 @@
 from typing import Callable, Tuple, Dict
-import logging
+import os
 import libcst.matchers as m
 from dynapyt.analyses.BaseAnalysis import BaseAnalysis
 from dynapyt.utils.nodeLocator import get_parent_by_type
@@ -44,7 +44,7 @@ def pre_call(
             callee = temp
 
         # file name
-        key = dyn_ast.replace(".py.orig", "").replace("/", ".").split("tests.")[1]
+        key = dyn_ast.replace(".py.orig", "").replace(os.sep, ".").split("tests.")[1]
         # format = "file"
 
         if caller is None:

tests/trace_single_hook/version_write/program.py

@@ -1,4 +1,4 @@
 import os
 
-with open("/".join(__file__.split("/")[:-1]) + "/__init__.py", "r") as f:
+with open(os.sep.join(__file__.split(os.sep)[:-1]) + f"{os.sep}__init__.py", "r") as f:
     exec(f.read())

tutorial/solution-analyses/CallGraphAnalysis.py

@@ -3,7 +3,7 @@
 import libcst.matchers as m
 from dynapyt.analyses.BaseAnalysis import BaseAnalysis
 from dynapyt.utils.nodeLocator import get_parent_by_type
-import json
+import os
 from inspect import getmodule
 
 
@@ -42,7 +42,7 @@ def pre_call(
             callee = temp
 
         # file name
-        key = dyn_ast.replace(".py.orig", "").replace("/", ".")
+        key = dyn_ast.replace(".py.orig", "").replace(os.sep, ".")
         # format = "file"
 
         if caller is None: