diff --git a/i2/tests/test_wrapper.py b/i2/tests/test_wrapper.py index 9a5c9393..60ccdca3 100644 --- a/i2/tests/test_wrapper.py +++ b/i2/tests/test_wrapper.py @@ -6,6 +6,7 @@ import pytest from i2.wrapper import ( + Wrap, wrap, mk_ingress_from_name_mapper, rm_params, @@ -534,3 +535,66 @@ def decorate(func=None, *, multiplier=2): with pytest.raises(TypeError): decorate(_incr, func=_incr) + + +# --------------------------------------------------------------------------------------- +# Regression: Wrap must not mutate the ingress it is given. +# `preserve_signature` used to stamp `__signature__`/`__annotations__` onto the ingress +# object. A decorator normally defines one ingress and reuses it for every function it +# wraps, so that write corrupted the caller's function AND made every later Wrap built +# from the same ingress advertise the first-wrapped function's signature -- while still +# executing correctly, so nothing surfaced the lie. + + +def _shared_ingress(*args, **kwargs): + """A module-level ingress, of the shape `preserve_signature='auto'` acts on.""" + return args, kwargs + + +def test_wrap_does_not_mutate_the_ingress(): + """Wrapping must leave the caller's ingress object exactly as it was.""" + from inspect import signature + + def func(a: int) -> int: + return a + + before = signature(_shared_ingress) + Wrap(func, ingress=_shared_ingress) + + assert signature(_shared_ingress) == before + assert not hasattr(_shared_ingress, "__signature__") + + +def test_reused_ingress_gives_each_wrap_its_own_signature(): + """Two Wraps sharing one ingress must each report their own func's signature.""" + + def f(a: int) -> int: + return a + + def g(x: str, y: str) -> str: + return x + y + + wrapped_f = Wrap(f, ingress=_shared_ingress) + wrapped_g = Wrap(g, ingress=_shared_ingress) + + assert str(Sig(wrapped_f)) == "(a: int) -> int" + assert str(Sig(wrapped_g)) == "(x: str, y: str) -> str" + # ... and both still actually work + assert wrapped_f(3) == 3 + assert wrapped_g("a", "b") == "ab" + + +def test_wrap_signature_is_order_independent(): + """Building the two Wraps in the other order must give the same signatures.""" + + def f(a: int) -> int: + return a + + def g(x: str, y: str) -> str: + return x + y + + wrapped_g = Wrap(g, ingress=_shared_ingress) + wrapped_f = Wrap(f, ingress=_shared_ingress) + + assert str(Sig(wrapped_g)) == "(x: str, y: str) -> str" + assert str(Sig(wrapped_f)) == "(a: int) -> int" diff --git a/i2/wrapper.py b/i2/wrapper.py index 0b8605a2..6fc0dd2e 100644 --- a/i2/wrapper.py +++ b/i2/wrapper.py @@ -103,6 +103,11 @@ InnerKwargs = dict KwargsTrans = Callable[[OuterKwargs], InnerKwargs] +#: ``preserve_signature`` value meaning "decide per ingress" (see +#: :func:`_should_preserve_signature`). Named rather than spelled ``'auto'`` at each +#: use so the sentinel has exactly one definition. +AUTO_PRESERVE_SIGNATURE = "auto" + def identity(x): """Transparent function, returning what's been input""" @@ -205,92 +210,102 @@ def _defaults_and_kwdefaults_of_func(func: Callable): return sig._defaults_, sig._kwdefaults_ -def _should_preserve_signature(ingress, func, preserve_mode): - """Determine if signature should be auto-preserved from func to ingress. +def _is_generic_signature(func: Callable) -> bool: + """Whether ``func``'s signature is exactly ``(*args, **kwargs)``. - Parameters - ---------- - ingress : callable or None - The ingress function - func : callable - The wrapped function - preserve_mode : 'auto' | True | False - The preservation mode + Such a signature carries no interface information, which is what makes it safe + to show the wrapped function's signature in its place. - Returns - ------- - bool - True if signature should be preserved + >>> _is_generic_signature(lambda *a, **kw: None) + True + >>> _is_generic_signature(lambda x, *a, **kw: None) + False + >>> _is_generic_signature(lambda *a: None) + False + >>> _is_generic_signature(lambda x: None) + False """ - if preserve_mode is False: + try: + params = list(signature(func).parameters.values()) + except (ValueError, TypeError): # no inspectable signature (some builtins) return False - if preserve_mode is True: - return True + return [p.kind for p in params] == [Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD] - # 'auto' mode: preserve if ingress has generic (*args, **kwargs) - if ingress is None: - return False - try: - ingress_sig = signature(ingress) - except (ValueError, TypeError): - # Can't get signature, don't preserve - return False +def _should_preserve_signature(ingress, preserve_mode): + """Whether the wrapper should advertise the wrapped function's signature instead + of ``ingress``'s. + + :param ingress: The incoming data transformer, or ``None`` + :param preserve_mode: ``True`` (always), ``False`` (never), or ``'auto'`` + :return: ``True`` if the wrapped function's signature should be used - params = list(ingress_sig.parameters.values()) + ``'auto'`` preserves only when the ingress signature is ``(*args, **kwargs)`` and + the ingress does not declare a ``__signature__`` of its own -- i.e. only when the + ingress says nothing about its interface, so there is nothing to overwrite: - # Check if ingress has exactly (*args, **kwargs) signature - if len(params) != 2: + >>> _should_preserve_signature(lambda *a, **kw: None, 'auto') + True + >>> _should_preserve_signature(lambda a: None, 'auto') + False + + ``True``/``False`` are unconditional, and are the way to override ``'auto'``'s guess: + + >>> _should_preserve_signature(lambda a: None, True) + True + >>> _should_preserve_signature(lambda *a, **kw: None, False) + False + """ + if preserve_mode != AUTO_PRESERVE_SIGNATURE: + return bool(preserve_mode) + if ingress is None: return False + return _is_generic_signature(ingress) and not hasattr(ingress, "__signature__") - is_generic = ( - params[0].kind == Parameter.VAR_POSITIONAL - and params[1].kind == Parameter.VAR_KEYWORD - ) - # Only preserve if generic and doesn't already have __signature__ - return is_generic and not hasattr(ingress, "__signature__") +def _return_annotation_of(func: Callable) -> Any: + """The return annotation of ``func``, or ``empty`` if it has none or has no + inspectable signature at all.""" + try: + return Sig(func).return_annotation + except (ValueError, TypeError): + return empty def _get_return_annotation(func, egress): - """Get return annotation with smart fallback logic. + """Resolve the return annotation a :class:`Wrap` should advertise. - Fallback chain: egress annotation → func annotation → empty + :param func: The wrapped function + :param egress: The outgoing data transformer, or ``None`` + :return: The return annotation to use, or ``empty`` - Parameters - ---------- - func : callable - The wrapped function - egress : callable or None - The egress function + The egress wins when it declares a return annotation, since it is what actually + produces the wrapper's output. Otherwise the annotation falls back to ``func``'s, + on the assumption that an unannotated egress does not change the type: - Returns - ------- - annotation - The return annotation to use, or Parameter.empty - """ - func_sig = Sig(func) - func_return = func_sig.return_annotation + >>> def f(x: int) -> str: return str(x) + >>> def annotated_egress(out) -> bytes: return out.encode() + >>> def bare_egress(out): return out - if egress is None: - # No egress: use func's return annotation - return func_return if func_return is not Parameter.empty else empty + >>> _get_return_annotation(f, None) + + >>> _get_return_annotation(f, annotated_egress) + + >>> _get_return_annotation(f, bare_egress) + - # Egress provided: check its annotation first - try: - egress_sig = Sig(egress) - egress_return = egress_sig.return_annotation - except (ValueError, TypeError): - # Can't get egress signature, fall back to func - return func_return if func_return is not Parameter.empty else empty - - if egress_return is not Parameter.empty: - # Egress has annotation, use it - return egress_return + An unannotated ``func`` yields ``empty``, not ``None`` -- ``None`` is a legitimate + annotation meaning "returns None", so the two must stay distinguishable: - # Egress has no annotation: fall back to func's annotation - # Assumption: egress doesn't transform the type - return func_return if func_return is not Parameter.empty else empty + >>> def g(x): pass + >>> _get_return_annotation(g, None) is empty + True + """ + if egress is not None: + egress_return = _return_annotation_of(egress) + if egress_return is not empty: + return egress_return + return _return_annotation_of(func) class Wrap(_Wrap): @@ -565,7 +580,13 @@ class Wrap(_Wrap): """ def __init__( - self, func, ingress=None, egress=None, *, name=None, preserve_signature="auto" + self, + func, + ingress=None, + egress=None, + *, + name=None, + preserve_signature=AUTO_PRESERVE_SIGNATURE, ): super().__init__(func, ingress, egress, name=name) ingress_sig = Sig(func) @@ -576,7 +597,6 @@ def __init__( func ) else: - if isinstance(ingress, MakeFromFunc): func_to_ingress = ingress # it's not the ingress function itself # ... but an ingress factory: Should make the ingress in function of func @@ -585,15 +605,19 @@ def __init__( assert callable(ingress), f"Should be callable: {ingress}" self.ingress = ingress - # Apply signature preservation if needed - if _should_preserve_signature(self.ingress, func, preserve_signature): - # Preserve signature and annotations from func to ingress - self.ingress.__signature__ = signature(func) - self.ingress.__annotations__ = getattr(func, "__annotations__", {}) - - ingress_sig = Sig(self.ingress) + # Preserving means *this wrapper* presents func's interface. Read it from + # func directly; never write it onto self.ingress. The ingress belongs to + # the caller -- a decorator typically defines one and reuses it for every + # function it wraps -- so stamping __signature__ on it corrupted both the + # caller's object and every later Wrap built from the same ingress. + signature_source = ( + func + if _should_preserve_signature(self.ingress, preserve_signature) + else self.ingress + ) + ingress_sig = Sig(signature_source) self.__defaults__, self.__kwdefaults__ = _defaults_and_kwdefaults_of_func( - self.ingress + signature_source ) # Set egress @@ -974,9 +998,9 @@ def name_map(cls, wrapped, **old_to_new_name): """ new_to_old_name = {v: k for k, v in old_to_new_name.items()} - assert len(new_to_old_name) == len( - old_to_new_name - ), f"Inversion is not possible since {old_to_new_name=} has duplicate values." + assert len(new_to_old_name) == len(old_to_new_name), ( + f"Inversion is not possible since {old_to_new_name=} has duplicate values." + ) return cls( wrapped, partial(Pipe(items_with_mapped_keys, dict), key_mapper=new_to_old_name), @@ -2541,9 +2565,9 @@ def add_smart_defaults( """ names_not_in_func_arguments = smart_defaults.keys() - Sig(func).names - assert ( - not names_not_in_func_arguments - ), f"These weren't argument names of the {func} function: {names_not_in_func_arguments}" + assert not names_not_in_func_arguments, ( + f"These weren't argument names of the {func} function: {names_not_in_func_arguments}" + ) kwargs_trans = partial( complete_dict_applying_functions, _only_if_name_missing=_only_if_name_missing,