"""Regression: every failed capability must NOT land on ``Unhandled sync error: greenlet_spawn`` (prod 2026-04-23 → 2026-04-24, 37 occurrences in ``cmdb_sync_jobs.error_message``). Root cause: ``AsyncSession.rollback()`` expires every ORM instance in the identity map — NOT just the ones touched by the rolled-back transaction. ``expire_on_commit=False`` on the sessionmaker only applies to commits, not rollbacks. Post-rollback, accessing a column on an expired ORM instance triggers SQLAlchemy's sync lazy-refresh path and raises ``greenlet_spawn has not been called; can't call await_only()``. The code path that hit this: any capability raises → ``except`` block runs ``await db.rollback()`` (expires ``integration``) → loop ends → ``_send_sync_notification(..., integration.name or integration_id, ...)`` tries to re-read ``integration.name`` on the expired instance → greenlet_spawn. The fix has two halves, both pinned structurally by this test: 1. Cache ``integration.name`` / ``.provider`` / ``.secrets_*`` / ... into plain-string locals UP FRONT, before any rollback can fire. The post-loop notification path reads the local, never the ORM instance. 2. After each per-capability rollback, ``await db.refresh(integration)`` so the next iteration in a wildcard run can still access ``integration.id`` without triggering a sync lazy-load. This is a static AST-level check rather than an end-to-end sync run because reproducing greenlet_spawn deterministically requires a specific SQLAlchemy internal state that's awkward to set up — but the static fingerprint of the fix is unambiguous enough to catch any future refactor that regresses it. """ from __future__ import annotations import ast from pathlib import Path def _sync_source() -> str: return ( Path(__file__).resolve().parents[2] / "app" / "domains" / "cmdb" / "sync.py" ).read_text() def _find_func(src: str, name: str) -> ast.FunctionDef: for node in ast.walk(ast.parse(src)): if isinstance(node, ast.AsyncFunctionDef) and node.name == name: return node raise AssertionError(f"function {name!r} not found in sync.py") def test_run_sync_impl_caches_integration_attrs_up_front() -> None: """``_run_sync_impl`` must cache every integration attribute it needs post-loop into plain locals BEFORE the capability loop runs — otherwise a rollback inside the loop expires the instance and the post-loop re-read raises greenlet_spawn.""" fn = _find_func(_sync_source(), "_run_sync_impl") # Collect assigned names before the first `for` loop. pre_loop_targets: set[str] = set() for stmt in fn.body: if isinstance(stmt, ast.For): break for sub in ast.walk(stmt): if isinstance(sub, ast.Assign): for tgt in sub.targets: if isinstance(tgt, ast.Name): pre_loop_targets.add(tgt.id) if isinstance(sub, ast.AnnAssign) and isinstance(sub.target, ast.Name): pre_loop_targets.add(sub.target.id) required = { "integration_alias", # used by _send_sync_notification post-rollback "integration_provider", # used by get_provider_class } missing = required - pre_loop_targets assert not missing, ( f"_run_sync_impl must cache these locals before the capability " f"loop: {missing}" ) def test_send_sync_notification_uses_cached_alias_not_orm_attr() -> None: """The notification call site must use the cached ``integration_alias`` local (never ``integration.name``), since ``_send_sync_notification`` runs AFTER the capability loop — any capability failure has rolled back + expired the ORM instance.""" src = _sync_source() # Find every call to _send_sync_notification inside _run_sync_impl. fn = _find_func(src, "_run_sync_impl") notification_calls: list[ast.Call] = [] for sub in ast.walk(fn): if ( isinstance(sub, ast.Call) and isinstance(sub.func, ast.Name) and sub.func.id == "_send_sync_notification" ): notification_calls.append(sub) assert notification_calls, "expected at least one _send_sync_notification call" # At least one call (the POST-LOOP one) must pass integration_alias # and NONE of them may pass the attribute access integration.name. uses_cached_local = False for call in notification_calls: for arg in call.args: # Shape: positional `integration_alias` or `integration.name or ...` if isinstance(arg, ast.Name) and arg.id == "integration_alias": uses_cached_local = True if isinstance(arg, ast.BoolOp): for sub in ast.walk(arg): if ( isinstance(sub, ast.Attribute) and isinstance(sub.value, ast.Name) and sub.value.id == "integration" and sub.attr == "name" ): raise AssertionError( "_send_sync_notification call passes " "integration.name — use the cached " "integration_alias local instead to avoid " "post-rollback greenlet_spawn" ) assert uses_cached_local, ( "at least one _send_sync_notification call must pass the cached " "integration_alias local" ) def test_capability_except_blocks_refresh_integration_after_rollback() -> None: """Each ``except`` block that calls ``await db.rollback()`` must ALSO call ``await db.refresh(integration)`` to un-expire the ORM instance for the next capability iteration (wildcard full-sync).""" src = _sync_source() fn = _find_func(src, "_run_sync_impl") except_handlers: list[ast.ExceptHandler] = [ node for node in ast.walk(fn) if isinstance(node, ast.ExceptHandler) ] # The capability loop has three except handlers: ProviderAuthError, # CapabilityNotSupported, and generic Exception. Each must refresh. rollback_handlers = [] for h in except_handlers: has_rollback = False for sub in ast.walk(h): if ( isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute) and sub.func.attr == "rollback" ): has_rollback = True if has_rollback: rollback_handlers.append(h) assert len(rollback_handlers) >= 3, ( f"expected at least 3 except-handlers with rollback, got " f"{len(rollback_handlers)}" ) for h in rollback_handlers: refreshes_integration = False for sub in ast.walk(h): if ( isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute) and sub.func.attr == "refresh" ): # Check first positional arg is the name "integration". if ( sub.args and isinstance(sub.args[0], ast.Name) and sub.args[0].id == "integration" ): refreshes_integration = True assert refreshes_integration, ( f"except-handler at line {h.lineno} calls db.rollback() but " f"not db.refresh(integration); rollback expires the ORM " f"instance and the next capability / post-loop code will " f"hit greenlet_spawn" )