"""AST guard — no Core-level DML may bypass the access-cycle ledger.

``app.core.access_cycle_listener`` observes the SQLAlchemy unit of work.
It therefore sees ``session.delete(obj)`` and ``session.add(obj)``, and is
completely blind to ``session.execute(delete(AuthUserRole).where(...))``
— the statement goes straight to the database and no ORM event fires.

That blindness is not theoretical: before migration 484 there were five
such call sites (``location_admin_sync`` ×2, ``department_role_sync`` ×2,
the leaver guard). Each one silently stripped a role without closing its
holding interval, which in a point-in-time query reads as "still holds
it, forever" — the worst possible answer for the leaver strip in
particular. PR1 converted all five; this test stops the sixth.

The bar is deliberately blunt: ANY ``delete(...)`` / ``update(...)``
construct naming one of the two ledger-backed models, anywhere under
``app/``, fails. If a future batch path genuinely needs Core DML for
volume reasons, it must also write the ledger rows itself and be added
to ``_ALLOWED`` with a comment saying so — an explicit, reviewable
exception rather than a silent regression.
"""
from __future__ import annotations

import ast
from pathlib import Path

# Models whose lifecycle the ledger tracks. Core DML against either one
# bypasses ``app.core.access_cycle_listener`` entirely.
_LEDGER_BACKED_MODELS = {"AuthUserRole", "UserPermission"}

_DML_CONSTRUCTS = {"delete", "update"}

# Modules permitted to issue Core DML against the models above. Each
# entry needs a comment explaining how that module keeps the ledger
# correct without the ORM unit of work.
_ALLOWED: dict[str, str] = {}

_APP_ROOT = Path(__file__).resolve().parents[2] / "app"


def _offending_calls(tree: ast.AST) -> list[tuple[int, str]]:
    """Find ``delete(Model)`` / ``update(Model)`` calls in one module.

    Matches both the bare ``delete(AuthUserRole)`` import style and the
    aliased ``_delete(AuthUserRole)`` form the leaver guard used, by
    looking at the call's first positional argument rather than at the
    function name alone.
    """
    hits: list[tuple[int, str]] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or not node.args:
            continue
        func = node.func
        name = func.id if isinstance(func, ast.Name) else (
            func.attr if isinstance(func, ast.Attribute) else ""
        )
        # ``_delete`` / ``sa.delete`` / ``delete`` all normalise here.
        if name.lstrip("_") not in _DML_CONSTRUCTS:
            continue
        first = node.args[0]
        target = first.id if isinstance(first, ast.Name) else (
            first.attr if isinstance(first, ast.Attribute) else ""
        )
        if target in _LEDGER_BACKED_MODELS:
            hits.append((node.lineno, f"{name}({target})"))
    return hits


def test_no_core_dml_bypasses_the_access_cycle_ledger():
    violations: list[str] = []
    for path in sorted(_APP_ROOT.rglob("*.py")):
        rel = path.relative_to(_APP_ROOT.parent).as_posix()
        if rel in _ALLOWED:
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except SyntaxError:  # pragma: no cover - would fail the build elsewhere
            continue
        for lineno, call in _offending_calls(tree):
            violations.append(f"{rel}:{lineno} — {call}")

    assert not violations, (
        "Core-level DML against a ledger-backed model bypasses "
        "app.core.access_cycle_listener, so the holding interval is never "
        "closed and every point-in-time query keeps reporting the access "
        "as current. Use an ORM delete (load the rows, then "
        "``await db.delete(row)``), or add the module to _ALLOWED with a "
        "note on how it writes the ledger itself:\n  "
        + "\n  ".join(violations)
    )


def test_guard_detects_the_pattern_it_claims_to_detect():
    """The guard is worthless if its matcher silently stops matching.

    Exercises both call shapes that existed in the codebase before PR1:
    the bare import and the aliased one.
    """
    source = (
        "await db.execute(delete(AuthUserRole).where(x))\n"
        "await db.execute(_delete(UserPermission).where(y))\n"
        "await db.execute(sa.update(AuthUserRole).values(z))\n"
        "await db.delete(loaded_row)\n"          # the correct form
        "await db.execute(delete(SomeOtherModel))\n"  # unrelated model
    )
    hits = _offending_calls(ast.parse(source))
    assert [call for _lineno, call in hits] == [
        "delete(AuthUserRole)",
        "_delete(UserPermission)",
        "update(AuthUserRole)",
    ]