import textwrap
import types
import unittest
-
+import asyncio
PAIR = (0,1)
expected = func.events
self.assertEqual(events, expected)
-
class MonitoringEventsTest(MonitoringEventsBase, unittest.TestCase):
def test_just_pass(self):
class CheckEvents(MonitoringTestBase, unittest.TestCase):
- def check_events(self, func, expected, tool=TEST_TOOL, recorders=(ExceptionRecorder,)):
+ def get_events(self, func, tool, recorders):
try:
self.assertEqual(sys.monitoring._all_events(), {})
event_list = []
sys.monitoring.set_events(tool, 0)
for recorder in recorders:
sys.monitoring.register_callback(tool, recorder.event_type, None)
- self.assertEqual(event_list, expected)
+ return event_list
finally:
sys.monitoring.set_events(tool, 0)
for recorder in recorders:
sys.monitoring.register_callback(tool, recorder.event_type, None)
+ def check_events(self, func, expected, tool=TEST_TOOL, recorders=(ExceptionRecorder,)):
+ events = self.get_events(func, tool, recorders)
+ if events != expected:
+ print(events, file = sys.stderr)
+ self.assertEqual(events, expected)
+
+ def check_balanced(self, func, recorders):
+ events = self.get_events(func, TEST_TOOL, recorders)
+ self.assertEqual(len(events)%2, 0)
+ for r, h in zip(events[::2],events[1::2]):
+ r0 = r[0]
+ self.assertIn(r0, ("raise", "reraise"))
+ h0 = h[0]
+ self.assertIn(h0, ("handled", "unwind"))
+
+
+
class StopiterationRecorder(ExceptionRecorder):
event_type = E.STOP_ITERATION
-class ExceptionMontoringTest(CheckEvents):
+class ReraiseRecorder(ExceptionRecorder):
+
+ event_type = E.RERAISE
+
+ def __call__(self, code, offset, exc):
+ self.events.append(("reraise", type(exc)))
+
+class UnwindRecorder(ExceptionRecorder):
+
+ event_type = E.PY_UNWIND
+
+ def __call__(self, *args):
+ self.events.append(("unwind", None))
+
+class ExceptionHandledRecorder(ExceptionRecorder):
+
+ event_type = E.EXCEPTION_HANDLED
+
+ def __call__(self, code, offset, exc):
+ self.events.append(("handled", type(exc)))
- recorder = ExceptionRecorder
+class ExceptionMonitoringTest(CheckEvents):
+
+
+ exception_recorders = (
+ ExceptionRecorder,
+ ReraiseRecorder,
+ ExceptionHandledRecorder,
+ UnwindRecorder
+ )
def test_simple_try_except(self):
self.check_events(func1, [("raise", KeyError)])
+ def test_implicit_stop_iteration(self):
+
def gen():
yield 1
return 2
self.check_events(implicit_stop_iteration, [("raise", StopIteration)], recorders=(StopiterationRecorder,))
+ initial = [
+ ("raise", ZeroDivisionError),
+ ("handled", ZeroDivisionError)
+ ]
+
+ reraise = [
+ ("reraise", ZeroDivisionError),
+ ("handled", ZeroDivisionError)
+ ]
+
+ def test_explicit_reraise(self):
+
+ def func():
+ try:
+ try:
+ 1/0
+ except:
+ raise
+ except:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
+ def test_explicit_reraise_named(self):
+
+ def func():
+ try:
+ try:
+ 1/0
+ except Exception as ex:
+ raise
+ except:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
+ def test_implicit_reraise(self):
+
+ def func():
+ try:
+ try:
+ 1/0
+ except ValueError:
+ pass
+ except:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
+
+ def test_implicit_reraise_named(self):
+
+ def func():
+ try:
+ try:
+ 1/0
+ except ValueError as ex:
+ pass
+ except:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
+ def test_try_finally(self):
+
+ def func():
+ try:
+ try:
+ 1/0
+ finally:
+ pass
+ except:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
+ def test_async_for(self):
+
+ def func():
+
+ async def async_generator():
+ for i in range(1):
+ raise ZeroDivisionError
+ yield i
+
+ async def async_loop():
+ try:
+ async for item in async_generator():
+ pass
+ except Exception:
+ pass
+
+ try:
+ async_loop().send(None)
+ except StopIteration:
+ pass
+
+ self.check_balanced(
+ func,
+ recorders = self.exception_recorders)
+
class LineRecorder:
event_type = E.LINE
line3 = 3
self.check_events(func1, recorders = MANY_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'func1', sys.monitoring.MISSING),
('line', 'func1', 1),
('line', 'func1', 2),
('line', 'func1', 3),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2)])
def test_c_call(self):
line3 = 3
self.check_events(func2, recorders = MANY_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'func2', sys.monitoring.MISSING),
('line', 'func2', 1),
('line', 'func2', 2),
('call', 'append', [2]),
('C return', 'append', [2]),
('line', 'func2', 3),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2)])
def test_try_except(self):
line = 6
self.check_events(func3, recorders = MANY_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'func3', sys.monitoring.MISSING),
('line', 'func3', 1),
('line', 'func3', 2),
('line', 'func3', 4),
('line', 'func3', 5),
('line', 'func3', 6),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2)])
class InstructionRecorder:
def __call__(self, code, offset):
# Filter out instructions in check_events to lower noise
- if code.co_name != "check_events":
+ if code.co_name != "get_events":
self.events.append(("instruction", code.co_name, offset))
line3 = 3
self.check_events(func1, recorders = LINE_AND_INSTRUCTION_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func1', 1),
('instruction', 'func1', 2),
('instruction', 'func1', 4),
('instruction', 'func1', 10),
('instruction', 'func1', 12),
('instruction', 'func1', 14),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
def test_c_call(self):
line3 = 3
self.check_events(func2, recorders = LINE_AND_INSTRUCTION_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func2', 1),
('instruction', 'func2', 2),
('instruction', 'func2', 4),
('instruction', 'func2', 40),
('instruction', 'func2', 42),
('instruction', 'func2', 44),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
def test_try_except(self):
line = 6
self.check_events(func3, recorders = LINE_AND_INSTRUCTION_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func3', 1),
('instruction', 'func3', 2),
('line', 'func3', 2),
('instruction', 'func3', 30),
('instruction', 'func3', 32),
('instruction', 'func3', 34),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
def test_with_restart(self):
def func1():
line3 = 3
self.check_events(func1, recorders = LINE_AND_INSTRUCTION_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func1', 1),
('instruction', 'func1', 2),
('instruction', 'func1', 4),
('instruction', 'func1', 10),
('instruction', 'func1', 12),
('instruction', 'func1', 14),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
sys.monitoring.restart_events()
self.check_events(func1, recorders = LINE_AND_INSTRUCTION_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func1', 1),
('instruction', 'func1', 2),
('instruction', 'func1', 4),
('instruction', 'func1', 10),
('instruction', 'func1', 12),
('instruction', 'func1', 14),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
class TestInstallIncrementallly(MonitoringTestBase, unittest.TestCase):
('branch', 'func', 2, 2)])
self.check_events(func, recorders = JUMP_BRANCH_AND_LINE_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func', 1),
('line', 'func', 2),
('branch', 'func', 2, 2),
('jump', 'func', 4, 2),
('line', 'func', 2),
('branch', 'func', 2, 2),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
def test_except_star(self):
self.check_events(func, recorders = JUMP_BRANCH_AND_LINE_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func', 1),
('line', 'func', 2),
('line', 'func', 3),
('jump', 'func', 5, 5),
('jump', 'func', 5, '[offset=112]'),
('branch', 'func', '[offset=118]', '[offset=120]'),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
self.check_events(func, recorders = FLOW_AND_LINE_RECORDERS, expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('line', 'func', 1),
('line', 'func', 2),
('line', 'func', 3),
('jump', 'func', 5, '[offset=112]'),
('branch', 'func', '[offset=118]', '[offset=120]'),
('return', None),
- ('line', 'check_events', 11)])
+ ('line', 'get_events', 11)])
class TestLoadSuperAttr(CheckEvents):
RECORDERS = CallRecorder, LineRecorder, CRaiseRecorder, CReturnRecorder
"""
d = self._exec_super(codestr, optimized)
expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'f', sys.monitoring.MISSING),
('line', 'f', 1),
('call', 'method', d["b"]),
('call', 'method', 1),
('line', 'method', 1),
('line', 'method', 1),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2),
]
return d["f"], expected
"""
d = self._exec_super(codestr, optimized)
expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'f', sys.monitoring.MISSING),
('line', 'f', 1),
('line', 'f', 2),
('C raise', 'super', 1),
('line', 'f', 3),
('line', 'f', 4),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2),
]
return d["f"], expected
"""
d = self._exec_super(codestr, optimized)
expected = [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'f', sys.monitoring.MISSING),
('line', 'f', 1),
('call', 'method', d["b"]),
('C return', 'super', sys.monitoring.MISSING),
('line', 'method', 2),
('line', 'method', 1),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2)
]
return d["f"], expected
def get_expected(name, call_method, ns):
repr_arg = 0 if name == "int" else sys.monitoring.MISSING
return [
- ('line', 'check_events', 10),
+ ('line', 'get_events', 10),
('call', 'f', sys.monitoring.MISSING),
('line', 'f', 1),
('call', 'method', ns["c"]),
('C return', '__repr__', repr_arg),
] if call_method else []
),
- ('line', 'check_events', 11),
+ ('line', 'get_events', 11),
('call', 'set_events', 2),
]