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

src/dynapyt/analyses/TraceAll.py

@@ -1358,7 +1358,7 @@ def exception(
         self.log(iid, "Caught", caught, "from", exceptions)
 
     def enter_with(self, dyn_ast: str, iid: int, ctx_manager: ContextManager) -> None:
-        """Hook for entering a with statement.
+        """Hook for entering a with item.
 
 
         Parameters
@@ -1376,7 +1376,7 @@ def enter_with(self, dyn_ast: str, iid: int, ctx_manager: ContextManager) -> Non
         self.log(iid, "Entered with")
 
     def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
-        """Hook for exiting a with statement.
+        """Hook for exiting a with item.
 
 
         Parameters

src/dynapyt/instrument/CodeInstrumenter.py

@@ -1876,30 +1876,22 @@ def leave_CompFor(self, original_node, updated_node):
             iter=generator_call, target=original_node.target
         )
 
-    def leave_With(self, original_node, updated_node):
+    def leave_WithItem(self, original_node, updated_node):
         if (
             "enter_with" not in self.selected_hooks
             and "exit_with" not in self.selected_hooks
         ):
             return updated_node
-
+    
         iid = self.__create_iid(original_node)
         ast_arg = cst.Arg(value=cst.Name("_dynapyt_ast_"))
         iid_arg = cst.Arg(value=cst.Integer(value=str(iid)))
 
-        ctx_manager_arg = cst.Arg(value=updated_node.items[0].item)
+        item = updated_node
+        ctx_manager_arg = cst.Arg(value=item.item)
         callee_name = cst.Attribute(
             value=cst.Name(value="_rt"), attr=cst.Name(value="_enter_with_")
         )
         self.to_import.add("_enter_with_")
         call = cst.Call(func=callee_name, args=[ast_arg, iid_arg, ctx_manager_arg])
-        with_item = cst.WithItem(
-            item=call,
-            asname=updated_node.items[0].asname,
-            comma=updated_node.items[0].comma,
-        )
-        with_items = [with_item]
-
-        return updated_node.with_changes(
-            items=with_items,
-        )
+        return updated_node.with_changes(item=call)

tests/trace_single_hook/with/with_multiple_items/expected.txt

@@ -0,0 +1,12 @@
+begin execution
+with statement entered
+with statement entered
+Line read from expected.txt:  begin execution
+
+Line read from analysis.py:  from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+content has been read from the files
+with statement exited
+with statement exited
+file has been closed
+end execution

tests/trace_single_hook/with/with_multiple_items/program.py

@@ -0,0 +1,13 @@
+import os
+
+dir_name = os.path.dirname(os.path.realpath(__file__))
+file_path_1 = os.path.join(dir_name, "expected.txt")
+file_path_2 = os.path.join(dir_name, "analysis.py")
+with open(file_path_1, "r") as file1, open(file_path_2, "r") as file2:
+    content = file1.readline()
+    print("Line read from expected.txt: ", content)
+    content = file2.readline()
+    print("Line read from analysis.py: ", content)
+    print("content has been read from the files")
+
+print("file has been closed")

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_1/analysis.py

@@ -0,0 +1,14 @@
+from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+class TestAnalysis(BaseAnalysis):
+    def begin_execution(self) -> None:
+        print("begin execution")
+
+    def enter_with(self, dyn_ast: str, iid: int, ctx_manager):
+        print(f"with statement entered")
+
+    def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
+        print(f"with statement exited")
+
+    def end_execution(self) -> None:
+        print("end execution")

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_1/expected.txt

@@ -0,0 +1,11 @@
+begin execution
+with statement entered
+with statement entered
+Line read from expected.txt: begin execution
+
+Line read from analysis.py: from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+content has been read from the files
+with statement exited
+with statement exited
+end execution

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_1/program.py

@@ -0,0 +1,19 @@
+import os
+
+dir_name = os.path.dirname(os.path.realpath(__file__))
+file_path_one = os.path.join(dir_name, "expected.txt")
+file_path_two = os.path.join(dir_name, "analysis.py")
+
+file1 = open(file_path_one, "r")
+file2 = open(file_path_two, "r")
+with file1, file2:
+    content = file1.readline()
+    print("Line read from expected.txt: " + content)
+    content = file2.readline()
+    print("Line read from analysis.py: " + content)
+    print("content has been read from the files")
+
+
+
+
+

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_2/analysis.py

@@ -0,0 +1,14 @@
+from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+class TestAnalysis(BaseAnalysis):
+    def begin_execution(self) -> None:
+        print("begin execution")
+
+    def enter_with(self, dyn_ast: str, iid: int, ctx_manager):
+        print(f"with statement entered")
+
+    def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
+        print(f"with statement exited")
+
+    def end_execution(self) -> None:
+        print("end execution")

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_2/expected.txt

@@ -0,0 +1,15 @@
+begin execution
+with statement entered
+with statement entered
+with statement entered
+Line read from expected.txt: begin execution
+
+Line read from expected.txt: begin execution
+
+Line read from analysis.py: from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+content has been read from the files
+with statement exited
+with statement exited
+with statement exited
+end execution

tests/trace_single_hook/with/with_multiple_items_as_specifier_case_2/program.py

