diff --git a/changelog.d/receipt-review-lows.fixed.md b/changelog.d/receipt-review-lows.fixed.md new file mode 100644 index 00000000..a60f7542 --- /dev/null +++ b/changelog.d/receipt-review-lows.fixed.md @@ -0,0 +1 @@ +Receipts no longer refuse a measure helper whose quoted `Annotated` metadata contains `|` (for example `List['Annotated[float, "weekly | monthly"]']`), still refuse a union hidden under a name a helper rebinds (`from typing import Tuple as Annotated`), and inspect deeply quoted annotations without a recursion error. A parameter path such as `'gov.hmrc.income_tax.rates.uk[0].threshold'` no longer counts toward the inspection limit, and a lazy loader whose module `__getattr__` raises no longer crashes the receipt. Receipts now refuse a typing generic over a local import of a module not yet loaded (for example `Optional[statistics.NormalDist]`), an attribute of something other than a module inside a typing generic or subscripted with a union (`List[pd.DataFrame.dtypes]`, `HTTPStatus.OK.phrase[0 | 0]`), a function holding a string too deeply nested to parse, and a function whose forward references exceed 100,000 inspected nodes. The `Literal` and `Annotated` exemption is withheld from a function when code anywhere in the module could rebind typing names by a route the receipt can see (`sys.modules`, `setattr` through an alias, a forward reference that calls code or stores to an attribute or subscript). The docs now say which uses of `id` and the other refused names are refused, and which routes the exemption does not follow. diff --git a/docs/receipts.md b/docs/receipts.md index 3e829a3f..e4bc2ec6 100644 --- a/docs/receipts.md +++ b/docs/receipts.md @@ -83,7 +83,7 @@ The accepted module shape also refuses these constructs: Anywhere in the module, code may not: - tell equal values apart by identity: `is` is allowed only against `None`, `True`, `False`, `...` or `type(...)`, and `type` may not be rebound. The compiler may make equal immutable constants one object, and a digest of values cannot tell one object from two equal ones. -- use `id`, `operator.is_` or `is_not`, `globals`, `vars`, `locals`, `exec`, `eval`, `compile`, `__import__` or `breakpoint`, however imported, or import `builtins`; +- use `id`, `operator.is_` or `is_not`, `globals`, `vars`, `locals`, `exec`, `eval`, `compile`, `__import__` or `breakpoint`, however imported, or import `builtins`. Any use of one of these names as a variable is refused: a read, an assignment target or a `del`, even when it is a parameter or local variable (reading a household identifier parameter named `id` is refused; name it `household_id`), because telling a local from the builtin would need scope analysis. Other positions are not variable uses and are allowed: a parameter or function of that spelling that the code never reads, an attribute (`simulation.id`, though `.is_` and `.is_not` are refused) and a keyword argument (`id=3`); - read annotation values through `typing.get_type_hints`, `inspect.signature`, `inspect.getfullargspec`, `get_annotations` or `from_callable`; these explicit metadata readers remain outside the accepted module shape. - declare a name `global`. - read dunder attributes other than `__init__` and `__name__`, or dunder names other than `__name__` and `__file__`. Reading a function's or module's own metadata is introspection. @@ -95,7 +95,7 @@ Put measure values in literal constants or in default arguments. - **Versions.** At receipt build time, any recorded version of policyengine, policyengine-uk or policyengine-core must still match its installed version. Independently, each package's distribution metadata must not have been written after this process started. The builder compares the latest modification or change time of its `.dist-info` directory and of the existing `METADATA`, `RECORD`, `INSTALLER` and `direct_url.json` files against `psutil.Process().create_time()`, and refuses metadata written more than two seconds after it. On Linux psutil builds that start time from the boot time in whole seconds plus the start offset in clock ticks, so it can read up to about a second early (one second plus one tick). Without the margin, a process started just after an install finished would be refused. The margin is a blind window: metadata written within two seconds after the reported start is accepted, so an install that lands in that window is not detected. Past it, metadata is refused regardless of import order or whether a version was recorded; an unavailable metadata location is refused too. Restart the process and run again. This also refuses installing a package and then importing it in the same process, such as a notebook `pip install` without a restart, even when that import loaded the new code, unless the install finishes within two seconds of the process starting. - **Functions.** The source is compiled as a whole, so each function compiles in its module's context, imports included. The modifier and every other function defined in that file must use the module's own globals and match that code, nested code and constants included. Constants are compared by exact type, by IEEE 754 bytes (so a NaN's sign counts) and by how they share objects. Annotation functions that Python 3.14 compiles (`__annotate__`) are the compiler's, not the module's, and a generic function (`def helper[T](...)`, Python 3.12+) is checked through the wrapper the compiler makes to bind its type parameters. - **Function names.** A `def` name must still be bound to that function, not to another function of the same file or any other object. -- **Code inside functions.** It runs after import, so a type it builds is made then: an expression in a function body, or the annotations, type-parameter bounds and class annotations of a helper defined there (quoted or not). typing caches its generic aliases (`List`, `Dict`, `Optional`, `Annotated`) and user generics (`Generic` subclasses, Pydantic models) by argument equality, and a union equals one holding its members in any order. So `List[Union[Decimal, float]]` built there can return an alias cached for `List[Union[float, Decimal]]`, and Pydantic tries a union's members in order. A function whose code subscripts such a generic with a union or `Literal` is refused. Builtin generics, `collections.abc` classes and classes whose `__class_getitem__` is implemented in C (NumPy's `ndarray` and `dtype`) build a new alias in source order and are allowed (`list[Decimal | float]`), as is a bare `Union[Decimal, float]`. Resolution is conservative. A name stands for every object the module, or any import anywhere inside the function, binds it to, so a helper importing `typing.Sequence` is caught even when a sibling imports `collections.abc.Sequence`. Every string in the function's code that parses as an expression is inspected as a type, since a forward reference such as `Optional['List[Union[Decimal, float]]']` is evaluated when something resolves it. `Literal` values and `Annotated` metadata are left as strings. Local aliases of typing objects (`L = List`) are not followed. The module's own function signatures are compared against their source instead (below). +- **Code inside functions.** It runs after import, so a type it builds is made then: an expression in a function body, or the annotations, type-parameter bounds and class annotations of a helper defined there (quoted or not). typing caches its generic aliases (`List`, `Dict`, `Optional`, `Annotated`) and user generics (`Generic` subclasses, Pydantic models) by argument equality, and a union equals one holding its members in any order. So `List[Union[Decimal, float]]` built there can return an alias cached for `List[Union[float, Decimal]]`, and Pydantic tries a union's members in order. A function whose code subscripts such a generic with a union or `Literal` is refused. Builtin generics, `collections.abc` classes and classes whose `__class_getitem__` is implemented in C (NumPy's `ndarray` and `dtype`) build a new alias in source order and are allowed (`list[Decimal | float]`), as is a bare `Union[Decimal, float]`. Resolution is conservative. A name stands for every object the module, or any import anywhere inside the function, binds it to, so a helper importing `typing.Sequence` is caught even when a sibling imports `collections.abc.Sequence`. An import is resolved as Python binds it (`import a.b as c` binds the package's attribute `b`). One whose target is not loaded when the receipt is built, and an attribute of anything but a loaded module or one the module lacks then, could stand for any object: subscripting one with a union is refused, as is a typing generic subscripted with one, and the message says the receipt cannot resolve the name. Every string in the function's code that parses as an expression subscripting something the receipt can name, or holding another string, is inspected as a type, since a forward reference such as `Optional['List[Union[Decimal, float]]']` is evaluated when something resolves it (a label, or a parameter path such as `'gov.hmrc.income_tax.rates.uk[0].threshold'`, subscripts nothing the receipt can name and is skipped); past 100,000 such nodes in one function, or for a string too deeply nested to parse, the function is refused rather than left uninspected. `Literal` values and `Annotated` metadata are left as strings, but only where every object the name can stand for is that form and no code the receipt can see could change what the name, or any attribute in its chain, stands for. Any of these makes the strings under the name inspected as types: an assignment, a parameter, a `:=` target or lambda parameter inside a forward reference, an attribute store or deletion of that name (`typing.Annotated = Tuple`, `del t.Annotated`) or `setattr` or `delattr` with it, in the function or in any other function of the module (a function can call a sibling that does it), or a local import of a name or module path that code stores to. A function gets no such exemption at all when it or any other function of the module uses `modules` as an attribute, an imported name or a string (`sys.modules`, `from sys import modules`, `getattr(sys, 'modules')`), or uses `setattr`, `delattr` or `getattr` other than as a direct call with a literal name (`put = setattr`, `map(setattr, ...)`, a computed name); or when a forward reference in the function that calls something or stores to an attribute or subscript (and so runs code: `[0 for t.__dict__['Annotated'] in ...]` runs `__setitem__`) does either, or names one of those builtins, `import_module`, a refused name or a dunder. A string that does neither, such as a label `'id'` or `'__main__'`, runs no code; apart from the string `'modules'` and the inspection limit, it never costs the exemption, though a store or binding in it still counts. Like the rest of the receipt, this guards an honest producer; it is not a sandbox. Some routes are not followed: local aliases of typing objects (`L = List`, or `Amount = Union[Decimal, float]` and then `List[Amount]`), `getattr(typing, 'List')`, `importlib.import_module('typing')` results, `operator.getitem` or `reduce` over typing objects, what a call inside a forward reference does beyond the names above, forward references in other functions of the module, forward references built at run time (f-strings, concatenation), and changes made from other modules. A function using them can build such an alias unnoticed, so use a builtin generic there. The module's own function signatures are compared against their source instead (below). - **Function metadata.** Each function's name, qualified name and docstring must still be what its `def` made, compared as the compiler stores them. A generic function's `__type_params__` must still be bare type parameters of the names and kinds its source declares, in order, with no bound, constraints, default or declared variance. Each annotation must still hold what its source names, resolved against the module's globals and builtins (and those type parameters): named objects are compared by identity; reconstructed generics, unions and `ParamSpec` argument proxies must have the same origin and arguments, recursively and in order. Source arguments are resolved separately so typing's cache cannot hide a reordered union. Parameterised typing aliases are refused when their source order cannot be checked independently of that cache. Subscripted annotations that do not expose a typing origin and arguments, such as generated Pydantic model classes, are also refused. Quoted annotations and strings stored by `from __future__ import annotations` must match their source, and their resolvable type expressions must also match source order. This includes annotations consumed implicitly by libraries such as `functools.singledispatch`. It must use the builtins module and carry no other attributes (`helper.amount = 240` is state). - **Defaults.** Literal default arguments in the source, such as `rate=0.21`, must still be the function's defaults, with the same type. - **Literal constants.** Every literal constant must still hold its value, compared exactly: the same types throughout, floats by their bits, dictionary keys in order, for every literal kind including bytes, sets and complex numbers. diff --git a/src/policyengine/provenance/receipt.py b/src/policyengine/provenance/receipt.py index 248c1e0f..c1a6c525 100644 --- a/src/policyengine/provenance/receipt.py +++ b/src/policyengine/provenance/receipt.py @@ -81,7 +81,7 @@ import struct import sys import zipfile -from collections.abc import Callable, Iterator, Mapping, Sequence +from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence from importlib import metadata from importlib.metadata import PackageNotFoundError from pathlib import Path @@ -1263,7 +1263,12 @@ def _unwrapped(value: Any, module: Any, filename: str) -> Any: for _ in range(16): if _from_this_file(value, module, filename): return value - value = getattr(value, "__wrapped__", None) + try: + value = getattr(value, "__wrapped__", None) + except Exception: + # A module whose lazy __getattr__ raises something other than + # AttributeError wraps nothing (#632 round 6, L1). + return None if value is None: return None return None @@ -2269,20 +2274,193 @@ def _holds_union(value: Any, depth: int = 0) -> bool: return any(_holds_union(argument, depth + 1) for argument in arguments) +_TYPE_PARAMETERS = tuple( + getattr(ast, name) + for name in ("TypeVar", "ParamSpec", "TypeVarTuple") + if hasattr(ast, name) +) + + +def _names_bound_without_import(nodes: Iterable[ast.AST]) -> set[str]: + """Every name these nodes bind other than by import: parameters (lambda + ones included), assignment, loop, ``with``, comprehension and ``:=`` + targets, deletions, ``except`` and ``match`` captures, nested function + and class names, type parameters, and ``global`` or ``nonlocal`` + declarations.""" + + names: set[str] = set() + for node in nodes: + if isinstance(node, ast.Name) and not isinstance(node.ctx, ast.Load): + names.add(node.id) + elif isinstance(node, ast.arg): + names.add(node.arg) + elif isinstance( + node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef) + ) or isinstance(node, _TYPE_PARAMETERS): + names.add(node.name) + elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)): + if node.name: + names.add(node.name) + elif isinstance(node, ast.MatchMapping) and node.rest: + names.add(node.rest) + elif isinstance(node, (ast.Global, ast.Nonlocal)): + names.update(node.names) + return names + + +#: Builtins that read or write an attribute named at run time. +_ATTRIBUTE_BUILTINS = ("setattr", "delattr", "getattr") + + +def _names_mutated_by_attribute(nodes: Iterable[ast.AST]) -> tuple[set[str], bool]: + """Names code can rebind without binding a name, and whether it can + rebind ones the receipt cannot name. + + The names: attribute stores and deletions (``sys.modules[__name__]. + Annotated = Tuple`` and ``typing.Annotated = Tuple`` store + ``Annotated``) and the literal names passed to ``setattr`` or + ``delattr`` called directly. Unsound (any name could change): any use of + ``modules`` (an attribute, an imported name or a string), through + which code can swap whole modules; ``getattr(x, "modules")``; and + ``setattr``, ``delattr`` or ``getattr`` used other than as a direct call + with a literal name (an alias, ``map``, ``partial``, a computed name, or + reached as an attribute). + """ + + nodes = list(nodes) + names: set[str] = set() + unsound = False + direct: set[int] = set() + for node in nodes: + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in _ATTRIBUTE_BUILTINS + and len(node.args) >= 2 + and isinstance(node.args[1], ast.Constant) + and isinstance(node.args[1].value, str) + ): + direct.add(id(node.func)) + if node.func.id != "getattr": + names.add(node.args[1].value) + elif node.args[1].value == "modules": + unsound = True + for node in nodes: + if isinstance(node, ast.Attribute): + if not isinstance(node.ctx, ast.Load): + names.add(node.attr) + if node.attr == "modules" or node.attr in _ATTRIBUTE_BUILTINS: + unsound = True + elif isinstance(node, ast.Name): + if node.id in _ATTRIBUTE_BUILTINS and id(node) not in direct: + unsound = True + elif isinstance(node, ast.alias) and node.name in ( + "modules", + *_ATTRIBUTE_BUILTINS, + ): + unsound = True + elif isinstance(node, ast.Constant) and node.value == "modules": + unsound = True + return names, unsound + + +#: Names a forward reference must not use: Pydantic evaluates it, so it runs +#: whatever it calls (``setattr(...) or int``, ``globals().update(...)``). +_CODE_RUNNING_NAMES = frozenset({*_ATTRIBUTE_BUILTINS, "import_module"}) | ( + _REFUSED_NAMES +) + + +def _runs_code(node: ast.AST) -> bool: + """Whether a node in a forward reference names something that could + change what other names stand for.""" + + if isinstance(node, ast.Name): + return node.id in _CODE_RUNNING_NAMES or _is_dunder(node.id) + if isinstance(node, ast.Attribute): + return node.attr in _CODE_RUNNING_NAMES or _is_dunder(node.attr) + return False + + +def _module_mutations(tree: ast.Module) -> tuple[set[str], bool]: + """``_names_mutated_by_attribute`` across every top-level function: a + function can call a sibling that rebinds a name for it.""" + + nodes: list[ast.AST] = [] + for statement in tree.body: + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + nodes.extend(ast.walk(statement.args)) + for root in statement.body: + nodes.extend(ast.walk(root)) + return _names_mutated_by_attribute(nodes) + + +def _imports_of_stored_names(nodes: Iterable[ast.AST], stored: set[str]) -> set[str]: + """Names local imports bind from a module path, or as a name, that code + stores to: ``typing.Annotated = Tuple`` then ``from typing import + Annotated as A`` rebinds ``A``.""" + + rebound: set[str] = set() + for node in nodes: + if isinstance(node, ast.Import): + for alias in node.names: + if stored.intersection(alias.name.split(".")): + rebound.add(alias.asname or alias.name.split(".")[0]) + elif isinstance(node, ast.ImportFrom): + path = set((node.module or "").split(".")) + for alias in node.names: + if stored.intersection(path | {alias.name}): + rebound.add(alias.asname or alias.name) + return rebound + + +#: How many nodes the forward references of one function may add before +#: the receipt stops inspecting them and refuses the function instead. +_INSPECTED_FORWARD_NODES = 100_000 + + +class _Unresolved: + """What a name stands for when the receipt cannot tell: an import whose + target is not loaded when the receipt is built, or an attribute of + something other than a loaded module. It could be any object.""" + + __slots__ = () + + def __repr__(self) -> str: + return "" + + +_UNRESOLVED = _Unresolved() + + def _function_bindings(module: Any, nodes: Sequence[ast.AST]) -> dict[str, list[Any]]: """What each name an import inside a function binds, across all its scopes. Every binding is kept, so a name one helper imports from typing and - another from collections.abc counts as both. + another from collections.abc counts as both. An import whose target is + not loaded yet binds :data:`_UNRESOLVED`: the import runs when the + function does, and could bind anything. """ bindings: dict[str, list[Any]] = {} for node in nodes: if isinstance(node, ast.Import): for alias in node.names: - name = alias.asname or alias.name.split(".")[0] - target = alias.name if alias.asname else name - bindings.setdefault(name, []).append(sys.modules.get(target)) + if not alias.asname: + name = alias.name.split(".")[0] + bindings.setdefault(name, []).append( + sys.modules.get(name, _UNRESOLVED) + ) + continue + # ``import a.b as c`` binds getattr(a, "b"), as Python does, + # with sys.modules only where the attribute is missing. + parts = alias.name.split(".") + value: Any = sys.modules.get(parts[0], _UNRESOLVED) + for depth, part in enumerate(parts[1:], start=2): + if value is _UNRESOLVED: + break + value = _module_attribute(value, part, ".".join(parts[:depth])) + bindings.setdefault(alias.asname, []).append(value) elif isinstance(node, ast.ImportFrom): try: source_name = ( @@ -2293,20 +2471,41 @@ def _function_bindings(module: Any, nodes: Sequence[ast.AST]) -> dict[str, list[ else node.module or "" ) except (ImportError, ValueError): - continue - source = sys.modules.get(source_name) + source_name = None + source = sys.modules.get(source_name) if source_name else None for alias in node.names: - if alias.name != "*": - bindings.setdefault(alias.asname or alias.name, []).append( - getattr( - source, - alias.name, - sys.modules.get(f"{source_name}.{alias.name}"), - ) + if alias.name == "*": + continue + if source is None: + value = ( + sys.modules.get(f"{source_name}.{alias.name}", _UNRESOLVED) + if source_name + else _UNRESOLVED ) + else: + value = _module_attribute( + source, alias.name, f"{source_name}.{alias.name}" + ) + bindings.setdefault(alias.asname or alias.name, []).append(value) return bindings +def _module_attribute(owner: Any, name: str, submodule: str = "") -> Any: + """What ``owner.name`` stands for, read as Python reads it: the module's + attribute (a lazy loader's ``__getattr__`` runs, as it will when the code + does), then the loaded submodule of that name. Anything but a module, or a + loader that raises, gives :data:`_UNRESOLVED` rather than an error.""" + + if not isinstance(owner, ModuleType): + return _UNRESOLVED + try: + return getattr(owner, name) + except AttributeError: + return sys.modules.get(submodule, _UNRESOLVED) if submodule else _UNRESOLVED + except Exception: + return _UNRESOLVED + + def _runtime_alias_problem(module: Any, tree: ast.Module) -> Optional[str]: """Where a module function's code builds a type typing may reuse reordered. @@ -2321,51 +2520,174 @@ def _runtime_alias_problem(module: Any, tree: ast.Module) -> Optional[str]: inside them. Resolution is conservative: a name stands for every object the module - or any import inside the function binds it to, and every string in the - function's code that parses as an expression is inspected as a type - (except ``Literal`` values and ``Annotated`` metadata), since a forward - reference is evaluated when something resolves it. Local aliases of - typing objects (``L = List``) are not followed. + or any import inside the function binds it to (an import not loaded yet, + or an attribute of anything but a loaded module, could be any object), + and every string in the function's code that parses as an expression + and could build a cached type (it subscripts something the receipt can + name, or holds a nested string) is inspected as a type, except + ``Literal`` values and ``Annotated`` metadata, since a forward reference + is evaluated when something resolves it. A function whose forward + references exceed ``_INSPECTED_FORWARD_NODES``, or hold a string too + deeply nested to parse, is refused. + + The ``Literal`` and ``Annotated`` exemptions are granted only where no + code the receipt can see could change what those names stand for: no + binding of the name in the function, no store or ``setattr``/``delattr`` + of it in the code of any module function or in a forward reference of + the function, no use of ``modules`` (an attribute, an imported name or a + string) or of ``setattr``, ``delattr`` or ``getattr`` other than a direct + call with a literal name in any module function, and no forward + reference of the function that calls something, or stores to an + attribute or subscript, while doing either or naming such a builtin, + ``import_module``, a refused name or a dunder. A string that does + neither (a label such as ``'id'``) runs no code. + + This guards an honest producer, as the receipt does throughout. It is not + a sandbox: it does not follow local aliases (``L = List``, ``Amount = + Union[Decimal, float]``), ``getattr(typing, "List")``, + ``importlib.import_module`` results, ``operator.getitem`` or ``reduce`` + over typing objects, what calls inside forward references do beyond the + names above, forward references in other module functions, f-string or + concatenated forward references, or changes made from other modules. """ namespace = vars(module) + # A sibling function the code calls can rebind names for it. + module_stored, module_unsound = _module_mutations(tree) for statement in tree.body: if not isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): continue nodes = [node for root in statement.body for node in ast.walk(root)] local = _function_bindings(module, nodes) + # Names the code rebinds other than by import, in any scope and in + # any forward reference (through := or a lambda's parameters), are + # not resolved, so they never earn the Literal or Annotated + # exemption. Past the size limit nothing does. + rebound = _names_bound_without_import([*ast.walk(statement.args), *nodes]) + # Names an attribute store or setattr can rebind, here or in any + # module function, count as rebound wherever they appear in a name's + # chain, as do local imports of them; code that can rebind names the + # receipt cannot see earns no exemption at all. + stored, unsound = _names_mutated_by_attribute(nodes) + stored |= module_stored + rebound |= stored | _imports_of_stored_names(nodes, stored) + exemptions_sound = not (unsound or module_unsound) + strings = [ + node + for node in nodes + if isinstance(node, ast.Constant) and isinstance(node.value, str) + ] + parsed_nodes = 0 + while strings and exemptions_sound: + text = strings.pop().value + try: + walked = list(ast.walk(ast.parse(text.strip(), mode="eval"))) + except (SyntaxError, ValueError): + continue + except RecursionError: + exemptions_sound = False + break + # Every string is inspected; only those that could bind a name + # or build a type count toward the limit, so labels never do. + if any( + isinstance(child, (ast.Subscript, ast.NamedExpr, ast.Lambda)) + or (isinstance(child, ast.Constant) and isinstance(child.value, str)) + for child in walked + ): + parsed_nodes += len(walked) + stored, unsound = _names_mutated_by_attribute(walked) + # A string that neither calls anything nor stores to an attribute + # or subscript (a label such as 'id' or '__main__') cannot run + # code; stores and bindings in it still count. A comprehension + # target such as t.__dict__['Annotated'] runs __setitem__ with no + # call (round 7 of #632), so such a store counts as a call. + calls = any( + isinstance(child, ast.Call) + or ( + isinstance(child, (ast.Attribute, ast.Subscript)) + and not isinstance(child.ctx, ast.Load) + ) + for child in walked + ) + if parsed_nodes > _INSPECTED_FORWARD_NODES or ( + calls and (unsound or any(_runs_code(child) for child in walked)) + ): + exemptions_sound = False + break + rebound |= _names_bound_without_import(walked) | stored + strings.extend( + node + for node in walked + if isinstance(node, ast.Constant) and isinstance(node.value, str) + ) def resolve(node: ast.AST) -> list[Any]: - if isinstance(node, ast.Name): - values = list(local.get(node.id, ())) - if node.id in namespace: - values.append(namespace[node.id]) - # A builtin (list, dict, type) never reuses equal aliases. - return values - if isinstance(node, ast.Attribute): - return [ - getattr(owner, node.attr) - for owner in resolve(node.value) - if isinstance(owner, ModuleType) and hasattr(owner, node.attr) - ] - return [] + # Iterative: a quoted forward reference can nest attributes deeper + # than the interpreter's recursion limit. + attributes: list[str] = [] + while isinstance(node, ast.Attribute): + attributes.append(node.attr) + node = node.value + if not isinstance(node, ast.Name): + return [] + values = list(local.get(node.id, ())) + if node.id in namespace: + values.append(namespace[node.id]) + # A builtin (list, dict, type) never reuses equal aliases. An + # attribute is followed only on a loaded module; on anything else, + # or one the module lacks now, it could be any object. + for attribute in reversed(attributes): + values = [_module_attribute(owner, attribute) for owner in values] + return values def is_form(node: ast.AST, *forms: str) -> bool: - return any(_is_typing_form(value, *forms) for value in resolve(node)) + """Whether any object the name can stand for is one of ``forms`` + (an unresolved one could be).""" + + return any( + value is _UNRESOLVED or _is_typing_form(value, *forms) + for value in resolve(node) + ) + + def is_only_form(node: ast.AST, *forms: str) -> bool: + """Whether every object the name can stand for is one of ``forms``. + + Exemptions need this: a name a helper rebinds (``from typing + import Tuple as Annotated``, ``Annotated = Tuple``, a parameter + named ``Annotated``) must not exempt the strings under it. + """ + + root = node + chain: list[str] = [] + while isinstance(root, ast.Attribute): + chain.append(root.attr) + root = root.value + if ( + not exemptions_sound + or not isinstance(root, ast.Name) + or root.id in rebound + or rebound.intersection(chain) + ): + return False + values = resolve(node) + return bool(values) and all( + _is_typing_form(value, *forms) for value in values + ) # Strings that are values, not forward references. ast.walk yields a # node before its children, so a Literal or Annotated subscript marks # its values before they are reached. values: set[int] = set() inspected: list[ast.AST] = list(nodes) + added = 0 position = 0 while position < len(inspected): node = inspected[position] position += 1 if isinstance(node, ast.Subscript): - if is_form(node.value, "Literal"): + if is_only_form(node.value, "Literal"): values.update(id(child) for child in ast.walk(node.slice)) - elif is_form(node.value, "Annotated") and isinstance( + elif is_only_form(node.value, "Annotated") and isinstance( node.slice, ast.Tuple ): for metadata in node.slice.elts[1:]: @@ -2374,54 +2696,124 @@ def is_form(node: ast.AST, *forms: str) -> bool: isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in values - and len(inspected) < 100_000 ): try: parsed = ast.parse(node.value.strip(), mode="eval").body except (SyntaxError, ValueError): continue - for child in ast.walk(parsed): + except RecursionError: + return ( + f"function {statement.name} holds a string too deeply " + "nested for the receipt to inspect as a type" + ) + walked = list(ast.walk(parsed)) + # Only a subscript of something the receipt can name, or a + # nested string that may hold one, can build a type typing + # caches; a label, or a subscript of an unbound name such as + # a parameter path ('gov.hmrc...rates.uk[0].threshold'), + # cannot, and is neither inspected nor counted. + if not any( + (isinstance(child, ast.Subscript) and resolve(child.value)) + or ( + isinstance(child, ast.Constant) and isinstance(child.value, str) + ) + for child in walked + ): + continue + added += len(walked) + if added > _INSPECTED_FORWARD_NODES: + return ( + f"function {statement.name} holds more forward-reference " + "type expressions than the receipt inspects" + ) + for child in walked: ast.copy_location(child, node) inspected.append(child) - def holds_union(expression: ast.AST) -> bool: - for child in ast.walk(expression): - if isinstance(child, ast.BinOp) and isinstance(child.op, ast.BitOr): - return True - if isinstance(child, ast.Subscript) and is_form( - child.value, "Union", "Optional", "Literal" - ): - return True - if isinstance(child, (ast.Name, ast.Attribute)) and any( - _holds_union(value) for value in resolve(child) - ): - return True + def holds_union(expression: ast.AST) -> Optional[str]: + """What in a type expression typing may reuse reordered: "a union" + (a union or ``Literal``), or "a name the receipt cannot resolve" + (which could be either); ``None`` if nothing. + + ``Literal`` values and ``Annotated`` metadata are values, not types, + so a quoted string there is never read as a type, including inside + a forward reference parsed here, when every object the subscripted + name can stand for is that form. The walk is iterative: quoted + layers can nest deeper than the interpreter's recursion limit. + """ + + pending = [expression] + while pending: + node = pending.pop() + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + return "a union" + if isinstance(node, ast.Subscript): + if is_form(node.value, "Union", "Optional", "Literal"): + return ( + "a union" + if any( + _is_typing_form(value, "Union", "Optional", "Literal") + for value in resolve(node.value) + ) + else "a name the receipt cannot resolve" + ) + if is_only_form(node.value, "Annotated"): + elements = ( + node.slice.elts + if isinstance(node.slice, ast.Tuple) + else [node.slice] + ) + pending.extend(elements[:1]) + continue if ( - isinstance(child, ast.Constant) - and isinstance(child.value, str) - and id(child) not in values + isinstance(node, (ast.Name, ast.Attribute)) + and id(node) not in values ): - try: - parsed = ast.parse(child.value.strip(), mode="eval").body - except (SyntaxError, ValueError): - continue - if holds_union(parsed): - return True - return False + resolved = resolve(node) + if any( + _holds_union(value) + for value in resolved + if value is not _UNRESOLVED + ): + return "a union" + if _UNRESOLVED in resolved: + return "a name the receipt cannot resolve" + if isinstance(node, ast.Constant): + if isinstance(node.value, str) and id(node) not in values: + try: + pending.append( + ast.parse(node.value.strip(), mode="eval").body + ) + except (SyntaxError, ValueError): + pass + except RecursionError: + return "a string too deeply nested to inspect" + continue + pending.extend(ast.iter_child_nodes(node)) + return None for node in inspected: if not isinstance(node, ast.Subscript): continue - if any( - _subscript_reuses_equal_aliases(value) for value in resolve(node.value) - ) and holds_union(node.slice): + if not any( + value is _UNRESOLVED or _subscript_reuses_equal_aliases(value) + for value in resolve(node.value) + ): + continue + held = holds_union(node.slice) + if held is not None: line = getattr(node, "lineno", statement.lineno) + try: + base = ast.unparse(node.value) + except RecursionError: + base = "a deeply nested expression" return ( f"function {statement.name} subscripts " - f"{ast.unparse(node.value)} with a union (line {line}); typing " + f"{base} with {held} (line {line}); typing " "can return an alias it cached for the same members in another " "order, so earlier code decides what runs (use a builtin " - "generic such as list[Decimal | float])" + "generic such as list[Decimal | float], and import what the " + "function uses at module level)" ) return None diff --git a/tests/test_execution_receipt.py b/tests/test_execution_receipt.py index 8625dcdb..33e9b5e5 100644 --- a/tests/test_execution_receipt.py +++ b/tests/test_execution_receipt.py @@ -29,6 +29,7 @@ from __future__ import annotations +import ast import copy import datetime import decimal @@ -43,10 +44,11 @@ import re import subprocess import sys +import typing from collections.abc import Mapping from importlib import metadata from pathlib import Path -from types import CodeType, SimpleNamespace +from types import CodeType, ModuleType, SimpleNamespace from typing import Union, get_args from urllib.parse import urlparse @@ -2048,6 +2050,41 @@ def test_builder_rechecks_each_composition_member_after_the_run( sys.modules.pop(name, None) +@pytest.mark.parametrize("stale_first", [True, False]) +def test_every_composition_member_is_rechecked_wherever_it_sits( + dataset_file, tmp_path, stale_first +): + """#579 round-5 review: a stale member is caught in any position.""" + + name = "w4_receipt_compose_position" + try: + module, _ = _import_source(tmp_path, name, MEASURE_SOURCE.format(rate=0.21)) + live = _measure_identity(module) + members = [live, _member(7)] if stale_first else [_member(7), live] + identity = composed_reform_identity(members, applied_before_data_load=False) + module.RATE = 0.30 + with pytest.raises(ReceiptError, match="modifier changed after its identity"): + _build(dataset_file, reform=identity, baseline=CURRENT_LAW) + finally: + sys.modules.pop(name, None) + + +@pytest.mark.parametrize("shape", ["alone", "mixed", "nested"]) +def test_a_current_law_member_that_gains_changes_is_caught(dataset_file, shape): + """#579 round-5 review: a member dropped as current law is still rechecked.""" + + scenario = _scenario() + member = scenario_reform_identity(scenario, code_revision=REVISION) + assert member.kind == "current_law" + members = [member] if shape == "alone" else [member, _member(7)] + identity = composed_reform_identity(members, applied_before_data_load=False) + if shape == "nested": + identity = composed_reform_identity([identity], applied_before_data_load=False) + scenario.parameter_changes = {"gov.hmrc.income_tax.rates.uk[0].rate": 0.21} + with pytest.raises(ReceiptError, match="changed after its identity was computed"): + _build(dataset_file, reform=identity, baseline=CURRENT_LAW) + + def test_composing_refuses_a_member_that_changed_since_it_was_bound(tmp_path): name = "w4_receipt_compose_stale_member" try: @@ -2075,6 +2112,18 @@ def test_an_unchanged_composition_still_builds_and_ignores_later_list_edits( assert loaded == identity and loaded._recompute_identity is None +def test_a_certificate_for_other_bytes_gives_an_uncertified_receipt(dataset_file): + """Round-16 review: a certificate whose dataset hash differs from the bytes + run is not an error, just not a certificate for this run.""" + + bundle = _bundle_for(dataset_file.read_bytes()) + bundle["data_releases"]["uk"]["certified_data_artifact"]["sha256"] = "0" * 64 + receipt = _build(dataset_file, bundle=bundle) + assert receipt.certification_status == "uncertified" + assert receipt.certificate is None + assert receipt.data.dataset_sha256 != "0" * 64 + + def test_unchanged_scenario_still_builds_after_a_run(dataset_file): scenario = _scenario(simulation_modifier=receipt_modifiers.raise_basic_rate_to) identity = scenario_reform_identity(scenario, code_revision=REVISION) @@ -3755,6 +3804,62 @@ def test_aliases_of_refused_names_are_refused(tmp_path, header, body, match): sys.modules.pop(name, None) +@pytest.mark.parametrize( + ("body", "refused"), + [ + pytest.param( + " def helper(id):\n return id\n\n" + " simulation.rate = helper(0.21)\n", + True, + id="read-parameter", + ), + pytest.param( + " for id in [0.21]:\n simulation.rate = 0.21\n", + True, + id="assignment-target", + ), + pytest.param( + " def helper(id):\n return 0.21\n\n" + " simulation.rate = helper(1)\n", + False, + id="unread-parameter", + ), + pytest.param( + " def id():\n return 0.21\n\n simulation.rate = 0.21\n", + False, + id="unread-function-name", + ), + pytest.param( + " simulation.id = 3\n simulation.rate = 0.21\n", + False, + id="attribute", + ), + pytest.param( + " def helper(**amounts):\n return 0.21\n\n" + " simulation.rate = helper(id=3)\n", + False, + id="keyword-argument", + ), + ], +) +def test_refused_names_are_refused_where_code_uses_them_as_variables( + tmp_path, body, refused +): + """Round-17 review P3: the guard reads names, as docs/receipts.md says.""" + + source = "def raise_rate(simulation):\n" + body + name = "w4_receipt_r17_names" + try: + module, _ = _import_source(tmp_path, name, source) + if refused: + with pytest.raises(ReceiptError, match="uses id"): + _identity(module.raise_rate) + else: + assert _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + def test_an_import_rebound_at_run_time_is_refused(tmp_path): """Round-7 review M4: the name must still be what its import binds.""" @@ -5018,6 +5123,59 @@ def test_a_quoted_special_form_with_a_flipped_flag_is_refused(tmp_path, quoted): sys.modules.pop(name, None) +def test_a_flag_poisoned_before_anything_binds_is_refused(tmp_path): + """Round-19 review: in a fresh process, poison typing's cached + ``Final['_receipt_forward_probe']`` before any binding, then bind a module + whose annotation reuses it. The interpreter's own flag still decides.""" + + name = "w4_receipt_r19_fresh_poison" + path = tmp_path / f"{name}.py" + path.write_text( + "from typing import Final\n\n\n" + "def _helper(value: Final['_receipt_forward_probe']):\n return value\n\n\n" + "def raise_rate(simulation):\n simulation.rate = 0.21\n" + ) + script = ( + "import importlib.util, sys\n" + "from typing import Final, get_args\n" + # Import first: importing policyengine builds enough typing aliases to + # evict a poisoned entry from typing's cache before the module reuses it. + "from policyengine.provenance.receipt import ReceiptError, _modifier_identity\n" + "alias = Final['_receipt_forward_probe']\n" + "(ref,) = get_args(alias)\n" + "ref.__forward_is_class__ = not ref.__forward_is_class__\n" + f"spec = importlib.util.spec_from_file_location({name!r}, {str(path)!r})\n" + "module = importlib.util.module_from_spec(spec)\n" + f"sys.modules[{name!r}] = module\n" + "spec.loader.exec_module(module)\n" + "assert module._helper.__annotations__['value'] is alias, 'not poisoned'\n" + "try:\n" + " _modifier_identity(module.raise_rate, REVISION)\n" + "except ReceiptError as error:\n" + " print('refused:', error)\n" + "else:\n" + " print('bound')\n" + ).replace("REVISION", repr(dict(REVISION))) + result = subprocess.run( + [sys.executable, "-c", script], + env={ + **os.environ, + "POLICYENGINE_SKIP_COUNTRY_IMPORTS": "1", + # This process's import path, absolute, so the child imports the + # same policyengine whatever its working directory. + "PYTHONPATH": os.pathsep.join( + str(Path(entry).resolve()) for entry in sys.path if entry + ), + }, + capture_output=True, + text=True, + check=True, + cwd=tmp_path, + ) + assert result.stdout.startswith("refused:"), result.stdout + result.stderr + assert "annotation for value" in result.stdout + + def test_subscripted_annotations_without_typing_origins_are_refused( tmp_path, monkeypatch ): @@ -5514,6 +5672,17 @@ def test_generic_type_parameters_must_be_those_the_def_made(tmp_path, change, ma sys.modules.pop(name, None) +def _deeply_quoted(leaf: str, depth: int = 80, layers: int = 4) -> str: + """``leaf`` under ``layers`` quoted forward references, each nesting it in + ``depth`` builtin tuples: within the parser's limit per string, and deeper + than a recursive walk of the whole expression can go.""" + + expression = leaf + for _ in range(layers): + expression = repr("tuple[int, " * depth + expression + "]" * depth) + return expression + + RUNTIME_ALIAS_HEAD = ( "import collections.abc\nimport typing\nfrom decimal import Decimal\n" "from typing import Annotated, Dict, List, Literal, Optional, Union\n\n\n" @@ -5584,6 +5753,13 @@ def test_generic_type_parameters_must_be_those_the_def_made(tmp_path, change, ma "List", id="quoted-class-body-annotation", ), + pytest.param( + # Round-16 review: Optional is a union too. + " shape = List[Optional[Decimal]]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="list-of-an-optional", + ), pytest.param( " shape = Union[Union[Decimal, float], None]\n" " simulation.rate = 0.21 if shape else 0.0\n", @@ -5663,6 +5839,380 @@ def test_generic_type_parameters_must_be_those_the_def_made(tmp_path, change, ma "RootModel", id="pydantic-generic-model", ), + pytest.param( + # Round-17 review P2: a helper rebinds Annotated to Tuple, so the + # quoted "metadata" may be a type argument. + " from typing import Tuple as Annotated\n\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="annotated-rebound-to-tuple", + ), + pytest.param( + " from typing import Tuple as Annotated\n\n" + " shape = List[Annotated[float, 'List[Union[Decimal, float]]']]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="unquoted-annotated-rebound-to-tuple", + ), + pytest.param( + # Round-18 review P2: an assignment rebinds the name too. + " Annotated = typing.Tuple\n\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="annotated-assigned-a-tuple", + ), + pytest.param( + " def helper(Annotated):\n" + " return List[Annotated[float, 'List[Union[Decimal, float]]']]\n\n" + " simulation.rate = 0.21 if helper(typing.Tuple) else 0.0\n", + "List", + id="nested-parameter-named-annotated", + ), + pytest.param( + " Literal = typing.Tuple\n\n" + " shape = Literal['List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "Literal", + id="literal-assigned-a-tuple", + ), + pytest.param( + # A forward reference evaluated without locals binds := targets + # in the module, so a later reference reads the new object. + " first = List['(Annotated := typing.Tuple)']\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="walrus-in-a-forward-reference", + ), + pytest.param( + " shape = List['(lambda Annotated: " + 'Annotated[float, "List[Union[Decimal, float]]"])(typing.Tuple)\']\n' + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="lambda-parameter-in-a-forward-reference", + ), + pytest.param( + # Round-19 review: an import not loaded when the receipt is built + # binds an unknown object, here in place of typing. + " import w4_receipt_r19_not_loaded as typing\n\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="annotated-through-an-unloaded-import", + ), + pytest.param( + " import w4_receipt_r19_not_loaded as aliases\n\n" + " shape = aliases.Sequence[Union[Decimal, float]]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "aliases.Sequence", + id="subscript-of-an-unloaded-import", + ), + pytest.param( + " from w4_receipt_r19_not_loaded import MONEY\n\n" + " shape = List[MONEY]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="list-of-a-name-from-an-unloaded-module", + ), + pytest.param( + " import w4_receipt_r19_not_loaded as MONEY\n\n" + " shape = List[MONEY]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="list-of-an-unloaded-module", + ), + pytest.param( + " from .w4_receipt_r19_not_loaded import MONEY\n\n" + " shape = List[MONEY]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="list-of-a-name-from-an-unresolvable-relative-import", + ), + # Round 20 (#632 round 4): code that changes what a name resolves to + # without binding it earns no exemption. + pytest.param( + " import sys\n\n" + " sys.modules[__name__].Annotated = typing.Tuple\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="module-global-rebound-through-sys-modules", + ), + pytest.param( + " setattr(typing, 'Annotated', typing.Tuple)\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="attribute-rebound-by-setattr", + ), + pytest.param( + " typing.Annotated = typing.Tuple\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="typing-attribute-stored", + ), + pytest.param( + " import sys\n import typing_extensions as te\n\n" + " sys.modules['typing_extensions'] = typing\n" + " shape = List['te.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="module-swapped-in-sys-modules", + ), + pytest.param( + # The root of an attribute chain rebound locally. + " from types import SimpleNamespace\n\n" + " typing = SimpleNamespace(Annotated=Union)\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="attribute-chain-root-rebound", + ), + pytest.param( + " import w4_receipt_r19_not_loaded as typing\n\n" + " shape = typing.Annotated[float, 'List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "typing.Annotated", + id="annotated-whose-only-binding-is-unresolved", + ), + pytest.param( + " class Annotated:\n pass\n\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="nested-class-named-annotated", + ), + pytest.param( + " match typing.Tuple:\n" + " case Annotated:\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="match-capture-named-annotated", + ), + pytest.param( + " try:\n shape = None\n" + " except Exception as Annotated:\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="except-capture-named-annotated", + ), + pytest.param( + # A := inside a string inside a string. + " first = List['\"(Annotated := typing.Tuple)\"']\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="walrus-in-a-nested-forward-reference", + ), + pytest.param( + " from w4_receipt_r15_aliases import Box\n\n" + " shape = Box.Sequence[Union[Decimal, float]]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "Box.Sequence", + id="attribute-of-an-imported-class", + ), + pytest.param( + " from typing import Tuple as Literal\n\n" + " shape = Literal['List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "Literal", + id="literal-rebound-to-tuple", + ), + pytest.param( + # Round-17 review P3: quoted layers nest deeper than the + # interpreter's recursion limit allows a recursive walk. + " def helper(value: List[" + + _deeply_quoted("Union[Decimal, float]") + + "]):\n return value\n\n" + " simulation.rate = 0.21\n", + "List", + id="union-under-deeply-quoted-layers", + ), + # Round 21 (#632 round 5): routes that rebind a name the receipt can + # see without binding it in the function. + pytest.param( + " typing.Annotated = typing.Tuple\n" + " from typing import Annotated as A\n\n" + " shape = List['A[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="import-renaming-a-stored-name", + ), + pytest.param( + " from sys import modules\n\n" + " modules['typing_extensions'] = typing\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="modules-imported-by-name", + ), + pytest.param( + " import sys\n\n" + " found = getattr(sys, 'modules')\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape and found else 0.0\n", + "List", + id="modules-read-by-getattr", + ), + pytest.param( + " put = setattr\n" + " put(typing, 'Annotated', typing.Tuple)\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="setattr-through-an-alias", + ), + pytest.param( + " list(map(setattr, [typing], ['Annotated'], [typing.Tuple]))\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="setattr-through-map", + ), + pytest.param( + ' first = \'setattr(typing, "Annot" + "ated", typing.Tuple) or int\'\n' + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-calling-setattr", + ), + pytest.param( + " first = 'globals().update(Annotated=typing.Tuple) or int'\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-calling-globals", + ), + pytest.param( + " first = 'sys.modules.update(typing_extensions=typing) or int'\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-using-modules", + ), + pytest.param( + " del typing.Annotated\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="attribute-deleted", + ), + pytest.param( + " delattr(typing, 'Annotated')\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="attribute-deleted-by-delattr", + ), + pytest.param( + # A string that holds only a nested string is inspected. + " shape = \"'List[Union[Decimal, float]]'\"\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="union-in-a-string-inside-a-string", + ), + pytest.param( + # Past the limit in the scan for bindings, no exemption is + # granted: 25,001 walrus forward references of 4 nodes each. + " names = [" + ", ".join("'(x := 1)'" for _ in range(25_001)) + "]\n" + " shape = Literal['List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if names and shape else 0.0\n", + "Literal", + id="exemptions-withheld-past-the-scan-limit", + ), + # Round 22 (#632 round 6): mutants of round 21's rules that moved + # money on inputs it refuses. + pytest.param( + " import importlib\n\n" + " kit = importlib.import_module('builtins')\n" + " kit.setattr(typing, 'Annotated', typing.Tuple)\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="setattr-reached-as-an-attribute", + ), + pytest.param( + " from sys import modules as found\n\n" + " found['typing_extensions'] = typing\n" + " shape = List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="modules-imported-under-another-name", + ), + pytest.param( + ' first = \'__builtins__.get("set" + "attr")(typing, "Annotated", typing.Tuple) or int\'\n' + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-calling-through-builtins", + ), + pytest.param( + " first = 'typing.__setattr__(\"Annotated\", typing.Tuple) or int'\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-calling-a-dunder", + ), + pytest.param( + # A store with no call inside a forward reference still counts. + " first = '[0 for typing.Annotated in [typing.Tuple]] and int'\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-storing-in-a-comprehension", + ), + # Round 7 of #632: a store to a subscript or attribute runs code + # (__setitem__, __setattr__) with no call, so it counts as one. + pytest.param( + " first = '[0 for typing.__dict__[\"Annotated\"] in [typing.Tuple]] and int'\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-storing-through-a-subscript", + ), + pytest.param( + " import sys\n\n" + " first = '[0 for sys.modules[\"typing_extensions\"] in [typing]] and int'\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-storing-into-modules", + ), + pytest.param( + " first = '[0 for typing.__class__ in [typing.__class__]] and int'\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if first and shape else 0.0\n", + "List", + id="forward-reference-storing-a-dunder-attribute", + ), + # The changelog's examples of an attribute of something other than a + # module, inside a typing generic or subscripted with a union. + pytest.param( + " import pandas as pd\n\n" + " shape = List[pd.DataFrame.dtypes]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "List", + id="typing-generic-over-a-class-attribute", + ), + pytest.param( + " from http import HTTPStatus\n\n" + " shape = HTTPStatus.OK.phrase[0 | 0]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "HTTPStatus.OK.phrase", + id="enum-member-attribute-subscripted-with-a-union", + ), + pytest.param( + " import w4_receipt_r19_not_loaded as aliases\n\n" + " shape = 'aliases.Sequence[Decimal | float]'\n" + " simulation.rate = 0.21 if shape else 0.0\n", + "aliases.Sequence", + id="forward-reference-to-an-unloaded-module", + ), ], ) def test_typing_aliases_built_with_a_union_inside_functions_are_refused( @@ -5682,12 +6232,16 @@ def test_typing_aliases_built_with_a_union_inside_functions_are_refused( _import_source( tmp_path, "w4_receipt_r15_aliases", - "from decimal import Decimal\n\nMONEY = Decimal | float\n", + "import typing\nfrom decimal import Decimal\n\nMONEY = Decimal | float\n\n\n" + "class Box:\n Sequence = typing.Sequence\n", ) module, _ = _import_source(tmp_path, name, source) with pytest.raises( ReceiptError, - match=rf"function raise_rate subscripts {re.escape(base)} with a union", + match=( + rf"function raise_rate subscripts {re.escape(base)} with " + r"(a union|a name the receipt cannot resolve)" + ), ) as refused: _identity(module.raise_rate) assert "cannot be bound" in str(refused.value) @@ -5756,11 +6310,110 @@ def test_typing_aliases_built_with_a_union_inside_functions_are_refused( " simulation.rate = rate(0.21)\n", id="literal-values-and-annotated-metadata", ), + pytest.param( + # Round-16 review: metadata inside a quoted Annotated is a value. + " def period(kind: List['Annotated[float, \"weekly | monthly\"]']):\n" + " return kind\n\n" + " def plain(kind: List[Annotated[float, 'weekly | monthly']]):\n" + " return kind\n\n" + " simulation.rate = 0.21 if period([1.0]) and plain([1.0]) else 0.0\n", + id="quoted-annotated-metadata", + ), pytest.param( " label = 'income_tax'\n path = 'gov.hmrc.rates[0]'\n" " simulation.rate = 0.21 if label and path else 0.0\n", id="ordinary-strings", ), + pytest.param( + " def helper(value: List[" + + _deeply_quoted("float") + + "]):\n return value\n\n" + " simulation.rate = 0.21\n", + id="deeply-quoted-layers-without-a-union", + ), + pytest.param( + " import w4_receipt_r19_not_loaded as aliases\n\n" + " shape = aliases.Sequence[float]\n" + " simulation.rate = 0.21 if shape else 0.0\n", + id="unloaded-import-without-a-union", + ), + pytest.param( + " def pick(table, mask):\n return table.rows[mask | 1]\n\n" + " simulation.rate = 0.21 if pick else 0.0\n", + id="parameter-attribute-subscript-with-bitwise-or", + ), + pytest.param( + # Round 20: Literal values and Annotated metadata that are enum + # members (attributes of a class) are values. + " from http import HTTPStatus\n\n" + " def accepted(code: Literal[HTTPStatus.OK, HTTPStatus.CREATED]):\n" + " return code\n\n" + " def weight(value: Annotated[float, HTTPStatus.OK]):\n" + " return value\n\n" + " simulation.rate = 0.21 if accepted and weight else 0.0\n", + id="enum-members-as-literal-values-and-metadata", + ), + pytest.param( + " shape = typing.Literal['List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if shape else 0.0\n", + id="attribute-form-literal-value", + ), + pytest.param( + # A store on the simulation is not a rebinding of a typing name. + " def period(kind: List['Annotated[float, \"weekly | monthly\"]']):\n" + " return kind\n\n" + " simulation.rate = 0.21 if period([1.0]) else 0.0\n" + " simulation.label = 'weekly'\n", + id="quoted-metadata-beside-a-simulation-store", + ), + pytest.param( + # Round 21 (#632 round 5): bracketed parameter paths subscript no + # name the receipt can resolve, so they are not counted; 7,000 of + # them were refused as too many to inspect. + "".join( + f" simulation.set_parameter(" + f"'gov.hmrc.income_tax.rates.uk[0].threshold_{index}', {index})\n" + for index in range(7_000) + ) + + " simulation.rate = 0.21\n", + id="many-bracketed-parameter-paths", + ), + pytest.param( + # Labels do not count toward the binding scan's limit either, so + # the changelog's quoted metadata keeps its exemption beside them. + " codes = [" + + ", ".join(f"'E{index:08d}'" for index in range(34_000)) + + "]\n" + " def period(kind: List['Annotated[float, \"weekly | monthly\"]']):\n" + " return kind\n\n" + " simulation.rate = 0.21 if codes and period([1.0]) else 0.0\n", + id="quoted-metadata-beside-many-labels", + ), + pytest.param( + " rate = getattr(simulation, 'rate', 0.0)\n" + " def period(kind: List['Annotated[float, \"weekly | monthly\"]']):\n" + " return kind\n\n" + " simulation.rate = rate if period([1.0]) else 0.0\n", + id="quoted-metadata-beside-a-literal-getattr", + ), + pytest.param( + # Round 22 (#632 round 6): a label calls nothing, so it cannot + # run code, whatever it names. + " import pandas as pd\n\n" + " frame = pd.DataFrame({'id': [1.0], 'person.id': [2.0]})\n" + " guard = ['eval', '__main__']\n" + " def period(amount: Annotated[float, 'weekly | monthly']):\n" + " return amount\n\n" + " simulation.rate = 0.21 if period(frame) is not None and guard else 0.0\n", + id="quoted-metadata-beside-labels-naming-builtins", + ), + pytest.param( + " modules = ['income_tax', 'vat']\n" + " def period(kind: List['Annotated[float, \"weekly | monthly\"]']):\n" + " return kind\n\n" + " simulation.rate = 0.21 if modules and period([1.0]) else 0.0\n", + id="quoted-metadata-beside-a-local-named-modules", + ), ], ) def test_ordinary_types_and_subscripts_inside_functions_bind(tmp_path, body): @@ -5777,6 +6430,374 @@ def test_ordinary_types_and_subscripts_inside_functions_bind(tmp_path, body): sys.modules.pop(name, None) +@pytest.mark.parametrize( + "poison", + [ + pytest.param(" typing.Annotated = typing.Tuple\n", id="stores-the-name"), + pytest.param( + " import sys\n\n sys.modules['typing_extensions'] = typing\n", + id="uses-modules", + ), + pytest.param( + " put = setattr\n put(typing, 'Annotated', typing.Tuple)\n", + id="aliases-setattr", + ), + pytest.param( + # Round 22 (#632 round 6, mutant m11). + " import sys\n from operator import attrgetter\n\n" + " attrgetter('modules')(sys)['typing_extensions'] = typing\n", + id="reads-modules-by-name", + ), + ], +) +def test_a_sibling_function_can_withhold_the_exemption(tmp_path, poison): + """Round 21 (#632 round 5): a function can call a sibling that rebinds a + typing name for it, so stores and unsound routes count module-wide.""" + + source = ( + RUNTIME_ALIAS_HEAD + "def poison():\n" + poison + "\n\n" + "def raise_rate(simulation):\n" + " poison()\n" + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n" + ) + name = "w4_receipt_r21_sibling" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="function raise_rate subscripts List"): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +@pytest.mark.parametrize( + ("sibling", "use"), + [ + pytest.param( + "async def poison():\n typing.Annotated = typing.Tuple\n", + " shape = List['typing.Annotated[float, \"List[Union[Decimal, float]]\"]']\n", + id="async-sibling-store", + ), + pytest.param( + "def poison():\n typing.Annotated = typing.Tuple\n", + " from typing import Annotated as A\n\n" + " shape = List['A[float, \"List[Union[Decimal, float]]\"]']\n", + id="sibling-store-then-import-rename", + ), + ], +) +def test_sibling_routes_withhold_the_exemption(tmp_path, sibling, use): + """Round 22 (#632 round 6, mutants m17 and m20).""" + + source = ( + RUNTIME_ALIAS_HEAD + sibling + "\n\n" + "def raise_rate(simulation):\n" + + use + + " simulation.rate = 0.21 if shape else 0.0\n" + ) + name = "w4_receipt_r22_sibling" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="function raise_rate subscripts List"): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +def test_a_name_a_module_loads_lazily_is_resolved_as_python_resolves_it(tmp_path): + """Round 22 (#632 round 6): numpy 2 and pydantic 2 provide names through + a module __getattr__ (np.polynomial, pydantic.HttpUrl). The receipt reads + them as Python does, so the verdict does not depend on whether anything + has touched the name yet.""" + + package = tmp_path / "w4_receipt_r22lazyok" + package.mkdir() + (package / "__init__.py").write_text( + "def __getattr__(name):\n" + " if name == 'Seq':\n" + " import typing\n\n" + " return typing.Sequence\n" + " raise AttributeError(name)\n" + ) + source = ( + "import w4_receipt_r22lazyok\n" + + RUNTIME_ALIAS_HEAD + + "def raise_rate(simulation):\n" + " def curve(points: Optional[w4_receipt_r22lazyok.Seq] = None):\n" + " return points\n\n" + " simulation.rate = 0.21 if curve() is None else 0.0\n" + ) + name = "w4_receipt_r22_lazy_ok" + sys.path.insert(0, str(tmp_path)) + try: + module, _ = _import_source(tmp_path, name, source) + assert "Seq" not in vars(sys.modules["w4_receipt_r22lazyok"]) + assert _identity(module.raise_rate) + finally: + sys.path.remove(str(tmp_path)) + for loaded in (name, "w4_receipt_r22lazyok"): + sys.modules.pop(loaded, None) + + +def test_a_store_on_the_middle_of_a_chain_withholds_the_exemption(tmp_path): + """Round 21 (#632 round 5, mutant n_chain_last_only): any attribute of + the chain, not only its last, counts.""" + + package = tmp_path / "w4_receipt_r21pkg" + package.mkdir() + (package / "__init__.py").write_text("") + (package / "real.py").write_text("from typing import Annotated\n") + source = ( + RUNTIME_ALIAS_HEAD + "import w4_receipt_r21pkg.real\n\n\n" + "def raise_rate(simulation):\n" + " w4_receipt_r21pkg.real = typing\n" + " shape = List['w4_receipt_r21pkg.real.Annotated" + '[float, "List[Union[Decimal, float]]"]\']\n' + " simulation.rate = 0.21 if shape else 0.0\n" + ) + name = "w4_receipt_r21_chain" + sys.path.insert(0, str(tmp_path)) + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="function raise_rate subscripts List"): + _identity(module.raise_rate) + finally: + sys.path.remove(str(tmp_path)) + for loaded in (name, "w4_receipt_r21pkg", "w4_receipt_r21pkg.real"): + sys.modules.pop(loaded, None) + + +@pytest.mark.parametrize("form", ["import-as", "from-import", "attribute-chain"]) +def test_a_lazy_loader_that_raises_does_not_crash_the_receipt(tmp_path, form): + """Round 21 (#632 round 5) and round 22 (round 6, m35 and L1): a package + whose module __getattr__ raises must not make building a receipt raise, + whether the name is imported inside the function or reached through a + module-level import; its submodule, not loaded yet, could be any object.""" + + package = tmp_path / "w4_receipt_r21lazy" + package.mkdir() + (package / "__init__.py").write_text( + "def __getattr__(name):\n raise RuntimeError('lazy loader ran')\n" + ) + importing = { + "import-as": " import w4_receipt_r21lazy.sub as sub\n", + "from-import": " from w4_receipt_r21lazy import sub\n", + "attribute-chain": " sub = None\n", + }[form] + shape = ( + "w4_receipt_r21lazy.sub.Sequence[float] if sub is None else None" + if form == "attribute-chain" + else "sub.Sequence[float]" + ) + head = ( + "import w4_receipt_r21lazy\n" + RUNTIME_ALIAS_HEAD + if form == "attribute-chain" + else RUNTIME_ALIAS_HEAD + ) + source = ( + head + "def raise_rate(simulation):\n" + importing + "\n" + f" shape = {shape}\n" + " simulation.rate = 0.21 if shape else 0.0\n" + ) + name = "w4_receipt_r21_lazy" + sys.path.insert(0, str(tmp_path)) + try: + import w4_receipt_r21lazy # noqa: F401 + + module, _ = _import_source(tmp_path, name, source) + assert _identity(module.raise_rate) + finally: + sys.path.remove(str(tmp_path)) + for loaded in (name, "w4_receipt_r21lazy"): + sys.modules.pop(loaded, None) + + +def test_a_top_level_helpers_parameter_rebinds_the_name(tmp_path): + """Round 20: the function's own parameters count, not only nested ones.""" + + source = ( + RUNTIME_ALIAS_HEAD + "def helper(Annotated):\n" + " return List['Annotated[float, \"List[Union[Decimal, float]]\"]']\n\n\n" + "def raise_rate(simulation):\n" + " simulation.rate = 0.21 if helper(typing.Tuple) else 0.0\n" + ) + name = "w4_receipt_r20_top_level_parameter" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="function helper subscripts List"): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +def _package(tmp_path, name, files): + root = tmp_path / name + root.mkdir() + for file, text in files.items(): + (root / file).write_text(text) + sys.path.insert(0, str(tmp_path)) + importlib.invalidate_caches() + return importlib.import_module(name) + + +def test_import_as_follows_the_package_attribute_as_python_does(tmp_path): + """Round 20: ``import pkg.real as aliases`` binds getattr(pkg, "real"), + not sys.modules["pkg.real"]; a package that rebinds the attribute must + not leave the receipt reading the submodule.""" + + names = ["w4r20pkg", "w4r20pkg.real", "w4r20tuple"] + (tmp_path / "w4r20tuple.py").write_text("from typing import Tuple as Annotated\n") + _package( + tmp_path, + "w4r20pkg", + { + "real.py": "from typing import Annotated\n", + "__init__.py": "from . import real as _real\nimport w4r20tuple as real\n", + }, + ) + source = ( + RUNTIME_ALIAS_HEAD + "def raise_rate(simulation):\n" + " import w4r20pkg.real as aliases\n\n" + " shape = List['aliases.Annotated[float, \"List[Union[Decimal, float]]\"]']\n" + " simulation.rate = 0.21 if shape else 0.0\n" + ) + module_name = "w4_receipt_r20_import_as" + try: + module, _ = _import_source(tmp_path, module_name, source) + with pytest.raises(ReceiptError, match="subscripts List with"): + _identity(module.raise_rate) + finally: + for name in [*names, module_name]: + sys.modules.pop(name, None) + sys.path.remove(str(tmp_path)) + + +def test_an_attribute_a_loaded_package_lacks_is_unresolved(tmp_path): + """Round 20: a submodule not imported yet could be any object.""" + + _package(tmp_path, "w4r20bare", {"__init__.py": ""}) + source = ( + RUNTIME_ALIAS_HEAD + "def raise_rate(simulation):\n" + " import w4r20bare.aliases\n\n" + " shape = w4r20bare.aliases.Sequence[Union[Decimal, float]]\n" + " simulation.rate = 0.21 if shape else 0.0\n" + ) + module_name = "w4_receipt_r20_missing_attribute" + try: + module, _ = _import_source(tmp_path, module_name, source) + with pytest.raises( + ReceiptError, match="subscripts w4r20bare.aliases.Sequence with a union" + ): + _identity(module.raise_rate) + finally: + for name in ["w4r20bare", module_name]: + sys.modules.pop(name, None) + sys.path.remove(str(tmp_path)) + + +def test_forward_references_past_the_inspection_limit_are_refused(tmp_path): + """Round 20: past the limit the function is refused, not left + uninspected.""" + + many = "List[" + "int, " * 120_000 + "int]" + source = ( + RUNTIME_ALIAS_HEAD + "def raise_rate(simulation):\n" + f" shape = List['{many}']\n" + " simulation.rate = 0.21 if shape else 0.0\n" + ) + name = "w4_receipt_r20_limit" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises( + ReceiptError, match="more forward-reference type expressions" + ): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +def test_labels_do_not_count_toward_the_inspection_limit(tmp_path): + """Strings that parse to no type (codes, labels) are not inspected, so + a long table of them leaves a real forward reference checked.""" + + codes = ", ".join(f"'E{index:08d}'" for index in range(60_000)) + source = ( + RUNTIME_ALIAS_HEAD + "def raise_rate(simulation):\n" + f" codes = [{codes}]\n" + " shape = List['List[Union[Decimal, float]]']\n" + " simulation.rate = 0.21 if codes and shape else 0.0\n" + ) + name = "w4_receipt_r20_labels" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="subscripts List with a union"): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +def test_a_string_too_deep_to_parse_is_refused(tmp_path): + """Round 20: a RecursionError while parsing fails closed.""" + + text = "x" + ".x" * 10_000 + try: + ast.parse(text, mode="eval") + except RecursionError: + pass + else: + pytest.skip("this Python parses the chain") + for statement in (f"shape = Literal['{text}']", f"label = '{text}'"): + # Under a typing subscript, and as a plain string, which could be + # evaluated as a forward reference elsewhere. + source = ( + RUNTIME_ALIAS_HEAD + "def raise_rate(simulation):\n" + f" {statement}\n" + " simulation.rate = 0.21\n" + ) + name = "w4_receipt_r20_deep_string" + try: + module, _ = _import_source(tmp_path, name, source) + with pytest.raises(ReceiptError, match="too deeply nested"): + _identity(module.raise_rate) + finally: + sys.modules.pop(name, None) + + +def test_a_deep_attribute_chain_is_resolved_and_reported_without_recursion(): + """Round-18 review: resolution and the refusal message must not recurse + once per attribute. The tree is built directly, deeper than the + interpreter's recursion limit, as no parse of source could be relied on + to produce it.""" + + chain = ModuleType("w4_receipt_r18_chain") + chain.me = chain + chain.List = typing.List + sys.modules[chain.__name__] = chain + try: + base: ast.expr = ast.Name("me", ast.Load()) + for _ in range(3 * sys.getrecursionlimit()): + base = ast.Attribute(base, "me", ast.Load()) + base = ast.Attribute(base, "List", ast.Load()) + function = ast.parse( + "def raise_rate(simulation):\n" + " from w4_receipt_r18_chain import me\n\n" + " shape = None\n" + ).body[0] + function.body[1].value = ast.Subscript( + base, ast.parse("Union[Decimal, float]", mode="eval").body, ast.Load() + ) + module = ModuleType("w4_receipt_r18_deep") + module.Union = typing.Union + module.Decimal = decimal.Decimal + problem = receipt_module._runtime_alias_problem( + module, ast.Module([function], []) + ) + assert problem is not None + assert "subscripts a deeply nested expression with a union" in problem + finally: + sys.modules.pop(chain.__name__, None) + + @pytest.mark.skipif( sys.version_info < (3, 14), reason="Union caches by equality before 3.14" ) @@ -6318,6 +7339,18 @@ def test_a_main_guard_redefining_a_function_is_refused(tmp_path): sys.modules.pop(name, None) +def test_a_string_subclass_hashes_as_its_value_not_its_str(): + """#588 round-4 review: payload strings are written with str.__str__.""" + + class Rendered(str): + def __str__(self): + return "Infinity" + + assert _reform_digest({CAP: {"2027": Rendered("1e3")}}) == _reform_digest( + {CAP: {"2027": "1e3"}} + ) + + def test_typed_infinities_are_spelled_by_float_hex(): """Round-2 review: the typed encoding of an infinite default is pinned."""