From 6dec6a36509f079c98ca8ecade4f886df5272614 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Fri, 13 Feb 2026 08:45:11 -0300 Subject: [PATCH] fix: handle VAR_POSITIONAL and kwargs precedence in _fast_bind cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The _fast_bind method had three bugs when replaying cached binding templates: 1. VAR_POSITIONAL (*args) params were stored as a single value (args[i]) instead of a tuple (args[i:]), causing TypeError when BoundArguments.args tried to extend() a non-iterable value. 2. KEYWORD_ONLY params after *args were not processed at all — the loop would break at the VAR_POSITIONAL and skip remaining params. 3. POSITIONAL_OR_KEYWORD params that share a name with a kwarg key should prefer the kwargs value (matching _full_bind behavior), but _fast_bind always used the positional arg. These bugs only manifested when the bind cache was shared across tests via the module-level signature_cache (different callables with the same signature sharing a SignatureAdapter instance). Fixes issues observed in SCXML W3C tests (test179, test527, test529, test562, test578) when run together with the bind cache from #550. --- statemachine/signature.py | 25 ++++++++--- tests/test_signature.py | 95 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 5 deletions(-) diff --git a/statemachine/signature.py b/statemachine/signature.py index 254fee52..ebdb1594 100644 --- a/statemachine/signature.py +++ b/statemachine/signature.py @@ -13,7 +13,7 @@ from typing import Tuple BindCacheKey = Tuple[int, FrozenSet[str]] -BindTemplate = Tuple[Tuple[str, ...], Optional[str]] # noqa: UP007 +BindTemplate = Tuple[Tuple[str, ...], Optional[str], Optional[str]] # noqa: UP007 def _make_key(method): @@ -89,12 +89,25 @@ def _fast_bind( kwargs: dict[str, Any], template: BindTemplate, ) -> BoundArguments: - param_names, kwargs_param_name = template + param_names, kwargs_param_name, var_positional_name = template arguments: dict[str, Any] = {} + past_var_positional = False for i, name in enumerate(param_names): - if i < len(args): - arguments[name] = args[i] + if name == var_positional_name: + # Collect all remaining positional args into a tuple + arguments[name] = args[i:] + past_var_positional = True + elif past_var_positional: + # After *args, remaining params are keyword-only + arguments[name] = kwargs.get(name) + elif i < len(args): + # Match _full_bind: if param is also in kwargs, kwargs wins + # (POSITIONAL_OR_KEYWORD params prefer kwargs over positional args) + if name in kwargs: + arguments[name] = kwargs[name] + else: + arguments[name] = args[i] else: arguments[name] = kwargs.get(name) @@ -124,6 +137,7 @@ def _full_bind( # noqa: C901 parameters_ex: Any = () kwargs_param = None kwargs_param_name: str | None = None + var_positional_name: str | None = None while True: # Let's iterate through the positional arguments and corresponding @@ -192,6 +206,7 @@ def _full_bind( # noqa: C901 values.extend(arg_vals) arguments[param.name] = tuple(values) param_names_used.append(param.name) + var_positional_name = param.name break if param.name in kwargs and param.kind != Parameter.POSITIONAL_ONLY: @@ -236,7 +251,7 @@ def _full_bind( # noqa: C901 # 'ignoring we got an unexpected keyword argument' pass - template: BindTemplate = (tuple(param_names_used), kwargs_param_name) + template: BindTemplate = (tuple(param_names_used), kwargs_param_name, var_positional_name) self._bind_cache[cache_key] = template return BoundArguments(self, arguments) # type: ignore[arg-type] diff --git a/tests/test_signature.py b/tests/test_signature.py index f8d37fbf..1cb68673 100644 --- a/tests/test_signature.py +++ b/tests/test_signature.py @@ -2,6 +2,7 @@ from functools import partial import pytest + from statemachine.dispatcher import callable_method from statemachine.signature import SignatureAdapter @@ -196,3 +197,97 @@ def test_kwargs_only_receives_unmatched_keys_with_positional(self): result2 = wrapped("X", target="Y") assert result2 == ("X", {"target": "Y"}) + + def test_var_positional_collected_as_tuple(self): + """VAR_POSITIONAL (*args) must be collected into a tuple on cache hit.""" + + def fn(*args, **kwargs): + return args, kwargs + + wrapped = callable_method(fn) + + result1 = wrapped(1, 2, 3, key="val") + assert result1 == ((1, 2, 3), {"key": "val"}) + + result2 = wrapped(4, 5, key="other") + assert result2 == ((4, 5), {"key": "other"}) + + def test_keyword_only_after_var_positional(self): + """KEYWORD_ONLY params after *args must be extracted from kwargs on cache hit.""" + + def fn(*args, event, **kwargs): + return args, event, kwargs + + wrapped = callable_method(fn) + + result1 = wrapped(100, event="ev1", source="s0") + assert result1 == ((100,), "ev1", {"source": "s0"}) + + result2 = wrapped(200, event="ev2", source="s1") + assert result2 == ((200,), "ev2", {"source": "s1"}) + + def test_positional_or_keyword_prefers_kwargs_over_positional(self): + """When a POSITIONAL_OR_KEYWORD param is in both args and kwargs, kwargs wins.""" + + def fn(event, source, target): + return event, source, target + + wrapped = callable_method(fn) + + # 1st call: positional arg provided but 'event' also in kwargs -> kwargs wins + result1 = wrapped("discarded_content", event="ev1", source="s0", target="t0") + assert result1 == ("ev1", "s0", "t0") + + # 2nd call: cache hit, same behavior expected + result2 = wrapped("other_content", event="ev2", source="s1", target="t1") + assert result2 == ("ev2", "s1", "t1") + + def test_empty_var_positional(self): + """Empty *args is handled correctly on cache hit.""" + + def fn(*args, **kwargs): + return args, kwargs + + wrapped = callable_method(fn) + + # 1st call with args + result1 = wrapped(1, key="val") + assert result1 == ((1,), {"key": "val"}) + + # 2nd call: only kwargs, no positional args — different cache key (len=0) + result2 = wrapped(key="val2") + assert result2 == ((), {"key": "val2"}) + + # 3rd call: hits cache for len=0 + result3 = wrapped(key="val3") + assert result3 == ((), {"key": "val3"}) + + def test_named_params_before_var_positional(self): + """Named params before *args are filled correctly on cache hit.""" + + def fn(a, b, *args, **kwargs): + return a, b, args, kwargs + + wrapped = callable_method(fn) + + result1 = wrapped(1, 2, 3, 4, key="val") + assert result1 == (1, 2, (3, 4), {"key": "val"}) + + result2 = wrapped(10, 20, 30, key="val2") + assert result2 == (10, 20, (30,), {"key": "val2"}) + + def test_kwargs_wins_with_var_positional_present(self): + """POSITIONAL_OR_KEYWORD before *args prefers kwargs when name matches.""" + + def fn(event, *args, **kwargs): + return event, args, kwargs + + wrapped = callable_method(fn) + + # 1st call: 'event' in both positional and kwargs — kwargs wins + result1 = wrapped("discarded", "extra", event="ev1", key="a") + assert result1 == ("ev1", ("extra",), {"key": "a"}) + + # 2nd call: cache hit, same behavior + result2 = wrapped("other", "more", event="ev2", key="b") + assert result2 == ("ev2", ("more",), {"key": "b"})