From ff0f7c0f1dab91241d61fe78955f7522ee04a7a8 Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Fri, 31 Jul 2026 20:39:15 +0200 Subject: [PATCH 01/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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("""