From ff0f7c0f1dab91241d61fe78955f7522ee04a7a8 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 20:39:15 +0200 Subject: [PATCH 01/13] test(rewrite): systematic coverage matrix for assertion rewriting The rewriter is tested through its failure messages, one regression test per issue. That makes it hard to say whether a change improved anything: there is no place that records what the rewriter cannot do yet. Add a matrix over expression types with four axes -- introspection depth, semantic equivalence, single evaluation, and evaluation order -- and record every known gap as a strict xfail tagged with a group name. A change that closes one names the group and flips its markers; strictness means it cannot forget. The evaluation-order axis is new. Comparing pass/fail is not enough: an operand read after a later walrus operator rebound its name runs exactly once and can still fail the assertion, having compared the wrong value. assert_evaluation_order() compares what check() returns, because the fragile operand is a bare name -- wrapping it in a call to observe it would hoist it and hide the bug. Behaviour is unchanged; this only describes it. --- changelog/14813.contrib.rst | 1 + testing/test_assertrewrite_coverage.py | 1677 ++++++++++++++++++++++++ 2 files changed, 1678 insertions(+) create mode 100644 changelog/14813.contrib.rst create mode 100644 testing/test_assertrewrite_coverage.py diff --git a/changelog/14813.contrib.rst b/changelog/14813.contrib.rst new file mode 100644 index 00000000000..4d41ab766da --- /dev/null +++ b/changelog/14813.contrib.rst @@ -0,0 +1 @@ +Added a systematic coverage matrix for assertion rewriting in ``testing/test_assertrewrite_coverage.py``, checking introspection depth, semantic equivalence, single evaluation and evaluation order across expression types. Known gaps are recorded as strict xfails tagged with a group name, so a change can state which group it closes. diff --git a/testing/test_assertrewrite_coverage.py b/testing/test_assertrewrite_coverage.py new file mode 100644 index 00000000000..054db3bb137 --- /dev/null +++ b/testing/test_assertrewrite_coverage.py @@ -0,0 +1,1677 @@ +"""Systematic coverage tests for assertion rewriting. + +This module provides a structured testing framework that verifies assertion +rewriting behavior across all expression types, checking: + +1. Introspection depth: failure messages contain expected intermediate values +2. Semantic correctness: rewritten code has identical behavior to original +3. Single evaluation: side-effecting expressions are not evaluated multiple times +4. Evaluation order: operands see the values Python would give them + +Known gaps are recorded as strict xfails whose reason starts with a group name, +so that a change can state which group it closes:: + + introspect-subscript container[key] is not decomposed + introspect-ifexp ternaries are not decomposed + introspect-method-call-flat obj.method() shows the bound method separately + introspect-container-literal list/dict/set literals are not decomposed + introspect-callable-variable a called variable shows + single-eval-walrus a walrus operator runs more than once + order-compare-left a comparison's left operand is read too late + order-call-argument an earlier argument is read too late + order-binop-left a binary operator's left operand is read too late + order-starred-argument a starred argument is read too late + order-name-rebound-by-call a call rebinding a name is not accounted for + order-chained-compare-lazy a chained comparison evaluates every comparator + +The xfails are strict on purpose: fixing the behaviour without flipping the +marker fails the suite. A change may remove entries from that list, never add +one -- a newly discovered gap needs an issue to point at first. + +The helpers here take source text rather than a function object, unlike +``getmsg()`` in ``test_assertrewrite.py``, because the axes above need to +compile the same text twice -- once rewritten and once not. +""" + +from __future__ import annotations + +import ast +from collections.abc import Callable +from collections.abc import Mapping +import copy +import sys +import textwrap +from typing import cast + +from _pytest.assertion.rewrite import rewrite_asserts +import pytest + + +# --------------------------------------------------------------------------- +# Test helpers +# --------------------------------------------------------------------------- + + +def _rewrite_source(src: str) -> ast.Module: + """Parse and rewrite assertions in source code.""" + tree = ast.parse(src) + rewrite_asserts(tree, src.encode()) + return tree + + +def get_failure_message( + src: str, + extra_ns: Mapping[str, object] | None = None, +) -> str: + """Compile rewritten source, execute it, and return the failure message. + + The source should contain a function named ``check`` with a failing assert. + Returns the AssertionError message string. + + Raises AssertionError via pytest.fail if the code does not raise. + """ + src = textwrap.dedent(src) + mod = _rewrite_source(src) + code = compile(mod, "", "exec") + ns: dict[str, object] = {} + if extra_ns is not None: + ns.update(extra_ns) + exec(code, ns) + func = cast(Callable[[], None], ns["check"]) + try: + func() + except AssertionError: + s = str(sys.exc_info()[1]) + if not s.startswith("assert"): + return "AssertionError: " + s + return s + else: + pytest.fail("check() did not raise AssertionError") + + +def assert_introspects( + src: str, + *, + must_contain: list[str], + must_not_contain: list[str] | None = None, + extra_ns: Mapping[str, object] | None = None, +) -> str: + """Verify a failing assert produces a message with expected intermediate values. + + Parameters + ---------- + src : str + Source code containing a ``check()`` function with a failing assertion. + must_contain : list[str] + Substrings that MUST appear in the failure message. + must_not_contain : list[str] | None + Substrings that must NOT appear in the failure message. + extra_ns : Mapping[str, object] | None + Additional namespace entries available during execution. + + Returns + ------- + str + The full failure message (for further inspection if needed). + """ + msg = get_failure_message(src, extra_ns=extra_ns) + for expected in must_contain: + assert expected in msg, ( + f"Expected {expected!r} in failure message.\nGot:\n{msg}" + ) + for unexpected in must_not_contain or []: + assert unexpected not in msg, ( + f"Did NOT expect {unexpected!r} in failure message.\nGot:\n{msg}" + ) + return msg + + +def assert_single_evaluation( + src: str, + *, + expected_call_count: int = 1, + extra_ns: Mapping[str, object] | None = None, +) -> None: + """Verify side-effecting expressions in assert are evaluated exactly once. + + The source should define a ``check()`` function and use a ``counter`` list + (provided via extra_ns or defined in the source) that tracks how many times + a side-effecting expression is evaluated. + + Parameters + ---------- + src : str + Source containing a ``check()`` function whose assert has side effects. + expected_call_count : int + How many times the side-effecting expression should be evaluated. + extra_ns : Mapping[str, object] | None + Additional namespace. Should include ``counter`` if not defined in src. + """ + src = textwrap.dedent(src) + mod = _rewrite_source(src) + code = compile(mod, "", "exec") + ns: dict[str, object] = {"counter": [0]} + if extra_ns is not None: + ns.update(extra_ns) + exec(code, ns) + func = cast(Callable[[], None], ns["check"]) + counter = cast(list[int], ns["counter"]) + counter[0] = 0 + try: + func() + except AssertionError: + pass + actual = counter[0] + assert actual == expected_call_count, ( + f"Expression evaluated {actual} times, expected {expected_call_count}" + ) + + +def assert_passes_when_true( + src: str, + *, + extra_ns: Mapping[str, object] | None = None, +) -> None: + """Verify rewritten assertion does not raise when the condition is true. + + Parameters + ---------- + src : str + Source containing a ``check()`` function with a passing assertion. + extra_ns : Mapping[str, object] | None + Additional namespace entries available during execution. + """ + src = textwrap.dedent(src) + mod = _rewrite_source(src) + code = compile(mod, "", "exec") + ns: dict[str, object] = {} + if extra_ns is not None: + ns.update(extra_ns) + exec(code, ns) + func = cast(Callable[[], None], ns["check"]) + func() + + +def assert_semantically_equivalent( + src: str, + *, + extra_ns: Mapping[str, object] | None = None, +) -> None: + """Verify rewritten code has same pass/fail semantics as unrewritten code. + + Runs the source both with and without rewriting, and asserts they agree + on whether an AssertionError is raised. + + Parameters + ---------- + src : str + Source containing a ``check()`` function with an assertion. + extra_ns : Mapping[str, object] | None + Additional namespace entries available during execution. + """ + src = textwrap.dedent(src) + + # Run without rewriting — use deepcopy of extra_ns to isolate mutable state + plain_code = compile(src, "", "exec") + plain_ns: dict[str, object] = {} + if extra_ns is not None: + plain_ns.update(copy.deepcopy(dict(extra_ns))) + exec(plain_code, plain_ns) + plain_func = cast(Callable[[], None], plain_ns["check"]) + plain_raised = False + try: + plain_func() + except AssertionError: + plain_raised = True + + # Run with rewriting — fresh deepcopy so mutations from first run don't leak + mod = _rewrite_source(src) + rewritten_code = compile(mod, "", "exec") + rewritten_ns: dict[str, object] = {} + if extra_ns is not None: + rewritten_ns.update(copy.deepcopy(dict(extra_ns))) + exec(rewritten_code, rewritten_ns) + rewritten_func = cast(Callable[[], None], rewritten_ns["check"]) + rewritten_raised = False + try: + rewritten_func() + except AssertionError: + rewritten_raised = True + + assert plain_raised == rewritten_raised, ( + f"Semantic mismatch: plain {'raised' if plain_raised else 'passed'}, " + f"rewritten {'raised' if rewritten_raised else 'passed'}" + ) + + +def _run_both(src: str, extra_ns: Mapping[str, object] | None) -> list[object]: + """Execute ``check()`` plain and rewritten, returning (raised, result) pairs.""" + src = textwrap.dedent(src) + outcomes: list[object] = [] + for code in ( + compile(src, "", "exec"), + compile(_rewrite_source(src), "", "exec"), + ): + ns: dict[str, object] = {} + if extra_ns is not None: + ns.update(copy.deepcopy(dict(extra_ns))) + exec(code, ns) + func = cast(Callable[[], object], ns["check"]) + try: + outcomes.append((False, func())) + except AssertionError: + outcomes.append((True, None)) + return outcomes + + +def assert_evaluation_order( + src: str, + *, + extra_ns: Mapping[str, object] | None = None, +) -> None: + """Verify rewriting preserves the values Python's evaluation order produces. + + ``check()`` should return whatever the order is observable through -- the + operand values it saw, a trace list it appended to, the final binding of a + name a walrus operator rebinds. Both runs must agree on that return value + and on whether ``AssertionError`` was raised. + + This is stricter than :func:`assert_semantically_equivalent`, which compares + pass/fail only. An operand read after a later walrus operator rebound its + name can still fail the assertion, just having compared the wrong value, and + it is evaluated exactly once either way -- so neither of the other axes sees + it. + + Note that the observation cannot wrap the fragile operand itself: a bare + name is exactly the case the rewriter leaves unhoisted, and putting a call + around it would hoist it and hide the bug. Observe through the result + instead. + + Parameters + ---------- + src : str + Source containing a ``check()`` function that returns its observations. + extra_ns : Mapping[str, object] | None + Additional namespace entries available during execution. + """ + plain, rewritten = _run_both(src, extra_ns) + assert plain == rewritten, ( + f"Evaluation order mismatch:\n plain (raised, result) = {plain}\n" + f" rewritten (raised, result) = {rewritten}" + ) + + +# --------------------------------------------------------------------------- +# Smoke tests for the helpers themselves +# --------------------------------------------------------------------------- + + +class TestHelpersSmokeTest: + """Verify the test helpers work correctly.""" + + def test_get_failure_message_returns_message(self) -> None: + msg = get_failure_message(""" +def check(): + assert 1 == 2 +""") + assert "assert 1 == 2" in msg + + def test_get_failure_message_fails_on_passing_assert(self) -> None: + with pytest.raises(pytest.fail.Exception, match="did not raise"): + get_failure_message(""" +def check(): + assert 1 == 1 +""") + + def test_assert_introspects_succeeds(self) -> None: + assert_introspects( + """ +def check(): + x = 3 + assert x == 5 +""", + must_contain=["assert 3 == 5"], + ) + + def test_assert_introspects_fails_on_missing(self) -> None: + with pytest.raises(AssertionError, match=r"Expected.*in failure"): + assert_introspects( + """ +def check(): + assert 1 == 2 +""", + must_contain=["this is not in the message"], + ) + + def test_assert_single_evaluation(self) -> None: + assert_single_evaluation(""" +def check(): + def inc(): + counter[0] += 1 + return False + assert inc() +""") + + def test_assert_passes_when_true(self) -> None: + assert_passes_when_true(""" +def check(): + assert 1 == 1 +""") + + def test_assert_semantically_equivalent_passing(self) -> None: + assert_semantically_equivalent(""" +def check(): + assert 1 == 1 +""") + + def test_assert_semantically_equivalent_failing(self) -> None: + assert_semantically_equivalent(""" +def check(): + assert 1 == 2 +""") + + def test_assert_semantically_equivalent_detects_mismatch(self) -> None: + # This would only trigger on a bug in the rewriter itself; + # for now just verify both paths execute without error. + assert_semantically_equivalent(""" +def check(): + x = [1, 2, 3] + assert len(x) == 3 +""") + + def test_assert_evaluation_order_passing(self) -> None: + assert_evaluation_order(""" +def check(): + value = 1 + assert value == 1 + return value +""") + + def test_assert_evaluation_order_detects_value_mismatch(self) -> None: + """A divergence both runs agree to pass on must still be caught. + + The rewritten function keeps its ``@py_assert`` temporaries in locals, + so this diverges in the return value while both runs pass. + """ + with pytest.raises(AssertionError, match="Evaluation order mismatch"): + assert_evaluation_order(""" +def check(): + assert 1 == 1 + return sorted(n for n in locals() if n.startswith("@py")) +""") + + +# --------------------------------------------------------------------------- +# Introspection matrix: verify what information each expression type exposes +# --------------------------------------------------------------------------- + + +class TestIntrospectionCompare: + """Comparisons (==, !=, <, >, <=, >=, in, not in, is, is not).""" + + def test_simple_equality(self) -> None: + assert_introspects( + """ +def check(): + x = 3 + assert x == 5 +""", + must_contain=["assert 3 == 5"], + ) + + def test_chained_compare(self) -> None: + # Chained compares only show the failing pair + assert_introspects( + """ +def check(): + x = 10 + assert 1 < x < 5 +""", + must_contain=["assert 10 < 5"], + ) + + def test_in_operator(self) -> None: + assert_introspects( + """ +def check(): + x = 4 + assert x in [1, 2, 3] +""", + must_contain=["assert 4 in [1, 2, 3]"], + ) + + def test_not_in_operator(self) -> None: + assert_introspects( + """ +def check(): + x = 2 + assert x not in [1, 2, 3] +""", + must_contain=["assert 2 not in [1, 2, 3]"], + ) + + def test_is_operator(self) -> None: + assert_introspects( + """ +def check(): + x = [] + y = [] + assert x is y +""", + must_contain=["assert [] is []"], + ) + + +class TestIntrospectionBoolOp: + """Boolean operations (and, or) with short-circuit.""" + + def test_and_both_shown(self) -> None: + assert_introspects( + """ +def check(): + a = True + b = False + assert a and b +""", + must_contain=["(True and False)"], + ) + + def test_or_both_shown(self) -> None: + assert_introspects( + """ +def check(): + a = False + b = False + assert a or b +""", + must_contain=["(False or False)"], + ) + + def test_and_short_circuit(self) -> None: + assert_introspects( + """ +def check(): + a = False + assert a and explode +""", + must_contain=["False"], + ) + + +class TestIntrospectionUnaryOp: + """Unary operations (not, ~, -, +).""" + + def test_not(self) -> None: + assert_introspects( + """ +def check(): + x = True + assert not x +""", + must_contain=["assert not True"], + ) + + def test_invert(self) -> None: + # ~(-1) == 0, which is falsy + assert_introspects( + """ +def check(): + x = -1 + assert ~x +""", + must_contain=["assert ~-1"], + ) + + +class TestIntrospectionBinOp: + """Binary operations (+, -, *, /, etc.).""" + + def test_addition(self) -> None: + assert_introspects( + """ +def check(): + x = 3 + y = 4 + assert x + y == 10 +""", + must_contain=["(3 + 4)"], + ) + + def test_subtraction(self) -> None: + assert_introspects( + """ +def check(): + x = 3 + y = 4 + assert x - y == 10 +""", + must_contain=["(3 - 4)"], + ) + + +class TestIntrospectionCall: + """Function/method calls.""" + + def test_simple_call_shows_result(self) -> None: + # Currently local functions show full repr in the "where" line + assert_introspects( + """ +def check(): + def f(): + return 42 + assert f() == 100 +""", + must_contain=["where 42 = ", "()"], + ) + + def test_call_with_args_shows_result(self) -> None: + assert_introspects( + """ +def check(): + def f(x): + return x * 2 + assert f(3) == 10 +""", + must_contain=["where 6 = ", "(3)"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-callable-variable: a called name shows its repr", + ) + def test_simple_call_clean_name(self) -> None: + """Ideally the message should show 'f()' not '()'.""" + assert_introspects( + """ +def check(): + def f(): + return 42 + assert f() == 100 +""", + must_contain=["where 42 = f()"], + must_not_contain=[" None: + assert_introspects( + """ +def check(): + class Obj: + def method(self): + return 42 + obj = Obj() + assert obj.method() == 100 +""", + must_contain=["42", "100"], + ) + + +class TestIntrospectionAttribute: + """Attribute access.""" + + def test_attribute_access(self) -> None: + assert_introspects( + """ +def check(): + class Obj: + x = 3 + def __repr__(self): + return "Obj()" + obj = Obj() + assert obj.x == 5 +""", + must_contain=["where 3 = Obj().x"], + ) + + +class TestIntrospectionName: + """Variable name display.""" + + def test_local_variable_shown(self) -> None: + assert_introspects( + """ +def check(): + result = 42 + assert result == 100 +""", + must_contain=["assert 42 == 100"], + ) + + +class TestIntrospectionSubscript: + """Subscript / indexing — currently hits generic_visit.""" + + @pytest.mark.xfail( + strict=True, + reason="introspect-subscript: container[key] is not decomposed", + ) + def test_dict_subscript_shows_key_and_container(self) -> None: + assert_introspects( + """ +def check(): + d = {"a": 1, "b": 2} + assert d["a"] == 99 +""", + must_contain=["where 1 = ", "['a']"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-subscript: container[key] is not decomposed", + ) + def test_list_subscript_shows_index_and_container(self) -> None: + assert_introspects( + """ +def check(): + items = [10, 20, 30] + assert items[1] == 99 +""", + must_contain=["where 20 = ", "[1]"], + ) + + def test_subscript_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + d = {"key": "value"} + assert d["key"] == "wrong" +""") + + def test_subscript_in_compare_shows_value(self) -> None: + """Even without decomposition, the value is shown in comparisons.""" + assert_introspects( + """ +def check(): + d = {"a": 1} + assert d["a"] == 99 +""", + must_contain=["assert 1 == 99"], + ) + + +class TestIntrospectionIfExp: + """Ternary / if-expression — currently hits generic_visit.""" + + @pytest.mark.xfail( + strict=True, + reason="introspect-ifexp: ternaries are not decomposed", + ) + def test_ifexp_shows_condition_value(self) -> None: + assert_introspects( + """ +def check(): + flag = True + assert (0 if flag else 1) == 1 +""", + must_contain=["if True else"], + ) + + def test_ifexp_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + flag = True + assert (0 if flag else 1) == 1 +""") + + @pytest.mark.xfail( + strict=True, + reason="introspect-ifexp: ternaries are not decomposed", + ) + def test_ifexp_in_compare_shows_result(self) -> None: + assert_introspects( + """ +def check(): + flag = True + assert (0 if flag else 1) == 99 +""", + must_contain=["assert 0 == 99", "if True else"], + ) + + def test_ifexp_short_circuit_true(self) -> None: + """Orelse branch must NOT be evaluated when condition is True.""" + assert_passes_when_true(""" +def check(): + flag = True + assert (1 if flag else (1/0)) == 1 +""") + + def test_ifexp_short_circuit_false(self) -> None: + """Body branch must NOT be evaluated when condition is False.""" + assert_passes_when_true(""" +def check(): + flag = False + assert (1/0 if flag else 1) == 1 +""") + + +class TestIntrospectionContainerLiteral: + """Container literals ([...], {...}, {k:v}) — currently hits generic_visit.""" + + @pytest.mark.xfail( + strict=True, + reason="introspect-container-literal: list/dict/set literals are not decomposed", + ) + def test_list_literal_shows_elements(self) -> None: + assert_introspects( + """ +def check(): + def f(): + return 99 + assert [f(), 2, 3] == [1, 2, 3] +""", + must_contain=["where 99 = f()"], + ) + + def test_list_literal_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + assert [1, 2, 3] == [1, 2, 4] +""") + + def test_dict_literal_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + assert {"a": 1} == {"a": 2} +""") + + +class TestIntrospectionComprehension: + """Comprehensions — currently hits generic_visit.""" + + def test_listcomp_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + assert [x * 2 for x in range(3)] == [0, 2, 5] +""") + + def test_listcomp_in_compare_shows_result(self) -> None: + assert_introspects( + """ +def check(): + assert [x * 2 for x in range(3)] == [0, 2, 5] +""", + must_contain=["[0, 2, 4]"], + ) + + +class TestIntrospectionFString: + """F-string expressions — currently hits generic_visit.""" + + def test_fstring_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + x = 42 + assert f"value={x}" == "value=99" +""") + + def test_fstring_in_compare_shows_result(self) -> None: + assert_introspects( + """ +def check(): + x = 42 + assert f"value={x}" == "value=99" +""", + must_contain=["value=42"], + ) + + +class TestIntrospectionMethodCall: + """Method calls — currently show the bound method as its own "where" line.""" + + @pytest.mark.xfail( + strict=True, + reason="introspect-method-call-flat: the bound method is shown separately", + ) + def test_method_call_flat_format(self) -> None: + """Method calls show 'where result = obj.method()' in one line.""" + assert_introspects( + """ +def check(): + class Obj: + def compute(self): + return 42 + def __repr__(self): + return "Obj()" + obj = Obj() + assert obj.compute() == 100 +""", + must_contain=["where 42 = Obj().compute()"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-method-call-flat: the bound method is shown separately", + ) + def test_method_call_no_bound_method_noise(self) -> None: + """No separate 'where compute = obj.compute' line.""" + msg = get_failure_message(""" +def check(): + class Obj: + def compute(self): + return 42 + def __repr__(self): + return "Obj()" + obj = Obj() + assert obj.compute() == 100 +""") + lines = msg.splitlines() + for line in lines: + assert "where compute = " not in line, ( + f"Noisy bound-method intermediate found:\n{msg}" + ) + + def test_callable_variable_shows_result(self) -> None: + # Current behavior: shows full function repr, not variable name + assert_introspects( + """ +def check(): + def factory(): + return 42 + fn = factory + assert fn() == 100 +""", + must_contain=["where 42 = ", "()"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-callable-variable: a called name shows its repr", + ) + def test_callable_variable_clean_name(self) -> None: + """Ideally should show 'fn()' not '()'.""" + assert_introspects( + """ +def check(): + def factory(): + return 42 + fn = factory + assert fn() == 100 +""", + must_contain=["where 42 = fn()"], + must_not_contain=[" None: + assert_introspects( + """ +def check(): + x = 10 + assert (y := x * 2) == 100 +""", + must_contain=["assert 20 == 100"], + ) + + def test_walrus_semantics_preserved(self) -> None: + assert_semantically_equivalent(""" +def check(): + x = 10 + assert (y := x * 2) == 100 +""") + + +# --------------------------------------------------------------------------- +# Single-evaluation tests: ensure no expression is evaluated multiple times +# --------------------------------------------------------------------------- + + +class TestSingleEvaluation: + """Verify the rewriter doesn't cause double-evaluation of side effects. + + Each test uses a counter to track how many times a side-effecting + expression is evaluated. The rewritten assert should evaluate each + expression exactly once, regardless of whether the assertion passes or fails. + """ + + def test_call_in_compare_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return 42 + assert side_effect() == 100 +""") + + def test_call_in_boolean_and_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return True + assert side_effect() and False +""") + + def test_call_in_boolean_or_short_circuit(self) -> None: + # With `or`, if first is truthy, second is NOT evaluated + assert_single_evaluation( + """ +def check(): + def first(): + counter[0] += 1 + return False + def second(): + counter[0] += 1 + return False + assert first() or second() +""", + expected_call_count=2, + ) + + def test_call_in_unary_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return True + assert not side_effect() +""") + + def test_call_in_binop_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return 5 + assert side_effect() + 1 == 100 +""") + + def test_attribute_access_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + class Obj: + @property + def prop(self): + counter[0] += 1 + return 42 + obj = Obj() + assert obj.prop == 100 +""") + + def test_subscript_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + class CountingDict(dict): + def __getitem__(self, key): + counter[0] += 1 + return super().__getitem__(key) + d = CountingDict(a=1) + assert d["a"] == 100 +""") + + @pytest.mark.xfail( + strict=True, + reason="single-eval-walrus: the walrus expression runs twice", + ) + def test_walrus_in_compare_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return 42 + assert (x := side_effect()) == 100 +""") + + @pytest.mark.xfail( + strict=True, + reason="single-eval-walrus: the walrus expression runs twice", + ) + def test_walrus_in_boolean_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return 42 + assert (x := side_effect()) and False +""") + + @pytest.mark.xfail( + strict=True, + reason="single-eval-walrus: the walrus expression runs twice", + ) + def test_walrus_in_chained_compare_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def side_effect(): + counter[0] += 1 + return 5 + assert 1 < (x := side_effect()) < 3 +""") + + def test_method_call_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + class Obj: + def compute(self): + counter[0] += 1 + return 42 + obj = Obj() + assert obj.compute() == 100 +""") + + def test_nested_calls_each_evaluated_once(self) -> None: + assert_single_evaluation( + """ +def check(): + def outer(x): + counter[0] += 1 + return x + 1 + def inner(): + counter[0] += 1 + return 5 + assert outer(inner()) == 100 +""", + expected_call_count=2, + ) + + def test_multiple_comparators_evaluated_once_each(self) -> None: + assert_single_evaluation( + """ +def check(): + def make_val(n): + counter[0] += 1 + return n + assert make_val(1) < make_val(5) < make_val(3) +""", + expected_call_count=3, + ) + + def test_ifexp_condition_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def cond(): + counter[0] += 1 + return True + assert (0 if cond() else 1) == 1 +""") + + def test_comprehension_generator_evaluated_once(self) -> None: + assert_single_evaluation(""" +def check(): + def items(): + counter[0] += 1 + return [1, 2, 3] + assert [x * 2 for x in items()] == [2, 4, 7] +""") + + +# --------------------------------------------------------------------------- +# Evaluation-order tests: operands must see the values Python gives them +# --------------------------------------------------------------------------- + + +class TestEvaluationOrder: + """Verify rewriting does not reorder operands against a walrus operator. + + The rewriter turns sub-expressions into statements that run in source + order, but an operand it leaves unhoisted is read only when the enclosing + expression is assembled -- after the statements belonging to the operands + that follow it. A walrus operator in one of those rebinds the name in + between, and the earlier operand then sees a value Python would never have + given it. + + Most operands are safe because ``generic_visit`` hoists them into a + temporary. The cases that pass here are guards: they keep that true as new + visitors take expression types away from ``generic_visit``. + """ + + @pytest.mark.xfail( + strict=True, + reason="order-compare-left: left operand is read after a walrus in the right one", + ) + def test_compare_left_operand_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + value = "Hello" + try: + assert value != identity(value := value.lower()) + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-compare-left: a failing comparison reports the post-walrus value", + ) + def test_compare_reports_left_operand(self) -> None: + assert_introspects( + """ +def check(): + def identity(v): + return v + value = 2 + assert value == identity(value := 3) +""", + must_contain=["assert 2 == 3"], + ) + + @pytest.mark.xfail( + strict=True, + reason="order-call-argument: an earlier argument is read after a later walrus", + ) + def test_call_earlier_argument_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + def collect(*values): + return values + value = "Hello" + try: + assert collect(value, identity(value := value.lower())) == ("Hello", "hello") + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-binop-left: left operand is read after a walrus in the right one", + ) + def test_binop_left_operand_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + value = 1 + try: + assert value + identity(value := 5) == 6 + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-starred-argument: a starred argument is read after a later walrus", + ) + def test_starred_argument_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + def collect(*values): + return values + items = [1] + try: + assert collect(*items, identity(items := [9])) == (1, [9]) + except AssertionError: + return "raised", items + return "passed", items +""") + + @pytest.mark.xfail( + strict=True, + reason="order-compare-left: a chained comparison re-reads its left operand", + ) + def test_chained_compare_operands_in_order(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + value = 1 + try: + assert value < identity(value := 5) < 9 + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-call-argument: a walrus argument is substituted into a later one", + ) + def test_bare_walrus_argument_in_order(self) -> None: + """A walrus argument is evaluated in place, before the ones after it.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + def collect(*values): + return values + try: + assert collect((x := 1), identity(x := 2)) == (1, 2) + except AssertionError: + return "raised", x + return "passed", x +""") + + def test_container_literal_operand_in_order(self) -> None: + """Guard: ``generic_visit`` hoists container literals into a temporary.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + value = 1 + try: + assert [value] == identity([value := 2]) + except AssertionError: + return "raised", value + return "passed", value +""") + + def test_fstring_operand_in_order(self) -> None: + """Guard: ``generic_visit`` hoists f-strings into a temporary.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + value = "a" + try: + assert f"{value}" != identity(str(value := "b")) + except AssertionError: + return "raised", value + return "passed", value +""") + + def test_attribute_operand_in_order(self) -> None: + """Guard: ``visit_Attribute`` hoists the attribute into a temporary.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + class Box: + def __init__(self, value): + self.value = value + box = Box(1) + try: + assert box.value == identity((box := Box(2)).value) + except AssertionError: + return "raised", box.value + return "passed", box.value +""") + + def test_subscript_container_in_order(self) -> None: + """Guard: the container is read before a walrus in the key rebinds it.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + first = {"k": 1} + second = {"k": 2} + box = first + try: + assert box[identity((box := second) and "k")] == 1 + except AssertionError: + return "raised", box is second + return "passed", box is second +""") + + def test_method_receiver_in_order(self) -> None: + """Guard: the receiver is read before a walrus in an argument rebinds it.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + class Box: + def take(self, value): + return value + obj = Box() + try: + assert obj.take(identity(obj := None)) is None + except AssertionError: + return "raised", obj + return "passed", obj +""") + + @pytest.mark.xfail( + strict=True, + reason="order-call-argument: a keyword argument is read after a later walrus", + ) + def test_keyword_argument_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + def collect(**kwargs): + return kwargs + value = 1 + try: + assert collect(a=value, b=identity(value := 2)) == {"a": 1, "b": 2} + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-call-argument: a ** argument is read after a later walrus", + ) + def test_double_star_argument_precedes_walrus(self) -> None: + assert_evaluation_order(""" +def check(): + def identity(v): + return v + def collect(**kwargs): + return kwargs + mapping = {"a": 1} + try: + assert collect(**mapping, b=identity(mapping := {"a": 9})) == {"a": 1, "b": {"a": 9}} + except AssertionError: + return "raised", mapping + return "passed", mapping +""") + + @pytest.mark.xfail( + strict=True, + reason="order-name-rebound-by-call: #14820, a call rebinding a global is not seen", + ) + def test_global_rebound_by_call_precedes_compare(self) -> None: + assert_evaluation_order(""" +count = 0 + +def bump(): + global count + count = 99 + return 0 + +def check(): + try: + assert count == bump() + except AssertionError: + return "raised", count + return "passed", count +""") + + @pytest.mark.xfail( + strict=True, + reason="order-name-rebound-by-call: #14820, a call rebinding a nonlocal is not seen", + ) + def test_nonlocal_rebound_by_call_precedes_compare(self) -> None: + assert_evaluation_order(""" +def check(): + value = 1 + def bump(): + nonlocal value + value = 99 + return 1 + try: + assert value == bump() + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-name-rebound-by-call: #14820, an earlier argument is read after a rebinding call", + ) + def test_global_rebound_by_call_precedes_argument(self) -> None: + assert_evaluation_order(""" +value = "a" + +def bump(): + global value + value = "b" + return "x" + +def collect(*args): + return args + +def check(): + try: + assert collect(value, bump()) == ("a", "x") + except AssertionError: + return "raised", value + return "passed", value +""") + + @pytest.mark.xfail( + strict=True, + reason="order-chained-compare-lazy: #14819, every comparator is evaluated", + ) + def test_chained_compare_stops_at_the_first_false(self) -> None: + assert_evaluation_order(""" +def check(): + trace = [] + def rec(label, value): + trace.append(label) + return value + try: + assert rec("a", 1) < rec("b", 0) < rec("c", 5) + except AssertionError: + return "raised", trace + return "passed", trace +""") + + @pytest.mark.xfail( + strict=True, + reason="order-chained-compare-lazy: #14819, an unreached comparator still raises", + ) + def test_chained_compare_unreached_operand_does_not_raise(self) -> None: + assert_evaluation_order(""" +def check(): + try: + assert 1 < 0 < 1 / 0 + except AssertionError: + return "raised", None + return "passed", None +""") + + def test_ifexp_branches_in_order(self) -> None: + """Guard: the condition is evaluated before the selected branch.""" + assert_evaluation_order(""" +def check(): + def identity(v): + return v + flag = True + try: + assert (1 if flag else 2) == identity(1 if (flag := False) or True else 3) + except AssertionError: + return "raised", flag + return "passed", flag +""") + + +# --------------------------------------------------------------------------- +# Edge cases: combinations of new visitors with existing ones +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Regression and edge-case tests combining multiple expression types.""" + + @pytest.mark.xfail( + strict=True, + reason="introspect-subscript: container[key] is not decomposed", + ) + def test_subscript_with_variable_key(self) -> None: + """Subscript where the key is a variable (not constant).""" + assert_introspects( + """ +def check(): + d = {"hello": 42} + key = "hello" + assert d[key] == 100 +""", + must_contain=["where 42 = ", "['hello']"], + ) + + def test_subscript_with_call_key(self) -> None: + """Subscript where the key is a function call.""" + assert_introspects( + """ +def check(): + d = {0: "zero", 1: "one"} + def get_key(): + return 0 + assert d[get_key()] == "wrong" +""", + must_contain=["'zero'", "'wrong'"], + ) + + def test_nested_subscript(self) -> None: + """Nested subscript: d[k1][k2].""" + assert_introspects( + """ +def check(): + d = {"a": {"b": 42}} + assert d["a"]["b"] == 100 +""", + must_contain=["42", "100"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-method-call-flat: the bound method is shown separately", + ) + def test_method_call_with_args(self) -> None: + """Method call with arguments shows flat format.""" + assert_introspects( + """ +def check(): + class Calculator: + def add(self, a, b): + return a + b + def __repr__(self): + return "Calc()" + c = Calculator() + assert c.add(2, 3) == 10 +""", + must_contain=["where 5 = Calc().add(2, 3)"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-method-call-flat: the bound method is shown separately", + ) + def test_chained_method_calls(self) -> None: + """Chained method call: obj.method1().method2().""" + assert_introspects( + """ +def check(): + class Builder: + def __init__(self, val=0): + self.val = val + def add(self, n): + return Builder(self.val + n) + def result(self): + return self.val + def __repr__(self): + return f"Builder({self.val})" + b = Builder() + assert b.add(5).result() == 100 +""", + must_contain=["where 5 = ", ".result()"], + ) + + def test_subscript_on_method_result(self) -> None: + """Subscript on method return value: obj.method()[key].""" + assert_introspects( + """ +def check(): + class Store: + def get_data(self): + return {"x": 42} + def __repr__(self): + return "Store()" + s = Store() + assert s.get_data()["x"] == 100 +""", + must_contain=["42", "100"], + ) + + @pytest.mark.xfail( + strict=True, + reason="introspect-ifexp: ternaries are not decomposed", + ) + def test_ifexp_with_call_condition(self) -> None: + """IfExp where condition is a function call.""" + assert_introspects( + """ +def check(): + def is_ready(): + return False + assert (1 if is_ready() else 0) == 1 +""", + must_contain=["if False else"], + ) + + def test_walrus_in_subscript(self) -> None: + """Walrus operator used as subscript key.""" + assert_semantically_equivalent(""" +def check(): + d = {1: "one", 2: "two"} + x = 1 + assert d[(y := x + 1)] == "wrong" +""") + + def test_method_call_single_evaluation(self) -> None: + """Method with side effects is only called once.""" + assert_single_evaluation(""" +def check(): + class Obj: + def compute(self): + counter[0] += 1 + return 42 + obj = Obj() + assert obj.compute() == 100 +""") + + def test_subscript_single_evaluation(self) -> None: + """Custom __getitem__ with side effects is only called once.""" + assert_single_evaluation(""" +def check(): + class CountingList: + def __init__(self, items): + self.items = items + def __getitem__(self, idx): + counter[0] += 1 + return self.items[idx] + def __repr__(self): + return repr(self.items) + lst = CountingList([10, 20, 30]) + assert lst[1] == 99 +""") + + def test_ifexp_condition_single_evaluation(self) -> None: + """IfExp condition with side effects is only evaluated once.""" + assert_single_evaluation(""" +def check(): + def check_flag(): + counter[0] += 1 + return True + assert (0 if check_flag() else 1) == 99 +""") + + def test_complex_assertion_semantics(self) -> None: + """Complex assertion combining multiple new visitors.""" + assert_semantically_equivalent(""" +def check(): + class Config: + def __init__(self): + self.data = {"timeout": 30} + def get(self, key): + return self.data[key] + cfg = Config() + flag = True + assert (cfg.get("timeout") if flag else 0) > 60 +""") + + def test_assert_with_message_still_works(self) -> None: + """Assert with a custom message still works with new visitors.""" + msg = get_failure_message(""" +def check(): + d = {"key": 42} + assert d["key"] == 100, "custom failure message" +""") + assert "custom failure message" in msg + + @pytest.mark.xfail( + strict=True, + reason="introspect-method-call-flat: the bound method is shown separately", + ) + def test_method_call_on_global(self) -> None: + """Method call on a global/module-level object.""" + assert_introspects( + """ +items = [1, 2, 3] +def check(): + assert items.count(99) == 1 +""", + must_contain=["where 0 = [1, 2, 3].count(99)"], + ) From c7b890858263ca7aaaac01f89a5464d475b355d9 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 8 May 2026 10:37:47 +0200 Subject: [PATCH 02/13] fix(rewrite): prevent walrus operator double evaluation in assertions Fixes #14445 - assertion rewriting evaluated NamedExpr (:=) expressions multiple times, causing side effects to fire repeatedly. The root cause was the `variables_overwrite` mechanism which stored and re-evaluated NamedExpr AST nodes in subsequent assertions, in `_call_reprcompare`'s results tuple, and in explanation formatting. The fix: - visit_NamedExpr: reference the target variable in explanations instead of re-evaluating the full expression - visit_Compare: assign left-side NamedExpr to a temp before right-side hoisting; freeze left_res when a comparator walrus targets the same name; replace NamedExpr entries in `results` with target variables - visit_BoolOp: capture short-circuit condition in a stable temp for the explanation path; remove walrus target rename logic - visit_Call: remove variables_overwrite substitution (walrus now properly assigns to user variables in its natural evaluation position) - Remove variables_overwrite, scope tracking, Sentinel class Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- src/_pytest/assertion/rewrite.py | 102 +++++++++++-------------------- testing/test_assertrewrite.py | 68 ++++++++++++++++++++- 2 files changed, 102 insertions(+), 68 deletions(-) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index dcc37df2c98..85ac4e34463 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -3,7 +3,6 @@ from __future__ import annotations import ast -from collections import defaultdict from collections.abc import Callable from collections.abc import Iterable from collections.abc import Iterator @@ -58,10 +57,6 @@ from _pytest.assertion import AssertionState -class Sentinel: - pass - - assertstate_key = StashKey["AssertionState"]() # pytest caches rewritten pycs in pycache dirs @@ -69,9 +64,6 @@ class Sentinel: PYC_EXT = ".py" + ((__debug__ and "c") or "o") PYC_TAIL = "." + PYTEST_TAG + PYC_EXT -# Special marker that denotes we have just left a scope definition -_SCOPE_END_MARKER = Sentinel() - class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader): """PEP302/PEP451 import hook which rewrites asserts.""" @@ -652,14 +644,8 @@ class AssertionRewriter(ast.NodeVisitor): .push_format_context() and .pop_format_context() which allows to build another %-formatted string while already building one. - :scope: A tuple containing the current scope used for variables_overwrite. - - :variables_overwrite: A dict filled with references to variables - that change value within an assert. This happens when a variable is - reassigned with the walrus operator - - This state, except the variables_overwrite, is reset on every new assert - statement visited and used by the other visitors. + This state is reset on every new assert statement visited and used by + the other visitors. """ def __init__( @@ -675,10 +661,6 @@ def __init__( else: self.enable_assertion_pass_hook = False self.source = source - self.scope: tuple[ast.AST, ...] = () - self.variables_overwrite: defaultdict[tuple[ast.AST, ...], dict[str, str]] = ( - defaultdict(dict) - ) def run(self, mod: ast.Module) -> None: """Find all assert statements in *mod* and rewrite them.""" @@ -728,16 +710,9 @@ def run(self, mod: ast.Module) -> None: mod.body[pos:pos] = imports # Collect asserts. - self.scope = (mod,) - nodes: list[ast.AST | Sentinel] = [mod] + nodes: list[ast.AST] = [mod] while nodes: node = nodes.pop() - if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): - self.scope = tuple((*self.scope, node)) - nodes.append(_SCOPE_END_MARKER) - if node == _SCOPE_END_MARKER: - self.scope = self.scope[:-1] - continue assert isinstance(node, ast.AST) for name, field in ast.iter_fields(node): if isinstance(field, list): @@ -964,15 +939,17 @@ def visit_Assert(self, assert_: ast.Assert) -> list[ast.stmt]: return self.statements def visit_NamedExpr(self, name: ast.NamedExpr) -> tuple[ast.NamedExpr, str]: - # This method handles the 'walrus operator' repr of the target - # name if it's a local variable or _should_repr_global_name() - # thinks it's acceptable. + # Return the NamedExpr as-is so it evaluates in its natural position + # (preserving left-to-right evaluation order). For the explanation, + # reference the target variable (already assigned by the walrus) to + # avoid re-evaluating the expression. locs = ast.Call(self.builtin("locals"), [], []) target_id = name.target.id + target_name = ast.Name(target_id, ast.Load()) inlocs = ast.Compare(ast.Constant(target_id), [ast.In()], [locs]) - dorepr = self.helper("_should_repr_global_name", name) + dorepr = self.helper("_should_repr_global_name", target_name) test = ast.BoolOp(ast.Or(), [inlocs, dorepr]) - expr = ast.IfExp(test, self.display(name), ast.Constant(target_id)) + expr = ast.IfExp(test, self.display(target_name), ast.Constant(target_id)) return name, self.explanation_param(expr) def visit_Name(self, name: ast.Name) -> tuple[ast.Name, str]: @@ -998,20 +975,9 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: for i, v in enumerate(boolop.values): if i: fail_inner: list[ast.stmt] = [] - # cond is set in a prior loop iteration below - self.expl_stmts.append(ast.If(cond, fail_inner, [])) # noqa: F821 + # expl_cond is set in a prior loop iteration below + self.expl_stmts.append(ast.If(expl_cond, fail_inner, [])) # noqa: F821 self.expl_stmts = fail_inner - match v: - # Check if the left operand is an ast.NamedExpr and the value has already been visited - case ast.Compare( - left=ast.NamedExpr(target=ast.Name(id=target_id)) - ) if target_id in [ - e.id for e in boolop.values[:i] if hasattr(e, "id") - ]: - pytest_temp = self.variable() - self.variables_overwrite[self.scope][target_id] = v.left # type:ignore[assignment] - # mypy's false positive, we're checking that the 'target' attribute exists. - v.left.target.id = pytest_temp # type:ignore[attr-defined] self.push_format_context() res, expl = self.visit(v) body.append(ast.Assign([ast.Name(res_var, ast.Store())], res)) @@ -1022,8 +988,16 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: cond: ast.expr = res if is_or: cond = ast.UnaryOp(ast.Not(), cond) + # Capture the condition in a temp variable so the explanation + # path (which runs after walrus operators may have modified + # the original variable) sees the correct truthiness. + cond_var = self.variable() + body.append(ast.Assign([ast.Name(cond_var, ast.Store())], cond)) + expl_cond: ast.expr = ast.Name(cond_var, ast.Load()) # noqa: F841 inner: list[ast.stmt] = [] - self.statements.append(ast.If(cond, inner, [])) + self.statements.append( + ast.If(ast.Name(cond_var, ast.Load()), inner, []) + ) self.statements = body = inner self.statements = save self.expl_stmts = fail_save @@ -1053,19 +1027,10 @@ def visit_Call(self, call: ast.Call) -> tuple[ast.Name, str]: new_args = [] new_kwargs = [] for arg in call.args: - if isinstance(arg, ast.Name) and arg.id in self.variables_overwrite.get( - self.scope, {} - ): - arg = self.variables_overwrite[self.scope][arg.id] # type:ignore[assignment] res, expl = self.visit(arg) arg_expls.append(expl) new_args.append(res) for keyword in call.keywords: - match keyword.value: - case ast.Name(id=id) if id in self.variables_overwrite.get( - self.scope, {} - ): - keyword.value = self.variables_overwrite[self.scope][id] # type:ignore[assignment] res, expl = self.visit(keyword.value) new_kwargs.append(ast.keyword(keyword.arg, res)) if keyword.arg: @@ -1100,17 +1065,13 @@ def visit_Attribute(self, attr: ast.Attribute) -> tuple[ast.Name, str]: def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: self.push_format_context() - # We first check if we have overwritten a variable in the previous assert - match comp.left: - case ast.Name(id=name_id) if name_id in self.variables_overwrite.get( - self.scope, {} - ): - comp.left = self.variables_overwrite[self.scope][name_id] # type: ignore[assignment] - case ast.NamedExpr(target=ast.Name(id=target_id)): - self.variables_overwrite[self.scope][target_id] = comp.left # type: ignore[assignment] left_res, left_expl = self.visit(comp.left) if isinstance(comp.left, ast.Compare | ast.BoolOp): left_expl = f"({left_expl})" + # If the left operand is a NamedExpr, assign it to a temp so the + # walrus executes before any right-side expressions are hoisted. + if isinstance(left_res, ast.NamedExpr): + left_res = self.assign(left_res) res_variables = [self.variable() for i in range(len(comp.ops))] load_names: list[ast.expr] = [ast.Name(v, ast.Load()) for v in res_variables] store_names = [ast.Name(v, ast.Store()) for v in res_variables] @@ -1119,13 +1080,16 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: syms: list[ast.expr] = [] results = [left_res] for i, op, next_operand in it: + # If the next operand is a walrus that assigns to the same name as + # the current left_res, we must freeze left_res's value before the + # walrus modifies it. match (next_operand, left_res): case ( ast.NamedExpr(target=ast.Name(id=target_id)), ast.Name(id=name_id), ) if target_id == name_id: - next_operand.target.id = self.variable() - self.variables_overwrite[self.scope][name_id] = next_operand # type: ignore[assignment] + left_res = self.assign(left_res) + results[-1] = left_res next_res, next_expl = self.visit(next_operand) if isinstance(next_operand, ast.Compare | ast.BoolOp): @@ -1138,6 +1102,12 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: res_expr = ast.copy_location(ast.Compare(left_res, [op], [next_res]), comp) self.statements.append(ast.Assign([store_names[i]], res_expr)) left_res, left_expl = next_res, next_expl + # Replace NamedExpr entries in results with their target variable + # to avoid re-evaluating walrus operators in the explanation path. + results = [ + ast.Name(r.target.id, ast.Load()) if isinstance(r, ast.NamedExpr) else r + for r in results + ] # Use pytest.assertion.util._reprcompare if that's available. expl_call = self.helper( "_call_reprcompare", diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index 9740bf3c05e..59a0de8d5d3 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -1711,7 +1711,7 @@ def test_walrus_operator_change_boolean_value(): ) result = pytester.runpytest() assert result.ret == 1 - result.stdout.fnmatch_lines(["*assert not (True and False is False)"]) + result.stdout.fnmatch_lines(["*assert not (False and False is False)"]) def test_assertion_walrus_operator_boolean_none_fails( self, pytester: Pytester @@ -1725,7 +1725,7 @@ def test_walrus_operator_change_boolean_value(): ) result = pytester.runpytest() assert result.ret == 1 - result.stdout.fnmatch_lines(["*assert not (True and None is None)"]) + result.stdout.fnmatch_lines(["*assert not (None and None is None)"]) def test_assertion_walrus_operator_value_changes_cleared_after_each_test( self, pytester: Pytester @@ -1869,6 +1869,70 @@ def test_2(): assert result.ret == 0 +class TestIssue14445: + """Regression tests for #14445: walrus operator double evaluation.""" + + def test_walrus_no_double_eval_basic(self, pytester: Pytester) -> None: + """Walrus captures the value at assignment time, not re-evaluated later.""" + pytester.makepyfile( + """ + class Counter: + def __init__(self): + self.value = 0 + def increment(self): + self.value += 1 + + def test_walrus_in_assertion_basic(): + c = Counter() + assert (before := c.value) == 0 + c.increment() + assert before != (after := c.value) + """ + ) + result = pytester.runpytest() + assert result.ret == 0 + + def test_walrus_no_double_eval_running_counter(self, pytester: Pytester) -> None: + """Walrus increments fire exactly once per assert statement.""" + pytester.makepyfile( + """ + def test_walrus_running_counter(): + count = 0 + items = [] + items.append("a") + assert (count := count + 1) == len(items) + items.append("b") + assert (count := count + 1) == len(items) + items.append("c") + assert (count := count + 1) == len(items) + assert count == 3 + """ + ) + result = pytester.runpytest() + assert result.ret == 0 + + def test_walrus_no_double_eval_in_function_call(self, pytester: Pytester) -> None: + """Walrus in function call arguments not evaluated twice.""" + pytester.makepyfile( + """ + call_count = 0 + + def side_effect(): + global call_count + call_count += 1 + return call_count + + def test_walrus_side_effect(): + assert (val := side_effect()) == 1 + assert val == 1 + assert (val := side_effect()) == 2 + assert val == 2 + """ + ) + result = pytester.runpytest() + assert result.ret == 0 + + @pytest.mark.skipif( sys.maxsize <= (2**31 - 1), reason="Causes OverflowError on 32bit systems" ) From b2dfadc0b12d44e6ad7f62e83d6ff5b7548e955e Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 8 May 2026 12:59:20 +0200 Subject: [PATCH 03/13] Add changelog fragment for #14445 Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- changelog/14445.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/14445.bugfix.rst diff --git a/changelog/14445.bugfix.rst b/changelog/14445.bugfix.rst new file mode 100644 index 00000000000..aaae0c615f5 --- /dev/null +++ b/changelog/14445.bugfix.rst @@ -0,0 +1 @@ +Fixed assertion rewriting evaluating walrus operator (``:=``) expressions multiple times, causing incorrect test results when the expression had side effects (e.g., incrementing a counter or calling a function). From 400a848ed09e46d33e056de5b9b6ed0355937c71 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 8 May 2026 13:04:00 +0200 Subject: [PATCH 04/13] test(rewrite): add xfail tests for remaining walrus edge cases Add tests for two remaining walrus double-evaluation scenarios: - Bare NamedExpr as BoolOp operand evaluated twice via condition check - Same walrus target in chained comparison evaluated multiple times Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- testing/test_assertrewrite.py | 40 +++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index 59a0de8d5d3..957aada75d0 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -1932,6 +1932,46 @@ def test_walrus_side_effect(): result = pytester.runpytest() assert result.ret == 0 + @pytest.mark.xfail(reason="BoolOp condition re-evaluates walrus operand") + def test_walrus_no_double_eval_in_boolop(self, pytester: Pytester) -> None: + """Bare walrus as a BoolOp operand must not be evaluated twice.""" + pytester.makepyfile( + """ + call_count = 0 + + def side_effect(): + global call_count + call_count += 1 + return call_count + + def test_walrus_boolop(): + assert (x := side_effect()) and x == 1 + assert call_count == 1 + """ + ) + result = pytester.runpytest() + assert result.ret == 0 + + @pytest.mark.xfail(reason="Chained compare re-evaluates walrus with same target") + def test_walrus_no_double_eval_chained_compare(self, pytester: Pytester) -> None: + """Same walrus target in chained comparison must evaluate each once.""" + pytester.makepyfile( + """ + call_count = 0 + + def track(value): + global call_count + call_count += 1 + return value + + def test_walrus_chained(): + assert (x := track(1)) < (x := track(3)) < (x := track(5)) + assert call_count == 3 + """ + ) + result = pytester.runpytest() + assert result.ret == 0 + @pytest.mark.skipif( sys.maxsize <= (2**31 - 1), reason="Causes OverflowError on 32bit systems" From f9973707b448bde7d4799d4a9919405842d79552 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 8 May 2026 13:05:41 +0200 Subject: [PATCH 05/13] fix(rewrite): avoid double evaluation of walrus in BoolOp condition Use the already-assigned res_var to build the short-circuit condition instead of the raw visitor result, preventing bare NamedExpr operands from being evaluated a second time when checking truthiness. Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- src/_pytest/assertion/rewrite.py | 9 +++++---- testing/test_assertrewrite.py | 1 - 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index 85ac4e34463..b6cde273507 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -985,12 +985,13 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: call = ast.Call(app, [expl_format], []) self.expl_stmts.append(ast.Expr(call)) if i < levels: - cond: ast.expr = res + # Use res_var (already assigned above) rather than res directly, + # so that NamedExpr operands aren't evaluated a second time. + cond: ast.expr = ast.Name(res_var, ast.Load()) if is_or: cond = ast.UnaryOp(ast.Not(), cond) - # Capture the condition in a temp variable so the explanation - # path (which runs after walrus operators may have modified - # the original variable) sees the correct truthiness. + # Capture the condition in a stable temp for the explanation + # path — res_var is overwritten by subsequent operands. cond_var = self.variable() body.append(ast.Assign([ast.Name(cond_var, ast.Store())], cond)) expl_cond: ast.expr = ast.Name(cond_var, ast.Load()) # noqa: F841 diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index 957aada75d0..c5d4f9963b3 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -1932,7 +1932,6 @@ def test_walrus_side_effect(): result = pytester.runpytest() assert result.ret == 0 - @pytest.mark.xfail(reason="BoolOp condition re-evaluates walrus operand") def test_walrus_no_double_eval_in_boolop(self, pytester: Pytester) -> None: """Bare walrus as a BoolOp operand must not be evaluated twice.""" pytester.makepyfile( From 487fe38ffbd4519c4253476fb905f9e2058b74c1 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 8 May 2026 13:08:10 +0200 Subject: [PATCH 06/13] fix(rewrite): assign walrus comparators to temps in chained comparisons In a chained comparison like `(x := f()) < (x := g()) < (x := h())`, each NamedExpr comparator is now assigned to a temp variable so it evaluates exactly once. Previously the raw NamedExpr node would be reused as left_res in the next iteration, causing double evaluation. Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- src/_pytest/assertion/rewrite.py | 11 +++++------ testing/test_assertrewrite.py | 1 - 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index b6cde273507..77bf81b8d69 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -1095,6 +1095,11 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: next_res, next_expl = self.visit(next_operand) if isinstance(next_operand, ast.Compare | ast.BoolOp): next_expl = f"({next_expl})" + # Assign NamedExpr comparators to a temp so each walrus evaluates + # exactly once — critical for chained comparisons where the same + # node would otherwise be re-evaluated as left_res next iteration. + if isinstance(next_res, ast.NamedExpr): + next_res = self.assign(next_res) results.append(next_res) sym = BINOP_MAP[op.__class__] syms.append(ast.Constant(sym)) @@ -1103,12 +1108,6 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: res_expr = ast.copy_location(ast.Compare(left_res, [op], [next_res]), comp) self.statements.append(ast.Assign([store_names[i]], res_expr)) left_res, left_expl = next_res, next_expl - # Replace NamedExpr entries in results with their target variable - # to avoid re-evaluating walrus operators in the explanation path. - results = [ - ast.Name(r.target.id, ast.Load()) if isinstance(r, ast.NamedExpr) else r - for r in results - ] # Use pytest.assertion.util._reprcompare if that's available. expl_call = self.helper( "_call_reprcompare", diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index c5d4f9963b3..6e5225de992 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -1951,7 +1951,6 @@ def test_walrus_boolop(): result = pytester.runpytest() assert result.ret == 0 - @pytest.mark.xfail(reason="Chained compare re-evaluates walrus with same target") def test_walrus_no_double_eval_chained_compare(self, pytester: Pytester) -> None: """Same walrus target in chained comparison must evaluate each once.""" pytester.makepyfile( From 1189daeb28aede958bff80eed152f27d35cc478c Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Wed, 3 Jun 2026 10:55:00 +0200 Subject: [PATCH 07/13] fix(rewrite): show correct walrus values in BoolOp explanations When multiple walrus operators target the same variable in a BoolOp (e.g., `assert (x := side_effect()) and (x := False)`), the assertion explanation previously showed the final value of `x` for all operands because the format context evaluated lazily after all operands ran. Fix by tracking Name/NamedExpr operand values in stable @py_assert variables (via self.assign) immediately after evaluation, then pointing the explanation format context at the tracked copy. This uses the same value-tracking mechanism already used by visit_Call, visit_Attribute, etc. Fixes the case reported by @bluetech in PR review. Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Sonnet 4 --- src/_pytest/assertion/rewrite.py | 19 +++++++++---------- testing/test_assertrewrite.py | 22 ++++++++++++++++++++-- 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index 77bf81b8d69..8714be63f47 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -940,9 +940,8 @@ def visit_Assert(self, assert_: ast.Assert) -> list[ast.stmt]: def visit_NamedExpr(self, name: ast.NamedExpr) -> tuple[ast.NamedExpr, str]: # Return the NamedExpr as-is so it evaluates in its natural position - # (preserving left-to-right evaluation order). For the explanation, - # reference the target variable (already assigned by the walrus) to - # avoid re-evaluating the expression. + # (preserving left-to-right evaluation order in function calls, etc.). + # For the explanation, reference the target variable. locs = ast.Call(self.builtin("locals"), [], []) target_id = name.target.id target_name = ast.Name(target_id, ast.Load()) @@ -981,12 +980,17 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: self.push_format_context() res, expl = self.visit(v) body.append(ast.Assign([ast.Name(res_var, ast.Store())], res)) + # For Name/NamedExpr operands, track the value in a stable + # @py_assert variable so the explanation shows the value at + # evaluation time — even if a later walrus overwrites the name. + if isinstance(v, ast.NamedExpr | ast.Name): + tracked = self.assign(ast.Name(res_var, ast.Load())) + for key in self.stack[-1]: + self.stack[-1][key] = self.display(tracked) expl_format = self.pop_format_context(ast.Constant(expl)) call = ast.Call(app, [expl_format], []) self.expl_stmts.append(ast.Expr(call)) if i < levels: - # Use res_var (already assigned above) rather than res directly, - # so that NamedExpr operands aren't evaluated a second time. cond: ast.expr = ast.Name(res_var, ast.Load()) if is_or: cond = ast.UnaryOp(ast.Not(), cond) @@ -1069,8 +1073,6 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: left_res, left_expl = self.visit(comp.left) if isinstance(comp.left, ast.Compare | ast.BoolOp): left_expl = f"({left_expl})" - # If the left operand is a NamedExpr, assign it to a temp so the - # walrus executes before any right-side expressions are hoisted. if isinstance(left_res, ast.NamedExpr): left_res = self.assign(left_res) res_variables = [self.variable() for i in range(len(comp.ops))] @@ -1095,9 +1097,6 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: next_res, next_expl = self.visit(next_operand) if isinstance(next_operand, ast.Compare | ast.BoolOp): next_expl = f"({next_expl})" - # Assign NamedExpr comparators to a temp so each walrus evaluates - # exactly once — critical for chained comparisons where the same - # node would otherwise be re-evaluated as left_res next iteration. if isinstance(next_res, ast.NamedExpr): next_res = self.assign(next_res) results.append(next_res) diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index 6e5225de992..3d7643f0f8b 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -1711,7 +1711,7 @@ def test_walrus_operator_change_boolean_value(): ) result = pytester.runpytest() assert result.ret == 1 - result.stdout.fnmatch_lines(["*assert not (False and False is False)"]) + result.stdout.fnmatch_lines(["*assert not (True and False is False)"]) def test_assertion_walrus_operator_boolean_none_fails( self, pytester: Pytester @@ -1725,7 +1725,7 @@ def test_walrus_operator_change_boolean_value(): ) result = pytester.runpytest() assert result.ret == 1 - result.stdout.fnmatch_lines(["*assert not (None and None is None)"]) + result.stdout.fnmatch_lines(["*assert not (True and None is None)"]) def test_assertion_walrus_operator_value_changes_cleared_after_each_test( self, pytester: Pytester @@ -1970,6 +1970,24 @@ def test_walrus_chained(): result = pytester.runpytest() assert result.ret == 0 + def test_walrus_boolop_same_target_correct_explanation( + self, pytester: Pytester + ) -> None: + """Multiple walrus operators to the same name in a BoolOp must show + each operand's value at evaluation time, not the final value.""" + pytester.makepyfile( + """ + def side_effect(): + return True + + def test_walrus_boolop(): + assert (x := side_effect()) and (x := False) + """ + ) + result = pytester.runpytest() + assert result.ret == 1 + result.stdout.fnmatch_lines(["*assert (True and False)"]) + @pytest.mark.skipif( sys.maxsize <= (2**31 - 1), reason="Causes OverflowError on 32bit systems" From cc256d14bf1422d7c56c8d107eb39dae97fd9533 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Wed, 3 Jun 2026 11:55:59 +0200 Subject: [PATCH 08/13] scripts: add assert-rewrite dump and diff tools Add scripts to inspect and compare assertion rewriting across pytest versions, useful for tracking changes in the rewriter output. - dump-assert-rewrite.py: dumps the rewritten form of a Python file using any pytest version (via uv ephemeral env) or the local worktree. Supports source, ast, and compact output formats. - diff-assert-rewrite.py: compares rewrite output between two versions (or worktree) with colored unified diff, runs both sides in parallel. - example_asserts.py: example file covering common assertion patterns including bluetech's walrus-in-BoolOp case from #14445. Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Opus 4 --- scripts/diff-assert-rewrite.py | 202 +++++++++++++++++++++++++++++++++ scripts/dump-assert-rewrite.py | 166 +++++++++++++++++++++++++++ scripts/example_asserts.py | 42 +++++++ 3 files changed, 410 insertions(+) create mode 100644 scripts/diff-assert-rewrite.py create mode 100644 scripts/dump-assert-rewrite.py create mode 100644 scripts/example_asserts.py diff --git a/scripts/diff-assert-rewrite.py b/scripts/diff-assert-rewrite.py new file mode 100644 index 00000000000..070b4040d71 --- /dev/null +++ b/scripts/diff-assert-rewrite.py @@ -0,0 +1,202 @@ +"""Compare assert-rewrite output between two pytest versions (or worktree). + +Runs the dump script for both sides (in parallel) and shows a unified diff. +Supports all three output formats: source, ast, and compact. + +Usage:: + + # Two released versions: + python scripts/diff-assert-rewrite.py --left 7.4.0 --right 8.0.0 example.py + + # Release vs local worktree: + python scripts/diff-assert-rewrite.py --left 8.3.0 --right worktree example.py + + # Compact AST diff (strips position noise): + python scripts/diff-assert-rewrite.py --left 7.4.0 --right worktree -f compact example.py + + # All formats at once: + python scripts/diff-assert-rewrite.py --left 7.4.0 --right 8.0.0 -f all example.py +""" + +from __future__ import annotations + +import argparse +from concurrent.futures import ThreadPoolExecutor +import difflib +from pathlib import Path +import sys + + +_DUMP_SCRIPT = Path(__file__).resolve().parent / "dump-assert-rewrite.py" + +_FORMATS = ("source", "ast", "compact") + + +def _label(spec: str) -> str: + return "worktree" if spec == "worktree" else f"pytest=={spec}" + + +def get_dump(spec: str, file_path: Path, fmt: str) -> str: + """Run the dump script for a single side and return its output.""" + # Import inline so this file stays lightweight at module level. + import subprocess + + args = [sys.executable, str(_DUMP_SCRIPT)] + if spec == "worktree": + args.append("--worktree") + else: + args.extend(["--pytest-version", spec]) + args.extend(["--format", fmt, str(file_path)]) + + result = subprocess.run(args, capture_output=True, text=True, check=False) + if result.returncode != 0: + sys.stderr.write(result.stderr) + raise SystemExit(f"Dump failed for {_label(spec)}") + return result.stdout + + +def colored_diff(lines: list[str]) -> str: + """Apply ANSI colours to a unified-diff line list.""" + RED = "\033[31m" + GREEN = "\033[32m" + CYAN = "\033[36m" + RESET = "\033[0m" + + out: list[str] = [] + for line in lines: + if line.startswith(("---", "+++")): + out.append(f"{CYAN}{line}{RESET}") + elif line.startswith("-"): + out.append(f"{RED}{line}{RESET}") + elif line.startswith("+"): + out.append(f"{GREEN}{line}{RESET}") + elif line.startswith("@@"): + out.append(f"{CYAN}{line}{RESET}") + else: + out.append(line) + return "\n".join(out) + + +def show_diff( + left_text: str, + right_text: str, + *, + left_label: str, + right_label: str, + fmt: str, + context: int, + use_color: bool, +) -> bool: + """Print a unified diff; return True if differences were found.""" + if left_text == right_text: + return False + + diff_lines = list( + difflib.unified_diff( + left_text.splitlines(), + right_text.splitlines(), + fromfile=f"{left_label} [{fmt}]", + tofile=f"{right_label} [{fmt}]", + n=context, + ) + ) + + if use_color: + print(colored_diff(diff_lines)) + else: + print("\n".join(diff_lines)) + + return True + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--left", + required=True, + metavar="VER|worktree", + help="Left side: pytest version or 'worktree'", + ) + parser.add_argument( + "--right", + required=True, + metavar="VER|worktree", + help="Right side: pytest version or 'worktree'", + ) + parser.add_argument( + "--format", + "-f", + dest="fmt", + choices=(*_FORMATS, "all"), + default="source", + help="Output format (default: source)", + ) + parser.add_argument( + "--context", + "-C", + type=int, + default=3, + help="Context lines in diff (default: 3)", + ) + parser.add_argument( + "--no-color", + action="store_true", + help="Disable coloured output", + ) + parser.add_argument("file", type=Path, help="Python file to compare") + args = parser.parse_args(argv) + + if not args.file.is_file(): + raise SystemExit(f"File not found: {args.file}") + + formats = _FORMATS if args.fmt == "all" else (args.fmt,) + use_color = not args.no_color and sys.stdout.isatty() + left_label = _label(args.left) + right_label = _label(args.right) + + # Fetch all needed dumps in parallel (both sides x all formats). + jobs: dict[tuple[str, str], str] = {} + with ThreadPoolExecutor(max_workers=len(formats) * 2) as pool: + futures = { + (side, fmt): pool.submit(get_dump, spec, args.file, fmt) + for fmt in formats + for side, spec in [("left", args.left), ("right", args.right)] + } + for key, future in futures.items(): + jobs[key] = future.result() + + any_diff = False + for fmt in formats: + if len(formats) > 1: + header = f"=== {fmt} ===" + if use_color: + header = f"\033[1m{header}\033[0m" + print(header) + + had_diff = show_diff( + jobs[("left", fmt)], + jobs[("right", fmt)], + left_label=left_label, + right_label=right_label, + fmt=fmt, + context=args.context, + use_color=use_color, + ) + if not had_diff: + print( + f"No differences in {fmt} output between {left_label} and {right_label}" + ) + else: + any_diff = True + + if len(formats) > 1: + print() + + raise SystemExit(1 if any_diff else 0) + + +if __name__ == "__main__": + main() diff --git a/scripts/dump-assert-rewrite.py b/scripts/dump-assert-rewrite.py new file mode 100644 index 00000000000..75e0b0d734f --- /dev/null +++ b/scripts/dump-assert-rewrite.py @@ -0,0 +1,166 @@ +"""Dump the assert-rewritten form of a Python file for a specific pytest version. + +Uses ``uv run`` to execute the rewriter in an ephemeral environment, so any +released pytest version can be inspected without installing it globally. + +Usage:: + + # Rewritten source (default): + python scripts/dump-assert-rewrite.py --pytest-version 8.0.0 example.py + + # Using local worktree: + python scripts/dump-assert-rewrite.py --worktree example.py + + # Compact AST (best for diffing -- no position attributes): + python scripts/dump-assert-rewrite.py --worktree --format compact example.py + + # Full AST with positions: + python scripts/dump-assert-rewrite.py --pytest-version 7.4.0 --format ast example.py +""" + +from __future__ import annotations + +import argparse +import os +from pathlib import Path +import subprocess +import sys +import textwrap + + +# Self-contained script executed inside the target pytest environment. +# Reads source from stdin, writes the rewritten form to stdout. +_WORKER_SCRIPT = textwrap.dedent("""\ + import ast + import sys + + source = sys.stdin.buffer.read() + fmt = sys.argv[1] + + try: + from _pytest.assertion.rewrite import rewrite_asserts + except ImportError: + sys.exit("Could not import rewrite_asserts from this pytest version") + + tree = ast.parse(source) + try: + rewrite_asserts(tree, source) + except TypeError: + # pytest < 6 did not accept the source parameter + tree = ast.parse(source) + rewrite_asserts(tree) + + ast.fix_missing_locations(tree) + + if fmt == "source": + print(ast.unparse(tree)) + elif fmt == "ast": + print(ast.dump(tree, indent=2)) + elif fmt == "compact": + print(ast.dump(tree, indent=2, include_attributes=False)) + else: + sys.exit(f"Unknown format: {fmt!r}") +""") + + +def run_worker( + *, + pytest_version: str | None, + worktree: bool, + file_content: bytes, + fmt: str, +) -> str: + """Execute the worker script and return its stdout.""" + if worktree: + repo_root = Path(__file__).resolve().parent.parent + env = os.environ.copy() + env["PYTHONPATH"] = ( + str(repo_root / "src") + os.pathsep + env.get("PYTHONPATH", "") + ) + cmd = [sys.executable, "-c", _WORKER_SCRIPT, fmt] + else: + assert pytest_version is not None + cmd = [ + "uv", + "run", + "--no-project", + "--with", + f"pytest=={pytest_version}", + "--", + "python", + "-c", + _WORKER_SCRIPT, + fmt, + ] + env = None + + try: + result = subprocess.run( + cmd, input=file_content, capture_output=True, check=False, env=env + ) + except FileNotFoundError as exc: + if "uv" in str(exc): + raise SystemExit( + "'uv' not found — install it: https://docs.astral.sh/uv/" + ) from exc + raise + + if result.returncode != 0: + label = version_label(pytest_version=pytest_version, worktree=worktree) + sys.stderr.buffer.write(result.stderr) + raise SystemExit(f"Worker failed for {label}") + + return result.stdout.decode() + + +def version_label(*, pytest_version: str | None = None, worktree: bool = False) -> str: + """Human-readable label for a pytest source.""" + return "worktree" if worktree else f"pytest=={pytest_version}" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + src = parser.add_mutually_exclusive_group(required=True) + src.add_argument( + "--pytest-version", + metavar="VER", + help="Released pytest version (e.g. 8.0.0)", + ) + src.add_argument( + "--worktree", + action="store_true", + help="Use the local worktree's src/", + ) + parser.add_argument( + "--format", + dest="fmt", + choices=("source", "ast", "compact"), + default="source", + help="Output format (default: source)", + ) + parser.add_argument("file", type=Path, help="Python file to rewrite") + return parser + + +def main(argv: list[str] | None = None) -> None: + args = build_parser().parse_args(argv) + + if not args.file.is_file(): + raise SystemExit(f"File not found: {args.file}") + + output = run_worker( + pytest_version=args.pytest_version, + worktree=args.worktree, + file_content=args.file.read_bytes(), + fmt=args.fmt, + ) + sys.stdout.write(output) + if output and not output.endswith("\n"): + sys.stdout.write("\n") + + +if __name__ == "__main__": + main() diff --git a/scripts/example_asserts.py b/scripts/example_asserts.py new file mode 100644 index 00000000000..22571d9995f --- /dev/null +++ b/scripts/example_asserts.py @@ -0,0 +1,42 @@ +"""Example file for dump-assert-rewrite.py and diff-assert-rewrite.py. + +Includes bluetech's walrus-in-BoolOp case from the #14445 review, plus +other common assertion patterns useful for spotting rewrite changes. +""" + +from __future__ import annotations + + +def side_effect() -> bool: + return True + + +def test_walrus_boolop() -> None: + """Bluetech's example: walrus reassignment inside BoolOp.""" + assert (x := side_effect()) and (x := False) # noqa: F841 + + +def test_simple_equality() -> None: + x = 1 + assert x == 2 + + +def test_comparison() -> None: + a = [1, 2, 3] + b = [1, 2, 4] + assert a == b + + +def test_boolean() -> None: + x = True + y = False + assert x and y + + +def test_membership() -> None: + assert 5 in [1, 2, 3] + + +def test_message() -> None: + value = 42 + assert value > 100, f"expected > 100, got {value}" From e6931af1a674abe650aeeaa3d7c5dfcec48c0ea0 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Wed, 3 Jun 2026 11:56:28 +0200 Subject: [PATCH 09/13] refactor(rewrite): minimal snapshots in BoolOp for walrus conflicts Replace the blanket snapshot-all-operands approach with a targeted one: pre-scan the BoolOp to find walrus targets, then only snapshot operands whose value a later walrus would corrupt. Snapshot rules: - NamedExpr (non-last): always, to avoid re-evaluating side effects - Name with later walrus conflict: to freeze the pre-overwrite value - Everything else: use res directly (stable @py_assert or plain name) Non-walrus BoolOps now generate identical code to 8.3.5 (no snapshots). Co-authored-by: Cursor AI Co-authored-by: Anthropic Claude Opus 4 --- src/_pytest/assertion/rewrite.py | 52 +++++++++++++++++++++----------- 1 file changed, 34 insertions(+), 18 deletions(-) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index 8714be63f47..13925712897 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -969,40 +969,56 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: body = save = self.statements fail_save = self.expl_stmts levels = len(boolop.values) - 1 + # Pre-scan: for each operand position, collect the set of variable + # names that a *later* operand's walrus operator will overwrite. + # An operand needs a snapshot only when its value references a name + # in this set (otherwise the explanation would show the post-walrus + # value instead of the value at evaluation time). + later_walrus_targets: list[set[str]] = [set() for _ in boolop.values] + seen: set[str] = set() + for idx in range(len(boolop.values) - 1, -1, -1): + later_walrus_targets[idx] = set(seen) + for node in ast.walk(boolop.values[idx]): + if isinstance(node, ast.NamedExpr): + seen.add(node.target.id) self.push_format_context() - # Process each operand, short-circuiting if needed. + # Process each operand, short-circuiting as needed. for i, v in enumerate(boolop.values): if i: fail_inner: list[ast.stmt] = [] - # expl_cond is set in a prior loop iteration below - self.expl_stmts.append(ast.If(expl_cond, fail_inner, [])) # noqa: F821 + # cond is set in a prior loop iteration below + self.expl_stmts.append(ast.If(cond, fail_inner, [])) # noqa: F821 self.expl_stmts = fail_inner self.push_format_context() res, expl = self.visit(v) body.append(ast.Assign([ast.Name(res_var, ast.Store())], res)) - # For Name/NamedExpr operands, track the value in a stable - # @py_assert variable so the explanation shows the value at - # evaluation time — even if a later walrus overwrites the name. - if isinstance(v, ast.NamedExpr | ast.Name): - tracked = self.assign(ast.Name(res_var, ast.Load())) + # Snapshot when the raw ``res`` node would be unsafe to reuse + # as a condition or explanation reference: + # - NamedExpr (non-last): reusing the node re-evaluates the + # walrus expression including any side effects. + # - Name whose variable a later walrus overwrites: the + # explanation would show the post-walrus value. + needs_snapshot = (isinstance(v, ast.NamedExpr) and i < levels) or ( + isinstance(v, ast.Name) and v.id in later_walrus_targets[i] + ) + if needs_snapshot: + snapshot = self.assign(ast.Name(res_var, ast.Load())) + res = snapshot for key in self.stack[-1]: - self.stack[-1][key] = self.display(tracked) + self.stack[-1][key] = self.display(snapshot) expl_format = self.pop_format_context(ast.Constant(expl)) call = ast.Call(app, [expl_format], []) self.expl_stmts.append(ast.Expr(call)) if i < levels: - cond: ast.expr = ast.Name(res_var, ast.Load()) + # Short-circuit: and → continue if truthy; or → if falsy. + # ``res`` is a stable reference (Name vars are only + # snapshotted when a later walrus would corrupt them; + # calls/compares return @py_assert vars from assign()). + cond: ast.expr = res if is_or: cond = ast.UnaryOp(ast.Not(), cond) - # Capture the condition in a stable temp for the explanation - # path — res_var is overwritten by subsequent operands. - cond_var = self.variable() - body.append(ast.Assign([ast.Name(cond_var, ast.Store())], cond)) - expl_cond: ast.expr = ast.Name(cond_var, ast.Load()) # noqa: F841 inner: list[ast.stmt] = [] - self.statements.append( - ast.If(ast.Name(cond_var, ast.Load()), inner, []) - ) + self.statements.append(ast.If(cond, inner, [])) self.statements = body = inner self.statements = save self.expl_stmts = fail_save From 6caf74a4ae1872086dec142a0133ce58e29c1229 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 19:52:36 +0200 Subject: [PATCH 10/13] fix(rewrite): freeze operands a later walrus would clobber The rewriter hoists each operand into its own statement, but a plain name is left as a bare load evaluated when the enclosing expression is assembled -- after the statements of the operands that follow it. A walrus operator in a later operand rebinds the name in between, so both the value used and the value reported were the post-walrus one, while Python evaluates the earlier operand first: assert value != identity(value := value.lower()) visit_BoolOp already guarded against this; extract its pre-scan as _walrus_targets() and add visit_operand() to apply the same freeze in visit_Compare, visit_Call and visit_BinOp. visit_Compare previously matched only a comparator that *was* a NamedExpr, missing walrus operators nested inside it; visit_Call did not guard at all, so an earlier argument saw a later argument's assignment. These cases predate the walrus rework -- they fail on main too. Closes the single-eval-walrus, order-compare-left, order-call-argument and order-binop-left groups in the coverage matrix. order-call-argument keeps one entry: a bare walrus argument is still substituted into a later one, which visit_operand does not yet see because the operand is a NamedExpr rather than a Name. Reported-by: Denis Scapin --- changelog/14445.bugfix.rst | 2 + src/_pytest/assertion/rewrite.py | 79 ++++++++++++++++---------- testing/test_assertrewrite_coverage.py | 40 ------------- 3 files changed, 50 insertions(+), 71 deletions(-) diff --git a/changelog/14445.bugfix.rst b/changelog/14445.bugfix.rst index aaae0c615f5..a9547581073 100644 --- a/changelog/14445.bugfix.rst +++ b/changelog/14445.bugfix.rst @@ -1 +1,3 @@ Fixed assertion rewriting evaluating walrus operator (``:=``) expressions multiple times, causing incorrect test results when the expression had side effects (e.g., incrementing a counter or calling a function). + +Operands preceding a walrus operator are now evaluated -- and reported -- before it rebinds their name, so ``assert value != identity(value := value.lower())`` keeps Python's left-to-right evaluation order. diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index 13925712897..0d95095dd67 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -540,6 +540,16 @@ def traverse_node(node: ast.AST) -> Iterator[ast.AST]: yield from traverse_node(child) +def _walrus_targets(nodes: Iterable[ast.expr]) -> set[str]: + """Return the names any walrus operator in *nodes* rebinds.""" + return { + sub.target.id + for node in nodes + for sub in ast.walk(node) + if isinstance(sub, ast.NamedExpr) + } + + @functools.lru_cache(maxsize=1) def _get_assertion_exprs(src: bytes) -> dict[int, str]: """Return a mapping from {lineno: "assertion test expression"}.""" @@ -961,6 +971,27 @@ def visit_Name(self, name: ast.Name) -> tuple[ast.Name, str]: expr = ast.IfExp(test, self.display(name), ast.Constant(name.id)) return name, self.explanation_param(expr) + def visit_operand( + self, operand: ast.expr, later: Sequence[ast.expr] + ) -> tuple[ast.expr, str]: + """Visit an operand, freezing it against walrus operators in *later*. + + Operands are rewritten into statements that run in source order, but + a plain name is left as a bare load evaluated at the very end, when + the enclosing expression is assembled. A walrus operator in a later + operand rebinds that name in between, so both the value used and the + value reported would be the post-walrus one -- Python evaluates the + earlier operand first. Copy the value into a temporary instead. + """ + specifiers = set(self.explanation_specifiers) + res, expl = self.visit(operand) + if isinstance(res, ast.Name) and res.id in _walrus_targets(later): + snapshot = self.assign(res) + for key in set(self.explanation_specifiers) - specifiers: + self.explanation_specifiers[key] = self.display(snapshot) + res = snapshot + return res, expl + def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: res_var = self.variable() expl_list = self.assign(ast.List([], ast.Load())) @@ -969,18 +1000,10 @@ def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: body = save = self.statements fail_save = self.expl_stmts levels = len(boolop.values) - 1 - # Pre-scan: for each operand position, collect the set of variable - # names that a *later* operand's walrus operator will overwrite. - # An operand needs a snapshot only when its value references a name - # in this set (otherwise the explanation would show the post-walrus - # value instead of the value at evaluation time). - later_walrus_targets: list[set[str]] = [set() for _ in boolop.values] - seen: set[str] = set() - for idx in range(len(boolop.values) - 1, -1, -1): - later_walrus_targets[idx] = set(seen) - for node in ast.walk(boolop.values[idx]): - if isinstance(node, ast.NamedExpr): - seen.add(node.target.id) + later_walrus_targets = [ + _walrus_targets(boolop.values[idx + 1 :]) + for idx in range(len(boolop.values)) + ] self.push_format_context() # Process each operand, short-circuiting as needed. for i, v in enumerate(boolop.values): @@ -1034,7 +1057,7 @@ def visit_UnaryOp(self, unary: ast.UnaryOp) -> tuple[ast.Name, str]: def visit_BinOp(self, binop: ast.BinOp) -> tuple[ast.Name, str]: symbol = BINOP_MAP[binop.op.__class__] - left_expr, left_expl = self.visit(binop.left) + left_expr, left_expl = self.visit_operand(binop.left, [binop.right]) right_expr, right_expl = self.visit(binop.right) explanation = f"({left_expl} {symbol} {right_expl})" res = self.assign( @@ -1043,16 +1066,19 @@ def visit_BinOp(self, binop: ast.BinOp) -> tuple[ast.Name, str]: return res, explanation def visit_Call(self, call: ast.Call) -> tuple[ast.Name, str]: - new_func, func_expl = self.visit(call.func) + # The callee and every argument are evaluated left to right, so each of + # them has to be frozen against walrus operators in what follows. + operands = [*call.args, *(keyword.value for keyword in call.keywords)] + new_func, func_expl = self.visit_operand(call.func, operands) arg_expls = [] new_args = [] new_kwargs = [] - for arg in call.args: - res, expl = self.visit(arg) + for i, arg in enumerate(call.args): + res, expl = self.visit_operand(arg, operands[i + 1 :]) arg_expls.append(expl) new_args.append(res) - for keyword in call.keywords: - res, expl = self.visit(keyword.value) + for i, keyword in enumerate(call.keywords, start=len(call.args)): + res, expl = self.visit_operand(keyword.value, operands[i + 1 :]) new_kwargs.append(ast.keyword(keyword.arg, res)) if keyword.arg: arg_expls.append(keyword.arg + "=" + expl) @@ -1086,7 +1112,7 @@ def visit_Attribute(self, attr: ast.Attribute) -> tuple[ast.Name, str]: def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: self.push_format_context() - left_res, left_expl = self.visit(comp.left) + left_res, left_expl = self.visit_operand(comp.left, comp.comparators) if isinstance(comp.left, ast.Compare | ast.BoolOp): left_expl = f"({left_expl})" if isinstance(left_res, ast.NamedExpr): @@ -1099,18 +1125,9 @@ def visit_Compare(self, comp: ast.Compare) -> tuple[ast.expr, str]: syms: list[ast.expr] = [] results = [left_res] for i, op, next_operand in it: - # If the next operand is a walrus that assigns to the same name as - # the current left_res, we must freeze left_res's value before the - # walrus modifies it. - match (next_operand, left_res): - case ( - ast.NamedExpr(target=ast.Name(id=target_id)), - ast.Name(id=name_id), - ) if target_id == name_id: - left_res = self.assign(left_res) - results[-1] = left_res - - next_res, next_expl = self.visit(next_operand) + next_res, next_expl = self.visit_operand( + next_operand, comp.comparators[i + 1 :] + ) if isinstance(next_operand, ast.Compare | ast.BoolOp): next_expl = f"({next_expl})" if isinstance(next_res, ast.NamedExpr): diff --git a/testing/test_assertrewrite_coverage.py b/testing/test_assertrewrite_coverage.py index 054db3bb137..63346224a7a 100644 --- a/testing/test_assertrewrite_coverage.py +++ b/testing/test_assertrewrite_coverage.py @@ -1000,10 +1000,6 @@ def __getitem__(self, key): assert d["a"] == 100 """) - @pytest.mark.xfail( - strict=True, - reason="single-eval-walrus: the walrus expression runs twice", - ) def test_walrus_in_compare_evaluated_once(self) -> None: assert_single_evaluation(""" def check(): @@ -1013,10 +1009,6 @@ def side_effect(): assert (x := side_effect()) == 100 """) - @pytest.mark.xfail( - strict=True, - reason="single-eval-walrus: the walrus expression runs twice", - ) def test_walrus_in_boolean_evaluated_once(self) -> None: assert_single_evaluation(""" def check(): @@ -1026,10 +1018,6 @@ def side_effect(): assert (x := side_effect()) and False """) - @pytest.mark.xfail( - strict=True, - reason="single-eval-walrus: the walrus expression runs twice", - ) def test_walrus_in_chained_compare_evaluated_once(self) -> None: assert_single_evaluation(""" def check(): @@ -1116,10 +1104,6 @@ class TestEvaluationOrder: visitors take expression types away from ``generic_visit``. """ - @pytest.mark.xfail( - strict=True, - reason="order-compare-left: left operand is read after a walrus in the right one", - ) def test_compare_left_operand_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): @@ -1133,10 +1117,6 @@ def identity(v): return "passed", value """) - @pytest.mark.xfail( - strict=True, - reason="order-compare-left: a failing comparison reports the post-walrus value", - ) def test_compare_reports_left_operand(self) -> None: assert_introspects( """ @@ -1149,10 +1129,6 @@ def identity(v): must_contain=["assert 2 == 3"], ) - @pytest.mark.xfail( - strict=True, - reason="order-call-argument: an earlier argument is read after a later walrus", - ) def test_call_earlier_argument_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): @@ -1168,10 +1144,6 @@ def collect(*values): return "passed", value """) - @pytest.mark.xfail( - strict=True, - reason="order-binop-left: left operand is read after a walrus in the right one", - ) def test_binop_left_operand_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): @@ -1204,10 +1176,6 @@ def collect(*values): return "passed", items """) - @pytest.mark.xfail( - strict=True, - reason="order-compare-left: a chained comparison re-reads its left operand", - ) def test_chained_compare_operands_in_order(self) -> None: assert_evaluation_order(""" def check(): @@ -1318,10 +1286,6 @@ def take(self, value): return "passed", obj """) - @pytest.mark.xfail( - strict=True, - reason="order-call-argument: a keyword argument is read after a later walrus", - ) def test_keyword_argument_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): @@ -1337,10 +1301,6 @@ def collect(**kwargs): return "passed", value """) - @pytest.mark.xfail( - strict=True, - reason="order-call-argument: a ** argument is read after a later walrus", - ) def test_double_star_argument_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): From d271ad5520a2185f4d04fe67689151e86e2dff06 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 20:42:11 +0200 Subject: [PATCH 11/13] fix(rewrite): freeze walrus and starred operands too visit_operand() only froze a bare name, so two other unhoisted operands kept being evaluated after everything that follows them: assert collect((x := 1), identity(x := 2)) == (1, 2) assert collect(*items, identity(items := [9])) == (1, [9]) A walrus operator left in place assigns once the enclosing expression is assembled, which is after the later arguments have run -- so the earlier argument saw the later assignment. A starred argument hid its value inside an ast.Starred, where the existing Name check could not see it. Closes the order-starred-argument group and the remaining order-call-argument entry in the coverage matrix. --- changelog/14814.bugfix.rst | 1 + src/_pytest/assertion/rewrite.py | 35 ++++++++++++++++++++------ testing/test_assertrewrite_coverage.py | 8 ------ 3 files changed, 28 insertions(+), 16 deletions(-) create mode 100644 changelog/14814.bugfix.rst diff --git a/changelog/14814.bugfix.rst b/changelog/14814.bugfix.rst new file mode 100644 index 00000000000..6e44ab4976f --- /dev/null +++ b/changelog/14814.bugfix.rst @@ -0,0 +1 @@ +Fixed assertion rewriting evaluating a walrus operator (``:=``) or a starred argument out of order when a later argument assigned to the same name, so ``assert collect(*items, identity(items := [9]))`` now passes the pre-assignment ``items``. diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index 0d95095dd67..a6c2b7e28db 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -977,19 +977,38 @@ def visit_operand( """Visit an operand, freezing it against walrus operators in *later*. Operands are rewritten into statements that run in source order, but - a plain name is left as a bare load evaluated at the very end, when - the enclosing expression is assembled. A walrus operator in a later - operand rebinds that name in between, so both the value used and the - value reported would be the post-walrus one -- Python evaluates the - earlier operand first. Copy the value into a temporary instead. + two of them stay unhoisted and are evaluated at the very end, when the + enclosing expression is assembled -- after everything that follows + them: + + * a plain name, which a walrus operator in a later operand rebinds in + between, so the value used and the value reported would be the + post-walrus one; + * a walrus operator itself, which would then assign in the wrong + order, and be visible to the operands that were meant to precede it. + + Either way Python evaluates the earlier operand first, so copy it into + a temporary here. A starred argument is unwrapped and rewrapped, its + value being subject to the same problem. """ specifiers = set(self.explanation_specifiers) res, expl = self.visit(operand) - if isinstance(res, ast.Name) and res.id in _walrus_targets(later): - snapshot = self.assign(res) + value = res.value if isinstance(res, ast.Starred) else res + if isinstance(value, ast.NamedExpr): + needs_freeze = bool(later) + elif isinstance(value, ast.Name): + needs_freeze = value.id in _walrus_targets(later) + else: + needs_freeze = False + if needs_freeze: + snapshot = self.assign(value) for key in set(self.explanation_specifiers) - specifiers: self.explanation_specifiers[key] = self.display(snapshot) - res = snapshot + res = ( + ast.copy_location(ast.Starred(snapshot, res.ctx), res) + if isinstance(res, ast.Starred) + else snapshot + ) return res, expl def visit_BoolOp(self, boolop: ast.BoolOp) -> tuple[ast.Name, str]: diff --git a/testing/test_assertrewrite_coverage.py b/testing/test_assertrewrite_coverage.py index 63346224a7a..a8b5da83182 100644 --- a/testing/test_assertrewrite_coverage.py +++ b/testing/test_assertrewrite_coverage.py @@ -1157,10 +1157,6 @@ def identity(v): return "passed", value """) - @pytest.mark.xfail( - strict=True, - reason="order-starred-argument: a starred argument is read after a later walrus", - ) def test_starred_argument_precedes_walrus(self) -> None: assert_evaluation_order(""" def check(): @@ -1189,10 +1185,6 @@ def identity(v): return "passed", value """) - @pytest.mark.xfail( - strict=True, - reason="order-call-argument: a walrus argument is substituted into a later one", - ) def test_bare_walrus_argument_in_order(self) -> None: """A walrus argument is evaluated in place, before the ones after it.""" assert_evaluation_order(""" From 6a2742ba5bd27c7c40853b30a327cf976ef2ada7 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 20:43:29 +0200 Subject: [PATCH 12/13] feat(rewrite): introspect container[key] in failure messages A subscript was opaque: the message showed the value it produced with no indication of which container or key it came from. Decompose it the way attribute access already is. The container goes through visit_operand() because taking the expression away from generic_visit() takes away the hoisting that kept it ordered -- without that, `assert box[identity(box := other)] == 1` would start reading the post-walrus container. The order-axis guard in the coverage matrix fails if this is dropped. Slices keep the generic treatment; decomposing start/stop/step is rarely what a failure message needs. Closes the introspect-subscript group in the coverage matrix. --- changelog/14815.improvement.rst | 4 ++++ src/_pytest/assertion/rewrite.py | 17 +++++++++++++++++ testing/test_assertrewrite_coverage.py | 14 +------------- 3 files changed, 22 insertions(+), 13 deletions(-) create mode 100644 changelog/14815.improvement.rst diff --git a/changelog/14815.improvement.rst b/changelog/14815.improvement.rst new file mode 100644 index 00000000000..0e316edea07 --- /dev/null +++ b/changelog/14815.improvement.rst @@ -0,0 +1,4 @@ +Assertion failure messages now decompose subscript expressions, showing the container and the key that produced a value:: + + assert 1 == 99 + + where 1 = {'a': 1, 'b': 2}['a'] diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index a6c2b7e28db..b43c7f76187 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -1117,6 +1117,23 @@ def visit_Starred(self, starred: ast.Starred) -> tuple[ast.Starred, str]: new_starred = ast.Starred(res, starred.ctx) return new_starred, "*" + expl + def visit_Subscript(self, subscript: ast.Subscript) -> tuple[ast.Name, str]: + if not isinstance(subscript.ctx, ast.Load): + return self.generic_visit(subscript) + # For Slice objects (a[1:3]), fall back to generic — decomposing + # start/stop/step is rarely useful in assertion messages. + if isinstance(subscript.slice, ast.Slice): + return self.generic_visit(subscript) + value, value_expl = self.visit_operand(subscript.value, [subscript.slice]) + slice_res, slice_expl = self.visit(subscript.slice) + res = self.assign( + ast.copy_location(ast.Subscript(value, slice_res, ast.Load()), subscript) + ) + res_expl = self.explanation_param(self.display(res)) + pat = "%s\n{%s = %s[%s]\n}" + expl = pat % (res_expl, res_expl, value_expl, slice_expl) + return res, expl + def visit_Attribute(self, attr: ast.Attribute) -> tuple[ast.Name, str]: if not isinstance(attr.ctx, ast.Load): return self.generic_visit(attr) diff --git a/testing/test_assertrewrite_coverage.py b/testing/test_assertrewrite_coverage.py index a8b5da83182..85a1c9a5716 100644 --- a/testing/test_assertrewrite_coverage.py +++ b/testing/test_assertrewrite_coverage.py @@ -639,12 +639,8 @@ def check(): class TestIntrospectionSubscript: - """Subscript / indexing — currently hits generic_visit.""" + """Subscript / indexing — has a dedicated visitor.""" - @pytest.mark.xfail( - strict=True, - reason="introspect-subscript: container[key] is not decomposed", - ) def test_dict_subscript_shows_key_and_container(self) -> None: assert_introspects( """ @@ -655,10 +651,6 @@ def check(): must_contain=["where 1 = ", "['a']"], ) - @pytest.mark.xfail( - strict=True, - reason="introspect-subscript: container[key] is not decomposed", - ) def test_list_subscript_shows_index_and_container(self) -> None: assert_introspects( """ @@ -1427,10 +1419,6 @@ def identity(v): class TestEdgeCases: """Regression and edge-case tests combining multiple expression types.""" - @pytest.mark.xfail( - strict=True, - reason="introspect-subscript: container[key] is not decomposed", - ) def test_subscript_with_variable_key(self) -> None: """Subscript where the key is a variable (not constant).""" assert_introspects( From 8bc3e80cd7a9e5b7559508ae59d3e2af6bf8654c Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 20:44:22 +0200 Subject: [PATCH 13/13] feat(rewrite): introspect the condition of a ternary A conditional expression showed only its result, so a failure gave no hint which way it went. Introspect the condition and report it as "(... if else ...)". The branches keep their original nodes: only the selected one may run, so neither can be hoisted into a statement. That leaves them evaluated after the condition, which is Python's order, so unlike the subscript container they need no freeze -- the order-axis guard covers it. Closes the introspect-ifexp group in the coverage matrix. --- changelog/14816.improvement.rst | 4 ++++ src/_pytest/assertion/rewrite.py | 14 ++++++++++++++ testing/test_assertrewrite_coverage.py | 14 +------------- 3 files changed, 19 insertions(+), 13 deletions(-) create mode 100644 changelog/14816.improvement.rst diff --git a/changelog/14816.improvement.rst b/changelog/14816.improvement.rst new file mode 100644 index 00000000000..ca3ba124728 --- /dev/null +++ b/changelog/14816.improvement.rst @@ -0,0 +1,4 @@ +Assertion failure messages now show the condition of a conditional expression:: + + assert 0 == 99 + + where 0 = (... if True else ...) diff --git a/src/_pytest/assertion/rewrite.py b/src/_pytest/assertion/rewrite.py index b43c7f76187..f0a0ae455a6 100644 --- a/src/_pytest/assertion/rewrite.py +++ b/src/_pytest/assertion/rewrite.py @@ -1117,6 +1117,20 @@ def visit_Starred(self, starred: ast.Starred) -> tuple[ast.Starred, str]: new_starred = ast.Starred(res, starred.ctx) return new_starred, "*" + expl + def visit_IfExp(self, ifexp: ast.IfExp) -> tuple[ast.Name, str]: + # Introspect the condition but keep the branches as they are: only the + # selected one may be evaluated, so neither can be hoisted. That also + # keeps them ordered after the condition, which is where Python puts + # them, so no freeze is needed here. + cond_res, cond_expl = self.visit(ifexp.test) + res = self.assign( + ast.copy_location(ast.IfExp(cond_res, ifexp.body, ifexp.orelse), ifexp) + ) + res_expl = self.explanation_param(self.display(res)) + pat = "%s\n{%s = (... if %s else ...)\n}" + expl = pat % (res_expl, res_expl, cond_expl) + return res, expl + def visit_Subscript(self, subscript: ast.Subscript) -> tuple[ast.Name, str]: if not isinstance(subscript.ctx, ast.Load): return self.generic_visit(subscript) diff --git a/testing/test_assertrewrite_coverage.py b/testing/test_assertrewrite_coverage.py index 85a1c9a5716..8396833d1eb 100644 --- a/testing/test_assertrewrite_coverage.py +++ b/testing/test_assertrewrite_coverage.py @@ -681,12 +681,8 @@ def check(): class TestIntrospectionIfExp: - """Ternary / if-expression — currently hits generic_visit.""" + """Ternary / if-expression — has a dedicated visitor.""" - @pytest.mark.xfail( - strict=True, - reason="introspect-ifexp: ternaries are not decomposed", - ) def test_ifexp_shows_condition_value(self) -> None: assert_introspects( """ @@ -704,10 +700,6 @@ def check(): assert (0 if flag else 1) == 1 """) - @pytest.mark.xfail( - strict=True, - reason="introspect-ifexp: ternaries are not decomposed", - ) def test_ifexp_in_compare_shows_result(self) -> None: assert_introspects( """ @@ -1515,10 +1507,6 @@ def __repr__(self): must_contain=["42", "100"], ) - @pytest.mark.xfail( - strict=True, - reason="introspect-ifexp: ternaries are not decomposed", - ) def test_ifexp_with_call_condition(self) -> None: """IfExp where condition is a function call.""" assert_introspects(