@@ -0,0 +1,22 @@
+import os
+
+dir_name = os.path.dirname(os.path.realpath(__file__))
+file_path_one = os.path.join(dir_name, "expected.txt")
+file_path_two = os.path.join(dir_name, "expected.txt")
+file_path_three = os.path.join(dir_name, "analysis.py")
+
+file1 = open(file_path_one, "r")
+file3 = open(file_path_three, "r")
+with file1, open(file_path_two, "r") as file2, file3:
+    content = file1.readline()
+    print("Line read from expected.txt: " + content)
+    content = file2.readline()
+    print("Line read from expected.txt: " + content)
+    content = file3.readline()
+    print("Line read from analysis.py: " + content)
+    print("content has been read from the files")
+
+
+
+
+

tests/trace_single_hook/with/with_multiple_items_suppress_exception/expected.txt

@@ -0,0 +1,18 @@
+begin execution
+with statement entered
+ContextManagerOne: __enter__
+with statement entered
+ContextManagerTwo: __enter__
+inside with block
+context manager one
+context manager two
+ContextManagerTwo: __exit__
+exc_type:  <class 'Exception'>
+exc_value:  exception raised inside with statement block
+with statement exited
+ContextManagerOne: __exit__
+exc_type:  None
+exc_value:  None
+with statement exited
+no exception raised
+end execution

tests/trace_single_hook/with/with_multiple_items_suppress_exception/program.py

@@ -0,0 +1,34 @@
+class ContextManagerOne:
+
+    def __enter__(self):
+        print('ContextManagerOne: __enter__')
+        return "context manager one"
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        print('ContextManagerOne: __exit__')
+        print("exc_type: ", exc_type)
+        print("exc_value: ", exc_value)
+        return True
+
+class ContextManagerTwo:
+
+    def __enter__(self):
+        print('ContextManagerTwo: __enter__')
+        return "context manager two"
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        print('ContextManagerTwo: __exit__')
+        print("exc_type: ", exc_type)
+        print("exc_value: ", exc_value)
+        return True
+
+try :    
+    with ContextManagerOne() as cm1, ContextManagerTwo() as cm2:
+        print('inside with block')
+        print(cm1)
+        print(cm2)
+        raise Exception("exception raised inside with statement block")
+except Exception as e:
+    print("exception caught: ", e)
+else:
+    print("no exception raised")

tests/trace_single_hook/with/with_multiple_raise_exception/expected.txt

@@ -0,0 +1,18 @@
+begin execution
+with statement entered
+ContextManagerOne: __enter__
+with statement entered
+ContextManagerTwo: __enter__
+inside with block
+context manager one
+context manager two
+ContextManagerTwo: __exit__
+exc_type:  <class 'Exception'>
+exc_value:  exception raised inside with statement block
+with statement exited
+ContextManagerOne: __exit__
+exc_type:  <class 'Exception'>
+exc_value:  exception raised inside with statement block
+with statement exited
+exception caught:  exception raised inside with statement block
+end execution

tests/trace_single_hook/with/with_multiple_raise_exception/program.py

@@ -0,0 +1,32 @@
+class ContextManagerOne:
+
+    def __enter__(self):
+        print('ContextManagerOne: __enter__')
+        return "context manager one"
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        print('ContextManagerOne: __exit__')
+        print("exc_type: ", exc_type)
+        print("exc_value: ", exc_value)
+
+class ContextManagerTwo:
+
+    def __enter__(self):
+        print('ContextManagerTwo: __enter__')
+        return "context manager two"
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        print('ContextManagerTwo: __exit__')
+        print("exc_type: ", exc_type)
+        print("exc_value: ", exc_value)
+
+try :    
+    with ContextManagerOne() as cm1, ContextManagerTwo() as cm2:
+        print('inside with block')
+        print(cm1)
+        print(cm2)
+        raise Exception("exception raised inside with statement block")
+except Exception as e:
+    print("exception caught: ", e)
+else:
+    print("no exception raised")

tests/trace_single_hook/with/with_raise_exception/analysis.py

@@ -0,0 +1,15 @@
+from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+class TestAnalysis(BaseAnalysis):
+
+    def begin_execution(self) -> None:
+        print("begin execution")
+
+    def enter_with(self, dyn_ast: str, iid: int, ctx_manager):
+        print(f"with statement entered")
+
+    def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
+        print(f"with statement exited")
+
+    def end_execution(self) -> None:
+        print("end execution")

tests/trace_single_hook/with/with_suppress_exception/analysis.py

@@ -0,0 +1,15 @@
+from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+
+class TestAnalysis(BaseAnalysis):
+    def begin_execution(self) -> None:
+        print("begin execution")
+
+    def enter_with(self, dyn_ast: str, iid: int, ctx_manager):
+        print(f"with statement entered")
+
+    def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
+        print(f"with statement exited")
+
+    def end_execution(self) -> None:
+        print("end execution")

tests/trace_single_hook/with/with_without_as_specifier/analysis.py

@@ -0,0 +1,14 @@
+from dynapyt.analyses.BaseAnalysis import BaseAnalysis
+
+class TestAnalysis(BaseAnalysis):
+    def begin_execution(self) -> None:
+        print("begin execution")
+
+    def enter_with(self, dyn_ast: str, iid: int, ctx_manager):
+        print(f"with statement entered")
+
+    def exit_with(self, dyn_ast: str, iid: int, is_suppressed: bool, exc_value):
+        print(f"with statement exited")
+
+    def end_execution(self) -> None:
+        print("end execution")