From da99e72771dbcd78de942c9d6395ee0121065b63 Mon Sep 17 00:00:00 2001 From: Shahar Naveh <50263213+ShaharNaveh@users.noreply.github.com> Date: Wed, 3 Jun 2026 16:51:41 +0300 Subject: [PATCH] Update `test_sys{log,_settrace}.py` to 3.14.5 (#8030) * Update `test_syslog.py` to 3.14.5 * Update `test_sys_settrace.py` to 3.14.5 --- Lib/test/test_sys_settrace.py | 1431 +++++++++++++++++++++++++++++---- Lib/test/test_syslog.py | 67 +- 2 files changed, 1357 insertions(+), 141 deletions(-) diff --git a/Lib/test/test_sys_settrace.py b/Lib/test/test_sys_settrace.py index 7eef1290d..54d31a1b6 100644 --- a/Lib/test/test_sys_settrace.py +++ b/Lib/test/test_sys_settrace.py @@ -6,8 +6,17 @@ import sys import difflib import gc from functools import wraps -import asyncio - +from test.support import import_helper, requires_subprocess, run_no_yield_async_fn +import contextlib +import os +import tempfile +import textwrap +import subprocess +import warnings +try: + import _testinternalcapi +except ImportError: + _testinternalcapi = None class tracecontext: """Context manager that traces its enter and exit.""" @@ -38,6 +47,20 @@ async def asynciter(iterable): for x in iterable: yield x +def clean_asynciter(test): + @wraps(test) + async def wrapper(*args, **kwargs): + cleanups = [] + def wrapped_asynciter(iterable): + it = asynciter(iterable) + cleanups.append(it.aclose) + return it + try: + return await test(*args, **kwargs, asynciter=wrapped_asynciter) + finally: + while cleanups: + await cleanups.pop()() + return wrapper # A very basic example. If this fails, we're in deep trouble. def basic(): @@ -300,6 +323,13 @@ generator_example.events = ([(0, 'call'), [(5, 'line'), (5, 'return')]) +def lineno_matches_lasti(frame): + last_line = None + for start, end, line in frame.f_code.co_lines(): + if start <= frame.f_lasti < end: + last_line = line + return last_line == frame.f_lineno + class Tracer: def __init__(self, trace_line_events=None, trace_opcode_events=None): self.trace_line_events = trace_line_events @@ -313,6 +343,7 @@ class Tracer: frame.f_trace_opcodes = self.trace_opcode_events def trace(self, frame, event, arg): + assert lineno_matches_lasti(frame) self._reconfigure_frame(frame) self.events.append((frame.f_lineno, event)) return self.trace @@ -324,7 +355,6 @@ class Tracer: return self.trace -@unittest.skip("TODO: RUSTPYTHON, gc") class TraceTestCase(unittest.TestCase): # Disable gc collection when tracing, otherwise the @@ -344,7 +374,7 @@ class TraceTestCase(unittest.TestCase): return Tracer() def compare_events(self, line_offset, events, expected_events): - events = [(l - line_offset, e) for (l, e) in events] + events = [(l - line_offset if l is not None else None, e) for (l, e) in events] if events != expected_events: self.fail( "events did not match expectation:\n" + @@ -383,36 +413,51 @@ class TraceTestCase(unittest.TestCase): finally: sys.settrace(None) + @unittest.skip("TODO: RUSTPYTHON") def test_01_basic(self): self.run_test(basic) + @unittest.skip("TODO: RUSTPYTHON") def test_02_arigo0(self): self.run_test(arigo_example0) + @unittest.skip("TODO: RUSTPYTHON") def test_02_arigo1(self): self.run_test(arigo_example1) + @unittest.skip("TODO: RUSTPYTHON") def test_02_arigo2(self): self.run_test(arigo_example2) + @unittest.skip("TODO: RUSTPYTHON") def test_03_one_instr(self): self.run_test(one_instr_line) + @unittest.skip("TODO: RUSTPYTHON") def test_04_no_pop_blocks(self): self.run_test(no_pop_blocks) + @unittest.skip("TODO: RUSTPYTHON") def test_05_no_pop_tops(self): self.run_test(no_pop_tops) + @unittest.skip("TODO: RUSTPYTHON") def test_06_call(self): self.run_test(call) + @unittest.skip("TODO: RUSTPYTHON") def test_07_raise(self): self.run_test(test_raise) + @unittest.skip("TODO: RUSTPYTHON") def test_08_settrace_and_return(self): self.run_test2(settrace_and_return) + @unittest.skip("TODO: RUSTPYTHON") def test_09_settrace_and_raise(self): self.run_test2(settrace_and_raise) + @unittest.skip("TODO: RUSTPYTHON") def test_10_ireturn(self): self.run_test(ireturn_example) + @unittest.skip("TODO: RUSTPYTHON") def test_11_tightloop(self): self.run_test(tightloop_example) + @unittest.skip("TODO: RUSTPYTHON") def test_12_tighterloop(self): self.run_test(tighterloop_example) + @unittest.skip("TODO: RUSTPYTHON") def test_13_genexp(self): self.run_test(generator_example) # issue1265: if the trace function contains a generator, @@ -426,6 +471,7 @@ class TraceTestCase(unittest.TestCase): self.compare_events(generator_example.__code__.co_firstlineno, tracer.events, generator_example.events) + @unittest.skip("TODO: RUSTPYTHON") def test_14_onliner_if(self): def onliners(): if True: x=False @@ -438,6 +484,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_15_loops(self): # issue1750076: "while" expression is skipped by debugger def for_example(): @@ -469,6 +516,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_16_blank_lines(self): namespace = {} exec("def f():\n" + "\n" * 256 + " pass", namespace) @@ -478,6 +526,7 @@ class TraceTestCase(unittest.TestCase): (257, 'line'), (257, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_17_none_f_trace(self): # Issue 20041: fix TypeError when f_trace is set to None. def func(): @@ -487,6 +536,7 @@ class TraceTestCase(unittest.TestCase): [(0, 'call'), (1, 'line')]) + @unittest.skip("TODO: RUSTPYTHON") def test_18_except_with_name(self): def func(): try: @@ -511,6 +561,7 @@ class TraceTestCase(unittest.TestCase): (9, 'line'), (9, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_19_except_with_finally(self): def func(): try: @@ -532,6 +583,7 @@ class TraceTestCase(unittest.TestCase): (7, 'line'), (7, 'return')]) + @unittest.expectedFailure # TODO: RUSTPYTHON def test_20_async_for_loop(self): class AsyncIteratorWrapper: def __init__(self, obj): @@ -608,6 +660,60 @@ class TraceTestCase(unittest.TestCase): self.compare_events(doit_async.__code__.co_firstlineno, tracer.events, events) + @unittest.expectedFailure # TODO: RUSTPYTHON + def test_async_for_backwards_jump_has_no_line(self): + async def arange(n): + for i in range(n): + yield i + async def f(): + async for i in arange(3): + if i > 100: + break # should never be traced + + tracer = self.make_tracer() + coro = f() + try: + sys.settrace(tracer.trace) + coro.send(None) + except Exception: + pass + finally: + sys.settrace(None) + + events = [ + (0, 'call'), + (1, 'line'), + (-3, 'call'), + (-2, 'line'), + (-1, 'line'), + (-1, 'return'), + (1, 'exception'), + (2, 'line'), + (1, 'line'), + (-1, 'call'), + (-2, 'line'), + (-1, 'line'), + (-1, 'return'), + (1, 'exception'), + (2, 'line'), + (1, 'line'), + (-1, 'call'), + (-2, 'line'), + (-1, 'line'), + (-1, 'return'), + (1, 'exception'), + (2, 'line'), + (1, 'line'), + (-1, 'call'), + (-2, 'line'), + (-2, 'return'), + (1, 'exception'), + (1, 'return'), + ] + self.compare_events(f.__code__.co_firstlineno, + tracer.events, events) + + @unittest.skip("TODO: RUSTPYTHON") def test_21_repeated_pass(self): def func(): pass @@ -619,6 +725,7 @@ class TraceTestCase(unittest.TestCase): (2, 'line'), (2, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_loop_in_try_except(self): # https://bugs.python.org/issue41670 @@ -636,6 +743,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_try_except_no_exception(self): def func(): @@ -643,16 +751,62 @@ class TraceTestCase(unittest.TestCase): 2 except: 4 - finally: + else: 6 + if False: + 8 + else: + 10 + if func.__name__ == 'Fred': + 12 + finally: + 14 self.run_and_compare(func, [(0, 'call'), (1, 'line'), (2, 'line'), (6, 'line'), - (6, 'return')]) + (7, 'line'), + (10, 'line'), + (11, 'line'), + (14, 'line'), + (14, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") + def test_try_exception_in_else(self): + + def func(): + try: + try: + 3 + except: + 5 + else: + 7 + raise Exception + finally: + 10 + except: + 12 + finally: + 14 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (7, 'line'), + (8, 'line'), + (8, 'exception'), + (10, 'line'), + (11, 'line'), + (12, 'line'), + (14, 'line'), + (14, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") def test_nested_loops(self): def func(): @@ -679,6 +833,7 @@ class TraceTestCase(unittest.TestCase): (4, 'line'), (4, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_if_break(self): def func(): @@ -704,6 +859,7 @@ class TraceTestCase(unittest.TestCase): (8, 'line'), (8, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_break_through_finally(self): def func(): @@ -736,10 +892,10 @@ class TraceTestCase(unittest.TestCase): (5, 'line'), (6, 'line'), (7, 'line'), - (10, 'line'), - (13, 'line'), - (13, 'return')]) + (10, 'line')] + + ([(13, 'line'), (13, 'return')] if __debug__ else [(10, 'return')])) + @unittest.skip("TODO: RUSTPYTHON") def test_continue_through_finally(self): def func(): @@ -773,10 +929,10 @@ class TraceTestCase(unittest.TestCase): (6, 'line'), (7, 'line'), (10, 'line'), - (3, 'line'), - (13, 'line'), - (13, 'return')]) + (3, 'line')] + + ([(13, 'line'), (13, 'return')] if __debug__ else [(3, 'return')])) + @unittest.skip("TODO: RUSTPYTHON") def test_return_through_finally(self): def func(): @@ -792,25 +948,65 @@ class TraceTestCase(unittest.TestCase): (4, 'line'), (4, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_try_except_with_wrong_type(self): def func(): try: - 2/0 - except IndexError: - 4 - finally: - return 6 + try: + 2/0 + except IndexError: + 5 + finally: + 7 + except: + pass + return 10 self.run_and_compare(func, [(0, 'call'), (1, 'line'), (2, 'line'), - (2, 'exception'), (3, 'line'), - (6, 'line'), - (6, 'return')]) + (3, 'exception'), + (4, 'line'), + (7, 'line'), + (8, 'line'), + (9, 'line'), + (10, 'line'), + (10, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") + def test_finally_with_conditional(self): + + # See gh-105658 + condition = True + def func(): + try: + try: + raise Exception + finally: + if condition: + result = 1 + result = 2 + except: + result = 3 + return result + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'exception'), + (5, 'line'), + (6, 'line'), + (8, 'line'), + (9, 'line'), + (10, 'line'), + (10, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") def test_break_to_continue1(self): def func(): @@ -834,6 +1030,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_break_to_continue2(self): def func(): @@ -857,6 +1054,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_break_to_break(self): def func(): @@ -875,6 +1073,7 @@ class TraceTestCase(unittest.TestCase): (5, 'line'), (5, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_nested_ifs(self): def func(): @@ -895,6 +1094,7 @@ class TraceTestCase(unittest.TestCase): (4, 'line'), (4, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_nested_ifs_with_and(self): def func(): @@ -918,6 +1118,7 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_nested_try_if(self): def func(): @@ -940,6 +1141,7 @@ class TraceTestCase(unittest.TestCase): (7, 'line'), (7, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_if_false_in_with(self): class C: @@ -966,6 +1168,7 @@ class TraceTestCase(unittest.TestCase): (-2, 'return'), (1, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_if_false_in_try_except(self): def func(): @@ -981,6 +1184,7 @@ class TraceTestCase(unittest.TestCase): (2, 'line'), (2, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_implicit_return_in_class(self): def func(): @@ -1000,6 +1204,7 @@ class TraceTestCase(unittest.TestCase): (3, 'return'), (1, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_try_in_try(self): def func(): try: @@ -1017,6 +1222,49 @@ class TraceTestCase(unittest.TestCase): (3, 'line'), (3, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") + def test_try_in_try_with_exception(self): + + def func(): + try: + try: + raise TypeError + except ValueError as ex: + 5 + except TypeError: + 7 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'exception'), + (4, 'line'), + (6, 'line'), + (7, 'line'), + (7, 'return')]) + + def func(): + try: + try: + raise ValueError + except ValueError as ex: + 5 + except TypeError: + 7 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'exception'), + (4, 'line'), + (5, 'line'), + (5, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") def test_if_in_if_in_if(self): def func(a=0, p=1, z=1): if p: @@ -1034,6 +1282,7 @@ class TraceTestCase(unittest.TestCase): (2, 'line'), (2, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_early_exit_with(self): class C: @@ -1080,6 +1329,7 @@ class TraceTestCase(unittest.TestCase): (-8, 'return'), (1, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_flow_converges_on_same_line(self): def foo(x): @@ -1116,6 +1366,7 @@ class TraceTestCase(unittest.TestCase): (1, 'line'), (1, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") def test_no_tracing_of_named_except_cleanup(self): def func(): @@ -1138,6 +1389,467 @@ class TraceTestCase(unittest.TestCase): (7, 'line'), (7, 'return')]) + @unittest.skip("TODO: RUSTPYTHON") + def test_tracing_exception_raised_in_with(self): + + class NullCtx: + def __enter__(self): + return self + def __exit__(self, *excinfo): + pass + + def func(): + try: + with NullCtx(): + 1/0 + except ZeroDivisionError: + pass + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (-5, 'call'), + (-4, 'line'), + (-4, 'return'), + (3, 'line'), + (3, 'exception'), + (2, 'line'), + (-3, 'call'), + (-2, 'line'), + (-2, 'return'), + (4, 'line'), + (5, 'line'), + (5, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_no_exception(self): + + def func(): + try: + 2 + except* Exception: + 4 + else: + 6 + if False: + 8 + else: + 10 + if func.__name__ == 'Fred': + 12 + finally: + 14 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (6, 'line'), + (7, 'line'), + (10, 'line'), + (11, 'line'), + (14, 'line'), + (14, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_named_no_exception(self): + + def func(): + try: + 2 + except* Exception as e: + 4 + else: + 6 + finally: + 8 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (6, 'line'), + (8, 'line'), + (8, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_exception_caught(self): + + def func(): + try: + raise ValueError(2) + except* ValueError: + 4 + else: + 6 + finally: + 8 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (2, 'exception'), + (3, 'line'), + (4, 'line'), + (8, 'line'), + (8, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_named_exception_caught(self): + + def func(): + try: + raise ValueError(2) + except* ValueError as e: + 4 + else: + 6 + finally: + 8 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (2, 'exception'), + (3, 'line'), + (4, 'line'), + (8, 'line'), + (8, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_exception_not_caught(self): + + def func(): + try: + try: + raise ValueError(3) + except* TypeError: + 5 + except ValueError: + 7 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'exception'), + (4, 'line'), + (6, 'line'), + (7, 'line'), + (7, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_named_exception_not_caught(self): + + def func(): + try: + try: + raise ValueError(3) + except* TypeError as e: + 5 + except ValueError: + 7 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'exception'), + (4, 'line'), + (6, 'line'), + (7, 'line'), + (7, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_try_except_star_nested(self): + + def func(): + try: + try: + raise ExceptionGroup( + 'eg', + [ValueError(5), TypeError('bad type')]) + except* TypeError as e: + 7 + except* OSError: + 9 + except* ValueError: + raise + except* ValueError: + try: + raise TypeError(14) + except* OSError: + 16 + except* TypeError as e: + 18 + return 0 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (4, 'line'), + (5, 'line'), + (3, 'line'), + (3, 'exception'), + (6, 'line'), + (7, 'line'), + (8, 'line'), + (10, 'line'), + (11, 'line'), + (12, 'line'), + (13, 'line'), + (14, 'line'), + (14, 'exception'), + (15, 'line'), + (17, 'line'), + (18, 'line'), + (19, 'line'), + (19, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_notrace_lambda(self): + #Regression test for issue 46314 + + def func(): + 1 + lambda x: 2 + 3 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_class_creation_with_docstrings(self): + + def func(): + class Class_1: + ''' the docstring. 2''' + def __init__(self): + ''' Another docstring. 4''' + self.a = 5 + + self.run_and_compare(func, + [(0, 'call'), + (1, 'line'), + (1, 'call'), + (1, 'line'), + (2, 'line'), + (3, 'line'), + (3, 'return'), + (1, 'return')]) + + @unittest.skip("TODO: RUSTPYTHON") + def test_class_creation_with_decorator(self): + def func(): + def decorator(arg): + def _dec(c): + return c + return _dec + + @decorator(6) + @decorator( + len([8]), + ) + class MyObject: + pass + + self.run_and_compare(func, [ + (0, 'call'), + (1, 'line'), + (6, 'line'), + (1, 'call'), + (2, 'line'), + (4, 'line'), + (4, 'return'), + (7, 'line'), + (8, 'line'), + (7, 'line'), + (1, 'call'), + (2, 'line'), + (4, 'line'), + (4, 'return'), + (10, 'line'), + (6, 'call'), + (6, 'line'), + (11, 'line'), + (11, 'return'), + (7, 'line'), + (2, 'call'), + (3, 'line'), + (3, 'return'), + (6, 'line'), + (2, 'call'), + (3, 'line'), + (3, 'return'), + (10, 'line'), + (10, 'return'), + ]) + + @support.cpython_only + def test_no_line_event_after_creating_generator(self): + # Spurious line events before call events only show up with C tracer + + # Skip this test if the _testcapi module isn't available. + _testcapi = import_helper.import_module('_testcapi') + + def gen(): + yield 1 + + def func(): + for _ in ( + gen() + ): + pass + + EXPECTED_EVENTS = [ + (0, 'call'), + (2, 'line'), + (-3, 'call'), + (-2, 'line'), + (-2, 'return'), + (1, 'line'), + (4, 'line'), + (2, 'line'), + (-2, 'call'), + (-2, 'return'), + (2, 'return'), + ] + + # C level events should be the same as expected and the same as Python level. + + events = [] + # Turning on and off tracing must be on same line to avoid unwanted LINE events. + _testcapi.settrace_to_record(events); func(); sys.settrace(None) + start_line = func.__code__.co_firstlineno + events = [ + (line-start_line, EVENT_NAMES[what]) + for (what, line, arg) in events + ] + self.assertEqual(events, EXPECTED_EVENTS) + + self.run_and_compare(func, EXPECTED_EVENTS) + + @unittest.skip("TODO: RUSTPYTHON") + def test_correct_tracing_quickened_call_class_init(self): + + class C: + def __init__(self): + self + + def func(): + C() + + EXPECTED_EVENTS = [ + (0, 'call'), + (1, 'line'), + (-3, 'call'), + (-2, 'line'), + (-2, 'return'), + (1, 'return')] + + self.run_and_compare(func, EXPECTED_EVENTS) + # Quicken + for _ in range(100): + func() + self.run_and_compare(func, EXPECTED_EVENTS) + + @unittest.skip("TODO: RUSTPYTHON") + def test_settrace_error(self): + raised = False + def error_once(frame, event, arg): + nonlocal raised + if not raised: + raised = True + raise Exception + return error + + try: + sys._getframe().f_trace = error_once + sys.settrace(error_once) + len([]) + except Exception as ex: + count = 0 + tb = ex.__traceback__ + while tb: + if tb.tb_frame.f_code.co_name == "test_settrace_error": + count += 1 + tb = tb.tb_next + if count == 0: + self.fail("Traceback is missing frame") + elif count > 1: + self.fail("Traceback has frame more than once") + else: + self.fail("No exception raised") + finally: + sys.settrace(None) + + @support.cpython_only + def test_testcapi_settrace_error(self): + + # Skip this test if the _testcapi module isn't available. + _testcapi = import_helper.import_module('_testcapi') + + try: + _testcapi.settrace_to_error([]) + len([]) + except Exception as ex: + count = 0 + tb = ex.__traceback__ + while tb: + if tb.tb_frame.f_code.co_name == "test_testcapi_settrace_error": + count += 1 + tb = tb.tb_next + if count == 0: + self.fail("Traceback is missing frame") + elif count > 1: + self.fail("Traceback has frame more than once") + else: + self.fail("No exception raised") + finally: + sys.settrace(None) + + @unittest.skip("TODO: RUSTPYTHON") + def test_very_large_function(self): + # There is a separate code path when the number of lines > (1 << 15). + d = {} + exec("""def f(): # line 0 + x = 0 # line 1 + y = 1 # line 2 + %s # lines 3 through (1 << 16) + x += 1 # + return""" % ('\n' * (1 << 16),), d) + f = d['f'] + + EXPECTED_EVENTS = [ + (0, 'call'), + (1, 'line'), + (2, 'line'), + (65540, 'line'), + (65541, 'line'), + (65541, 'return'), + ] + + self.run_and_compare(f, EXPECTED_EVENTS) + + +EVENT_NAMES = [ + 'call', + 'exception', + 'line', + 'return' +] + class SkipLineEventsTraceTestCase(TraceTestCase): """Repeat the trace tests, but with per-line events skipped""" @@ -1166,6 +1878,40 @@ class TraceOpcodesTestCase(TraceTestCase): def make_tracer(): return Tracer(trace_opcode_events=True) + @requires_subprocess() + def test_trace_opcodes_after_settrace(self): + """Make sure setting f_trace_opcodes after starting trace works even + if it's the first time f_trace_opcodes is being set. GH-103615""" + + code = textwrap.dedent(""" + import sys + + def opcode_trace_func(frame, event, arg): + if event == "opcode": + print("opcode trace triggered") + return opcode_trace_func + + sys.settrace(opcode_trace_func) + sys._getframe().f_trace = opcode_trace_func + sys._getframe().f_trace_opcodes = True + a = 1 + """) + + # We can't use context manager because Windows can't execute a file while + # it's being written + tmp = tempfile.NamedTemporaryFile(delete=False, suffix='.py') + tmp.write(code.encode('utf-8')) + tmp.close() + try: + p = subprocess.Popen([sys.executable, tmp.name], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + p.wait() + out = p.stdout.read() + finally: + os.remove(tmp.name) + p.stdout.close() + p.stderr.close() + self.assertIn(b"opcode trace triggered", out) + class RaisingTraceFuncTestCase(unittest.TestCase): def setUp(self): @@ -1205,16 +1951,16 @@ class RaisingTraceFuncTestCase(unittest.TestCase): self.fail("recursion counter not reset") # Test the handling of exceptions raised by each kind of trace event. - @unittest.skip("TODO: RUSTPYTHON, spurious error?") + @unittest.skip("TODO: RUSTPYTHON; spurious error?") def test_call(self): self.run_test_for_event('call') - @unittest.skip("TODO: RUSTPYTHON, spurious error?") + @unittest.skip("TODO: RUSTPYTHON; spurious error?") def test_line(self): self.run_test_for_event('line') - @unittest.skip("TODO: RUSTPYTHON, spurious error?") + @unittest.skip("TODO: RUSTPYTHON; spurious error?") def test_return(self): self.run_test_for_event('return') - @unittest.skip("TODO: RUSTPYTHON, spurious error?") + @unittest.skip("TODO: RUSTPYTHON; spurious error?") def test_exception(self): self.run_test_for_event('exception') @@ -1262,6 +2008,21 @@ class RaisingTraceFuncTestCase(unittest.TestCase): finally: sys.settrace(existing) + @unittest.skip("TODO: RUSTPYTHON") + def test_line_event_raises_before_opcode_event(self): + exception = ValueError("BOOM!") + def trace(frame, event, arg): + if event == "line": + raise exception + frame.f_trace_opcodes = True + return trace + def f(): + pass + with self.assertRaises(ValueError) as caught: + sys.settrace(trace) + f() + self.assertIs(caught.exception, exception) + # 'Jump' tests: assigning to frame.f_lineno within a trace function # moves the execution position - it's how debuggers implement a Jump @@ -1282,6 +2043,7 @@ class JumpTracer: def trace(self, frame, event, arg): if self.done: return + assert lineno_matches_lasti(frame) # frame.f_code.co_firstlineno is the first line of the decorator when # 'function' is decorated and the decorator may be written using # multiple physical lines when it is too long. Use the first line @@ -1328,6 +2090,8 @@ def no_jump_without_trace_function(): class JumpTestCase(unittest.TestCase): + unbound_locals = r"assigning None to [0-9]+ unbound local" + def setUp(self): self.addCleanup(sys.settrace, sys.gettrace()) sys.settrace(None) @@ -1339,33 +2103,49 @@ class JumpTestCase(unittest.TestCase): "Received: " + repr(received)) def run_test(self, func, jumpFrom, jumpTo, expected, error=None, - event='line', decorated=False): - tracer = JumpTracer(func, jumpFrom, jumpTo, event, decorated) + event='line', decorated=False, warning=None): + wrapped = func + while hasattr(wrapped, '__wrapped__'): + wrapped = wrapped.__wrapped__ + + tracer = JumpTracer(wrapped, jumpFrom, jumpTo, event, decorated) sys.settrace(tracer.trace) output = [] - if error is None: + + with contextlib.ExitStack() as stack: + if error is not None: + stack.enter_context(self.assertRaisesRegex(*error)) + if warning is not None: + stack.enter_context(self.assertWarnsRegex(*warning)) + else: + stack.enter_context(warnings.catch_warnings()) + warnings.simplefilter('error') func(output) - else: - with self.assertRaisesRegex(*error): - func(output) + sys.settrace(None) self.compare_jump_output(expected, output) def run_async_test(self, func, jumpFrom, jumpTo, expected, error=None, - event='line', decorated=False): - tracer = JumpTracer(func, jumpFrom, jumpTo, event, decorated) + event='line', decorated=False, warning=None): + wrapped = func + while hasattr(wrapped, '__wrapped__'): + wrapped = wrapped.__wrapped__ + + tracer = JumpTracer(wrapped, jumpFrom, jumpTo, event, decorated) sys.settrace(tracer.trace) output = [] - if error is None: - asyncio.run(func(output)) - else: - with self.assertRaisesRegex(*error): - asyncio.run(func(output)) + + with contextlib.ExitStack() as stack: + if error is not None: + stack.enter_context(self.assertRaisesRegex(*error)) + if warning is not None: + stack.enter_context(self.assertWarnsRegex(*warning)) + run_no_yield_async_fn(func, output) + sys.settrace(None) - asyncio.set_event_loop_policy(None) self.compare_jump_output(expected, output) - def jump_test(jumpFrom, jumpTo, expected, error=None, event='line'): + def jump_test(jumpFrom, jumpTo, expected, error=None, event='line', warning=None): """Decorator that creates a test that makes a jump from one place to another in the following code. """ @@ -1373,11 +2153,11 @@ class JumpTestCase(unittest.TestCase): @wraps(func) def test(self): self.run_test(func, jumpFrom, jumpTo, expected, - error=error, event=event, decorated=True) + error=error, event=event, decorated=True, warning=warning) return test return decorator - def async_jump_test(jumpFrom, jumpTo, expected, error=None, event='line'): + def async_jump_test(jumpFrom, jumpTo, expected, error=None, event='line', warning=None): """Decorator that creates a test that makes a jump from one place to another in the following asynchronous code. """ @@ -1385,24 +2165,65 @@ class JumpTestCase(unittest.TestCase): @wraps(func) def test(self): self.run_async_test(func, jumpFrom, jumpTo, expected, - error=error, event=event, decorated=True) + error=error, event=event, decorated=True, warning=warning) return test return decorator ## The first set of 'jump' tests are for things that are allowed: + @unittest.skip("TODO: RUSTPYTHON") @jump_test(1, 3, [3]) def test_jump_simple_forwards(output): output.append(1) output.append(2) output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 1, [1, 1, 2]) def test_jump_simple_backwards(output): output.append(1) output.append(2) - @jump_test(3, 5, [2, 5]) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 4, [5], warning=(RuntimeWarning, unbound_locals)) + def test_jump_is_none_forwards(output): + x = None + if x is None: + output.append(3) + else: + output.append(5) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(6, 5, [3, 5, 6]) + def test_jump_is_none_backwards(output): + x = None + if x is None: + output.append(3) + else: + output.append(5) + output.append(6) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(2, 4, [5]) + def test_jump_is_not_none_forwards(output): + x = None + if x is not None: + output.append(3) + else: + output.append(5) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(6, 5, [5, 5, 6]) + def test_jump_is_not_none_backwards(output): + x = None + if x is not None: + output.append(3) + else: + output.append(5) + output.append(6) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 5, [2, 5], warning=(RuntimeWarning, unbound_locals)) def test_jump_out_of_block_forwards(output): for i in 1, 2: output.append(2) @@ -1410,6 +2231,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(6, 1, [1, 3, 5, 1, 3, 5, 6, 7]) def test_jump_out_of_block_backwards(output): output.append(1) @@ -1420,16 +2242,20 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(4, 5, [3, 5]) - async def test_jump_out_of_async_for_block_forwards(output): + @clean_asynciter + async def test_jump_out_of_async_for_block_forwards(output, asynciter): for i in [1]: async for i in asynciter([1, 2]): output.append(3) output.append(4) output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(5, 2, [2, 4, 2, 4, 5, 6]) - async def test_jump_out_of_async_for_block_backwards(output): + @clean_asynciter + async def test_jump_out_of_async_for_block_backwards(output, asynciter): for i in [1]: output.append(2) async for i in asynciter([1]): @@ -1437,12 +2263,14 @@ class JumpTestCase(unittest.TestCase): output.append(5) output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(1, 2, [3]) def test_jump_to_codeless_line(output): output.append(1) # Jumping to this line should skip to the next one. output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 2, [1, 2, 3]) def test_jump_to_same_line(output): output.append(1) @@ -1450,6 +2278,7 @@ class JumpTestCase(unittest.TestCase): output.append(3) # Tests jumping within a finally block, and over one. + @unittest.skip("TODO: RUSTPYTHON") @jump_test(4, 9, [2, 9]) def test_jump_in_nested_finally(output): try: @@ -1462,6 +2291,7 @@ class JumpTestCase(unittest.TestCase): output.append(8) output.append(9) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(6, 7, [2, 7], (ZeroDivisionError, '')) def test_jump_in_nested_finally_2(output): try: @@ -1473,6 +2303,7 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(6, 11, [2, 11], (ZeroDivisionError, '')) def test_jump_in_nested_finally_3(output): try: @@ -1488,22 +2319,7 @@ class JumpTestCase(unittest.TestCase): output.append(11) output.append(12) - @jump_test(5, 11, [2, 4], (ValueError, 'after')) - def test_no_jump_over_return_try_finally_in_finally_block(output): - try: - output.append(2) - finally: - output.append(4) - output.append(5) - return - try: - output.append(8) - finally: - output.append(10) - pass - output.append(12) - - @unittest.skip("TODO: RUSTPYTHON, infinite loop never exits") + @unittest.skip("TODO: RUSTPYTHON; infinite loop never exits") @jump_test(3, 4, [1], (ValueError, 'after')) def test_no_jump_infinite_while_loop(output): output.append(1) @@ -1511,6 +2327,7 @@ class JumpTestCase(unittest.TestCase): output.append(3) output.append(4) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 4, [4, 4]) def test_jump_forwards_into_while_block(output): i = 1 @@ -1519,6 +2336,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) i += 1 + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 3, [3, 3, 3, 5]) def test_jump_backwards_into_while_block(output): i = 1 @@ -1527,30 +2345,35 @@ class JumpTestCase(unittest.TestCase): i += 1 output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [1, 3]) def test_jump_forwards_out_of_with_block(output): with tracecontext(output, 1): output.append(2) output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(2, 3, [1, 3]) async def test_jump_forwards_out_of_async_with_block(output): async with asynctracecontext(output, 1): output.append(2) output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 1, [1, 2, 1, 2, 3, -2]) def test_jump_backwards_out_of_with_block(output): output.append(1) with tracecontext(output, 2): output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(3, 1, [1, 2, 1, 2, 3, -2]) async def test_jump_backwards_out_of_async_with_block(output): output.append(1) async with asynctracecontext(output, 2): output.append(3) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 5, [5]) def test_jump_forwards_out_of_try_finally_block(output): try: @@ -1559,6 +2382,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 1, [1, 1, 3, 5]) def test_jump_backwards_out_of_try_finally_block(output): output.append(1) @@ -1567,6 +2391,7 @@ class JumpTestCase(unittest.TestCase): finally: output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 6, [6]) def test_jump_forwards_out_of_try_except_block(output): try: @@ -1576,6 +2401,7 @@ class JumpTestCase(unittest.TestCase): raise output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 1, [1, 1, 3]) def test_jump_backwards_out_of_try_except_block(output): output.append(1) @@ -1585,6 +2411,7 @@ class JumpTestCase(unittest.TestCase): output.append(5) raise + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 7, [4, 7, 8]) def test_jump_between_except_blocks(output): try: @@ -1596,6 +2423,19 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(5, 7, [4, 7, 8]) + def test_jump_from_except_to_finally(output): + try: + 1/0 + except ZeroDivisionError: + output.append(4) + output.append(5) + finally: + output.append(7) + output.append(8) + + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 6, [4, 6, 7]) def test_jump_within_except_block(output): try: @@ -1606,6 +2446,17 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(6, 1, [1, 5, 1, 5], warning=(RuntimeWarning, unbound_locals)) + def test_jump_over_try_except(output): + output.append(1) + try: + 1 / 0 + except ZeroDivisionError as e: + output.append(5) + x = 42 # has to be a two-instruction block + + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 4, [1, 4, 5, -4]) def test_jump_across_with(output): output.append(1) @@ -1614,6 +2465,7 @@ class JumpTestCase(unittest.TestCase): with tracecontext(output, 4): output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(2, 4, [1, 4, 5, -4]) async def test_jump_across_async_with(output): output.append(1) @@ -1622,6 +2474,7 @@ class JumpTestCase(unittest.TestCase): async with asynctracecontext(output, 4): output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(4, 5, [1, 3, 5, 6]) def test_jump_out_of_with_block_within_for_block(output): output.append(1) @@ -1631,6 +2484,7 @@ class JumpTestCase(unittest.TestCase): output.append(5) output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(4, 5, [1, 3, 5, 6]) async def test_jump_out_of_async_with_block_within_for_block(output): output.append(1) @@ -1640,6 +2494,7 @@ class JumpTestCase(unittest.TestCase): output.append(5) output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(4, 5, [1, 2, 3, 5, -2, 6]) def test_jump_out_of_with_block_within_with_block(output): output.append(1) @@ -1649,6 +2504,7 @@ class JumpTestCase(unittest.TestCase): output.append(5) output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(4, 5, [1, 2, 3, 5, -2, 6]) async def test_jump_out_of_async_with_block_within_with_block(output): output.append(1) @@ -1658,6 +2514,7 @@ class JumpTestCase(unittest.TestCase): output.append(5) output.append(6) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 6, [2, 4, 6, 7]) def test_jump_out_of_with_block_within_finally_block(output): try: @@ -1668,6 +2525,7 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(5, 6, [2, 4, 6, 7]) async def test_jump_out_of_async_with_block_within_finally_block(output): try: @@ -1678,6 +2536,7 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(8, 11, [1, 3, 5, 11, 12]) def test_jump_out_of_complex_nested_blocks(output): output.append(1) @@ -1693,7 +2552,8 @@ class JumpTestCase(unittest.TestCase): output.append(11) output.append(12) - @jump_test(3, 5, [1, 2, 5]) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 5, [1, 2, 5], warning=(RuntimeWarning, unbound_locals)) def test_jump_out_of_with_assignment(output): output.append(1) with tracecontext(output, 2) \ @@ -1701,7 +2561,8 @@ class JumpTestCase(unittest.TestCase): output.append(4) output.append(5) - @async_jump_test(3, 5, [1, 2, 5]) + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(3, 5, [1, 2, 5], warning=(RuntimeWarning, unbound_locals)) async def test_jump_out_of_async_with_assignment(output): output.append(1) async with asynctracecontext(output, 2) \ @@ -1709,6 +2570,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 6, [1, 6, 8, 9]) def test_jump_over_return_in_try_finally_block(output): output.append(1) @@ -1721,6 +2583,7 @@ class JumpTestCase(unittest.TestCase): output.append(8) output.append(9) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 8, [1, 3, 8, 10, 11, 13]) def test_jump_over_break_in_try_finally_block(output): output.append(1) @@ -1737,7 +2600,8 @@ class JumpTestCase(unittest.TestCase): break output.append(13) - @jump_test(1, 7, [7, 8]) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 7, [7, 8], warning=(RuntimeWarning, unbound_locals)) def test_jump_over_for_block_before_else(output): output.append(1) if not output: # always false @@ -1748,7 +2612,8 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) - @async_jump_test(1, 7, [7, 8]) + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(1, 7, [7, 8], warning=(RuntimeWarning, unbound_locals)) async def test_jump_over_async_for_block_before_else(output): output.append(1) if not output: # always false @@ -1761,19 +2626,20 @@ class JumpTestCase(unittest.TestCase): # The second set of 'jump' tests are for things that are not allowed: + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [1], (ValueError, 'after')) def test_no_jump_too_far_forwards(output): output.append(1) output.append(2) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, -2, [1], (ValueError, 'before')) def test_no_jump_too_far_backwards(output): output.append(1) output.append(2) - # TODO: RUSTPYTHON - @unittest.expectedFailure # Test each kind of 'except' line. + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [4], (ValueError, 'except')) def test_no_jump_to_except_1(output): try: @@ -1782,8 +2648,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) raise - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [4], (ValueError, 'except')) def test_no_jump_to_except_2(output): try: @@ -1792,8 +2657,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) raise - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [4], (ValueError, 'except')) def test_no_jump_to_except_3(output): try: @@ -1802,8 +2666,7 @@ class JumpTestCase(unittest.TestCase): output.append(4) raise e - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [4], (ValueError, 'except')) def test_no_jump_to_except_4(output): try: @@ -1812,14 +2675,14 @@ class JumpTestCase(unittest.TestCase): output.append(4) raise e + @unittest.skip("TODO: RUSTPYTHON") @jump_test(1, 3, [], (ValueError, 'into')) def test_no_jump_forwards_into_for_block(output): output.append(1) for i in 1, 2: output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(1, 3, [], (ValueError, 'into')) async def test_no_jump_forwards_into_async_for_block(output): output.append(1) @@ -1827,76 +2690,70 @@ class JumpTestCase(unittest.TestCase): output.append(3) pass + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 2, [2, 2], (ValueError, 'into')) def test_no_jump_backwards_into_for_block(output): for i in 1, 2: output.append(2) output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @async_jump_test(3, 2, [2, 2], (ValueError, 'into')) + + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(3, 2, [2, 2], (ValueError, "can't jump into the body of a for loop")) async def test_no_jump_backwards_into_async_for_block(output): async for i in asynciter([1, 2]): output.append(2) output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(1, 3, [], (ValueError, 'into')) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 3, [], (ValueError, 'stack')) def test_no_jump_forwards_into_with_block(output): output.append(1) with tracecontext(output, 2): output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @async_jump_test(1, 3, [], (ValueError, 'into')) + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(1, 3, [], (ValueError, 'stack')) async def test_no_jump_forwards_into_async_with_block(output): output.append(1) async with asynctracecontext(output, 2): output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(3, 2, [1, 2, -1], (ValueError, 'into')) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 2, [1, 2, -1], (ValueError, 'stack')) def test_no_jump_backwards_into_with_block(output): with tracecontext(output, 1): output.append(2) output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @async_jump_test(3, 2, [1, 2, -1], (ValueError, 'into')) + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(3, 2, [1, 2, -1], (ValueError, 'stack')) async def test_no_jump_backwards_into_async_with_block(output): async with asynctracecontext(output, 1): output.append(2) output.append(3) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(1, 3, [], (ValueError, 'into')) - def test_no_jump_forwards_into_try_finally_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 3, [3, 5]) + def test_jump_forwards_into_try_finally_block(output): output.append(1) try: output.append(3) finally: output.append(5) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(5, 2, [2, 4], (ValueError, 'into')) - def test_no_jump_backwards_into_try_finally_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(5, 2, [2, 4, 2, 4, 5]) + def test_jump_backwards_into_try_finally_block(output): try: output.append(2) finally: output.append(4) output.append(5) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(1, 3, [], (ValueError, 'into')) - def test_no_jump_forwards_into_try_except_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 3, [3]) + def test_jump_forwards_into_try_except_block(output): output.append(1) try: output.append(3) @@ -1904,10 +2761,9 @@ class JumpTestCase(unittest.TestCase): output.append(5) raise - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(6, 2, [2], (ValueError, 'into')) - def test_no_jump_backwards_into_try_except_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(6, 2, [2, 2, 6]) + def test_jump_backwards_into_try_except_block(output): try: output.append(2) except: @@ -1915,11 +2771,10 @@ class JumpTestCase(unittest.TestCase): raise output.append(6) - # TODO: RUSTPYTHON - @unittest.expectedFailure # 'except' with a variable creates an implicit finally block - @jump_test(5, 7, [4], (ValueError, 'into')) - def test_no_jump_between_except_blocks_2(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(5, 7, [4, 7, 8], warning=(RuntimeWarning, unbound_locals)) + def test_jump_between_except_blocks_2(output): try: 1/0 except ZeroDivisionError: @@ -1929,6 +2784,7 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(1, 5, [5]) def test_jump_into_finally_block(output): output.append(1) @@ -1937,6 +2793,7 @@ class JumpTestCase(unittest.TestCase): finally: output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 6, [2, 6, 7]) def test_jump_into_finally_block_from_try_block(output): try: @@ -1947,6 +2804,7 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 1, [1, 3, 1, 3, 5]) def test_jump_out_of_finally_block(output): output.append(1) @@ -1955,7 +2813,8 @@ class JumpTestCase(unittest.TestCase): finally: output.append(5) - @jump_test(1, 5, [], (ValueError, "into an 'except'")) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 5, [], (ValueError, "can't jump into an 'except' block as there's no exception")) def test_no_jump_into_bare_except_block(output): output.append(1) try: @@ -1963,7 +2822,8 @@ class JumpTestCase(unittest.TestCase): except: output.append(5) - @jump_test(1, 5, [], (ValueError, "into an 'except'")) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(1, 5, [], (ValueError, "can't jump into an 'except' block as there's no exception")) def test_no_jump_into_qualified_except_block(output): output.append(1) try: @@ -1971,9 +2831,8 @@ class JumpTestCase(unittest.TestCase): except Exception: output.append(5) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(3, 6, [2, 5, 6], (ValueError, "into an 'except'")) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 6, [2, 5, 6], (ValueError, "can't jump into an 'except' block as there's no exception")) def test_no_jump_into_bare_except_block_from_try_block(output): try: output.append(2) @@ -1984,7 +2843,8 @@ class JumpTestCase(unittest.TestCase): raise output.append(8) - @jump_test(3, 6, [2], (ValueError, "into an 'except'")) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 6, [2], (ValueError, "can't jump into an 'except' block as there's no exception")) def test_no_jump_into_qualified_except_block_from_try_block(output): try: output.append(2) @@ -1995,10 +2855,9 @@ class JumpTestCase(unittest.TestCase): raise output.append(8) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(7, 1, [1, 3, 6], (ValueError, "out of an 'except'")) - def test_no_jump_out_of_bare_except_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(7, 1, [1, 3, 6, 1, 3, 6, 7]) + def test_jump_out_of_bare_except_block(output): output.append(1) try: output.append(3) @@ -2007,10 +2866,9 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) - # TODO: RUSTPYTHON - @unittest.expectedFailure - @jump_test(7, 1, [1, 3, 6], (ValueError, "out of an 'except'")) - def test_no_jump_out_of_qualified_except_block(output): + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(7, 1, [1, 3, 6, 1, 3, 6, 7]) + def test_jump_out_of_qualified_except_block(output): output.append(1) try: output.append(3) @@ -2019,6 +2877,7 @@ class JumpTestCase(unittest.TestCase): output.append(6) output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(3, 5, [1, 2, 5, -2]) def test_jump_between_with_blocks(output): output.append(1) @@ -2027,6 +2886,7 @@ class JumpTestCase(unittest.TestCase): with tracecontext(output, 4): output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(3, 5, [1, 2, 5, -2]) async def test_jump_between_async_with_blocks(output): output.append(1) @@ -2035,6 +2895,7 @@ class JumpTestCase(unittest.TestCase): async with asynctracecontext(output, 4): output.append(5) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(5, 7, [2, 4], (ValueError, "after")) def test_no_jump_over_return_out_of_finally_block(output): try: @@ -2042,9 +2903,10 @@ class JumpTestCase(unittest.TestCase): finally: output.append(4) output.append(5) - return + return output.append(7) + @unittest.skip("TODO: RUSTPYTHON") @jump_test(7, 4, [1, 6], (ValueError, 'into')) def test_no_jump_into_for_block_before_else(output): output.append(1) @@ -2056,8 +2918,7 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @async_jump_test(7, 4, [1, 6], (ValueError, 'into')) async def test_no_jump_into_async_for_block_before_else(output): output.append(1) @@ -2069,18 +2930,17 @@ class JumpTestCase(unittest.TestCase): output.append(7) output.append(8) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") def test_no_jump_to_non_integers(self): self.run_test(no_jump_to_non_integers, 2, "Spam", [True]) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") def test_no_jump_without_trace_function(self): # Must set sys.settrace(None) in setUp(), else condition is not # triggered. no_jump_without_trace_function() + @unittest.skip("TODO: RUSTPYTHON") def test_large_function(self): d = {} exec("""def f(output): # line 0 @@ -2093,8 +2953,9 @@ class JumpTestCase(unittest.TestCase): output.append(x) # line 1007 return""" % ('\n' * 1000,), d) f = d['f'] - self.run_test(f, 2, 1007, [0]) + self.run_test(f, 2, 1007, [0], warning=(RuntimeWarning, self.unbound_locals)) + @unittest.skip("TODO: RUSTPYTHON") def test_jump_to_firstlineno(self): # This tests that PDB can jump back to the first line in a # file. See issue #1689458. It can only be triggered in a @@ -2114,8 +2975,7 @@ output.append(4) sys.settrace(None) self.compare_jump_output([2, 3, 2, 3, 4], namespace["output"]) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 3, [1], event='call', error=(ValueError, "can't jump from" " the 'call' trace event of a new frame")) def test_no_jump_from_call(output): @@ -2125,23 +2985,22 @@ output.append(4) nested() output.append(5) - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 1, [1], event='return', error=(ValueError, "can only jump from a 'line' trace event")) def test_no_jump_from_return_event(output): output.append(1) return - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.skip("TODO: RUSTPYTHON") @jump_test(2, 1, [1], event='exception', error=(ValueError, "can only jump from a 'line' trace event")) def test_no_jump_from_exception_event(output): output.append(1) 1 / 0 - @jump_test(3, 2, [2, 5], event='return') + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 2, [2, 2, 5], event='return') def test_jump_from_yield(output): def gen(): output.append(2) @@ -2149,6 +3008,300 @@ output.append(4) next(gen()) output.append(5) + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(2, 3, [1, 3], warning=(RuntimeWarning, unbound_locals)) + def test_jump_forward_over_listcomp(output): + output.append(1) + x = [i for i in range(10)] + output.append(3) + + # checking for segfaults. + # See https://github.com/python/cpython/issues/92311 + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 1, [], warning=(RuntimeWarning, unbound_locals)) + def test_jump_backward_over_listcomp(output): + a = 1 + x = [i for i in range(10)] + c = 3 + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(8, 2, [2, 7, 2], warning=(RuntimeWarning, unbound_locals)) + def test_jump_backward_over_listcomp_v2(output): + flag = False + output.append(2) + if flag: + return + x = [i for i in range(5)] + flag = 6 + output.append(7) + output.append(8) + + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(2, 3, [1, 3], warning=(RuntimeWarning, unbound_locals)) + async def test_jump_forward_over_async_listcomp(output): + output.append(1) + x = [i async for i in asynciter(range(10))] + output.append(3) + + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(3, 1, [], warning=(RuntimeWarning, unbound_locals)) + async def test_jump_backward_over_async_listcomp(output): + a = 1 + x = [i async for i in asynciter(range(10))] + c = 3 + + @unittest.skip("TODO: RUSTPYTHON") + @async_jump_test(8, 2, [2, 7, 2], warning=(RuntimeWarning, unbound_locals)) + async def test_jump_backward_over_async_listcomp_v2(output): + flag = False + output.append(2) + if flag: + return + x = [i async for i in asynciter(range(5))] + flag = 6 + output.append(7) + output.append(8) + + # checking for segfaults. + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 7, [], error=(ValueError, "stack")) + def test_jump_with_null_on_stack_load_global(output): + a = 1 + print( + output.append(3) + ) + output.append(5) + ( + ( # 7 + a + + + 10 + ) + + + 13 + ) + output.append(15) + + # checking for segfaults. + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(4, 8, [], error=(ValueError, "stack")) + def test_jump_with_null_on_stack_push_null(output): + a = 1 + f = print + f( + output.append(4) + ) + output.append(6) + ( + ( # 8 + a + + + 11 + ) + + + 14 + ) + output.append(16) + + # checking for segfaults. + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 7, [], error=(ValueError, "stack")) + def test_jump_with_null_on_stack_load_attr(output): + a = 1 + list.append( + output, 3 + ) + output.append(5) + ( + ( # 7 + a + + + 10 + ) + + + 13 + ) + output.append(15) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(2, 3, [1, 3], warning=(RuntimeWarning, unbound_locals)) + def test_jump_extended_args_unpack_ex_simple(output): + output.append(1) + _, *_, _ = output.append(2) or "Spam" + output.append(3) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(3, 4, [1, 4, 4, 5], warning=(RuntimeWarning, unbound_locals)) + def test_jump_extended_args_unpack_ex_tricky(output): + output.append(1) + ( + _, *_, _ + ) = output.append(4) or "Spam" + output.append(5) + + @unittest.skip("TODO: RUSTPYTHON") + @support.requires_resource('cpu') + def test_jump_extended_args_for_iter(self): + # In addition to failing when extended arg handling is broken, this can + # also hang for a *very* long time: + source = [ + "def f(output):", + " output.append(1)", + " for _ in spam:", + *(f" output.append({i})" for i in range(3, 100_000)), + f" output.append(100_000)", + ] + namespace = {} + exec("\n".join(source), namespace) + f = namespace["f"] + self.run_test(f, 2, 100_000, [1, 100_000], warning=(RuntimeWarning, self.unbound_locals)) + + @unittest.skip("TODO: RUSTPYTHON") + @jump_test(2, 3, [1, 3], warning=(RuntimeWarning, unbound_locals)) + def test_jump_or_pop(output): + output.append(1) + _ = output.append(2) and "Spam" + output.append(3) + + +class TestExtendedArgs(unittest.TestCase): + + def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) + sys.settrace(None) + + def count_traces(self, func): + # warmup + for _ in range(20): + func() + + counts = {"call": 0, "line": 0, "return": 0} + def trace(frame, event, arg): + counts[event] += 1 + return trace + + sys.settrace(trace) + func() + sys.settrace(None) + + return counts + + def test_trace_unpack_long_sequence(self): + ns = {} + code = "def f():\n (" + "y,\n "*300 + ") = range(300)" + exec(code, ns) + counts = self.count_traces(ns["f"]) + self.assertEqual(counts, {'call': 1, 'line': 301, 'return': 1}) + + def test_trace_lots_of_globals(self): + + count = 1000 + + code = """if 1: + def f(): + return ( + {} + ) + """.format("\n,\n".join(f"var{i}\n" for i in range(count))) + ns = {f"var{i}": i for i in range(count)} + exec(code, ns) + counts = self.count_traces(ns["f"]) + self.assertEqual(counts, {'call': 1, 'line': count * 2 + 1, 'return': 1}) + + +class TestEdgeCases(unittest.TestCase): + + def setUp(self): + self.addCleanup(sys.settrace, sys.gettrace()) + sys.settrace(None) + + def test_reentrancy(self): + def foo(*args): + ... + + def bar(*args): + ... + + class A: + def __call__(self, *args): + pass + + def __del__(self): + sys.settrace(bar) + + sys.settrace(A()) + sys.settrace(foo) + self.assertEqual(sys.gettrace(), bar) + + + def test_same_object(self): + def foo(*args): + ... + + sys.settrace(foo) + del foo + sys.settrace(sys.gettrace()) + + +class TestLinesAfterTraceStarted(TraceTestCase): + + @unittest.skip("TODO: RUSTPYTHON") + def test_events(self): + tracer = Tracer() + sys._getframe().f_trace = tracer.trace + sys.settrace(tracer.trace) + line = 4 + line = 5 + sys.settrace(None) + self.compare_events( + TestLinesAfterTraceStarted.test_events.__code__.co_firstlineno, + tracer.events, [ + (4, 'line'), + (5, 'line'), + (6, 'line')]) + + +class TestSetLocalTrace(TraceTestCase): + + def test_with_branches(self): + + def tracefunc(frame, event, arg): + if frame.f_code.co_name == "func": + frame.f_trace = tracefunc + line = frame.f_lineno - frame.f_code.co_firstlineno + events.append((line, event)) + return tracefunc + + def func(arg = 1): + N = 1 + if arg >= 2: + not_reached = 3 + else: + reached = 5 + if arg >= 3: + not_reached = 7 + else: + reached = 9 + the_end = 10 + + EXPECTED_EVENTS = [ + (0, 'call'), + (1, 'line'), + (2, 'line'), + (5, 'line'), + (6, 'line'), + (9, 'line'), + (10, 'line'), + (10, 'return'), + ] + + events = [] + sys.settrace(tracefunc) + sys._getframe().f_trace = tracefunc + func() + self.assertEqual(events, EXPECTED_EVENTS) + sys.settrace(None) + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_syslog.py b/Lib/test/test_syslog.py index 9ad2cc518..040800066 100644 --- a/Lib/test/test_syslog.py +++ b/Lib/test/test_syslog.py @@ -5,6 +5,7 @@ import sys import threading import time import unittest +from textwrap import dedent # XXX(nnorwitz): This test sucks. I don't know of a platform independent way # to verify that the messages were really logged. @@ -15,8 +16,7 @@ class Test(unittest.TestCase): def tearDown(self): syslog.closelog() - # TODO: RUSTPYTHON - @unittest.expectedFailure + @unittest.expectedFailure # TODO: RUSTPYTHON def test_openlog(self): syslog.openlog('python') # Issue #6697. @@ -80,6 +80,69 @@ class Test(unittest.TestCase): finally: sys.setswitchinterval(orig_si) + def test_subinterpreter_syslog(self): + # syslog.syslog() is not allowed in subinterpreters, but only if + # syslog.openlog() hasn't been called in the main interpreter yet. + with self.subTest('before openlog()'): + code = dedent(''' + import syslog + caught_error = False + try: + syslog.syslog('foo') + except RuntimeError: + caught_error = True + assert(caught_error) + ''') + res = support.run_in_subinterp(code) + self.assertEqual(res, 0) + + syslog.openlog() + try: + with self.subTest('after openlog()'): + code = dedent(''' + import syslog + syslog.syslog('foo') + ''') + res = support.run_in_subinterp(code) + self.assertEqual(res, 0) + finally: + syslog.closelog() + + def test_subinterpreter_openlog(self): + try: + code = dedent(''' + import syslog + caught_error = False + try: + syslog.openlog() + except RuntimeError: + caught_error = True + + assert(caught_error) + ''') + res = support.run_in_subinterp(code) + self.assertEqual(res, 0) + finally: + syslog.closelog() + + def test_subinterpreter_closelog(self): + syslog.openlog('python') + try: + code = dedent(''' + import syslog + caught_error = False + try: + syslog.closelog() + except RuntimeError: + caught_error = True + + assert(caught_error) + ''') + res = support.run_in_subinterp(code) + self.assertEqual(res, 0) + finally: + syslog.closelog() + if __name__ == "__main__": unittest.main()