tests/test_no_gpu_on_turn_path.py

"""SC-4 static guard: nothing on the turn path can reach a GPU.

The whole Phase 1 free-tier story rests on the turn loop completing with ``DISABLE_GPU=1``. The
deployed half of that proof is plan 01-09's; this file is the half that runs in the quick loop and
fails a build before it ever reaches the Space.

The scan is deliberately conservative. A false positive here costs one look at a diff; a false
negative is the exact thing SC-4 exists to prevent. So: any decorator whose dotted name ends in
``GPU`` anywhere under ``src/japanese_avatar/`` fails, full stop, and ``app.py`` may carry exactly
one such function - the ZeroGPU startup probe - only because the platform refuses to run a Space
without one (docs/HOSTING.md, first deploy record). That carve-out is spelled out by name rather
than by loosening the rule, and the call graph from every server function - ``turn`` and, since
plan 02-06, ``analyze`` / ``translate`` / ``language_info`` - is walked to show the probe is
unreachable from each. ``transformers`` joined the banned imports in 02-06: the translation path
is CTranslate2 + sentencepiece by construction (02-04), and a stray ``transformers`` import would
pull torch onto the turn path in one line.
"""

from __future__ import annotations

import ast
import importlib
import re
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
SRC = REPO_ROOT / "src" / "japanese_avatar"
APP = REPO_ROOT / "app.py"

#: The one @spaces.GPU function ZeroGPU insists on. 01-RESEARCH.md Open Question 2 called it
#: `gpu_healthcheck`; the name that actually shipped in plan 01-05 is below. Nothing on the turn
#: path may call it, import it, or be it.
GPU_PROBE_NAME = "zerogpu_probe"

BANNED_IMPORT_ROOTS = {"torch", "spaces", "transformers"}
BANNED_IMPORT_PATTERN = re.compile(r"^cuda")

#: Every gr.HTML server function registered in blocks.py. Each is a root of the reachability walk.
SERVER_ENTRY_POINTS = ["turn", "analyze", "translate", "language_info"]


def _turn_path_modules() -> list[Path]:
    modules = sorted(SRC.rglob("*.py"))
    assert modules, f"no modules under {SRC}"
    return modules


def _parse(path: Path) -> ast.Module:
    return ast.parse(path.read_text(encoding="utf-8"), filename=str(path))


def _decorator_name(node: ast.expr) -> str:
    """Dotted name of a decorator expression: `spaces.GPU(duration=1)` -> `spaces.GPU`."""
    if isinstance(node, ast.Call):
        node = node.func
    parts: list[str] = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
    return ".".join(reversed(parts))


def _gpu_decorated(tree: ast.Module) -> list[str]:
    hits = []
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
            for dec in node.decorator_list:
                name = _decorator_name(dec)
                if name == "GPU" or name.endswith(".GPU"):
                    hits.append(node.name)
    return hits


def _imports(tree: ast.Module) -> set[str]:
    roots: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                roots.add(alias.name.split(".")[0])
        elif isinstance(node, ast.ImportFrom) and node.module:
            roots.add(node.module.split(".")[0])
    return roots


def _called_names(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
    """Every bare or attribute name that is called inside ``fn``."""
    names: set[str] = set()
    for node in ast.walk(fn):
        if isinstance(node, ast.Call):
            target = node.func
            if isinstance(target, ast.Name):
                names.add(target.id)
            elif isinstance(target, ast.Attribute):
                names.add(target.attr)
    return names


def _reachable_from(entry: str) -> tuple[set[str], dict[str, list[ast.FunctionDef]]]:
    """Conservative call graph over every function defined in the package plus app.py.

    Names are matched by simple identifier, ignoring module boundaries, so an attribute call
    `tts.synthesize(...)` reaches every function called `synthesize` anywhere - and a name
    defined in several modules (``analyze`` is both ``nlp.analyzer.analyze`` and the server
    function ``ui.blocks.analyze``) contributes the calls of EVERY definition. Over-approximating
    reachability is the safe direction for this guard.
    """
    functions: dict[str, list[ast.FunctionDef]] = {}
    for path in [*_turn_path_modules(), APP]:
        for node in ast.walk(_parse(path)):
            if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
                functions.setdefault(node.name, []).append(node)
    assert entry in functions, f"blocks.{entry} is not defined"

    reachable: set[str] = set()
    frontier = [entry]
    while frontier:
        name = frontier.pop()
        if name in reachable:
            continue
        reachable.add(name)
        for definition in functions[name]:
            for called in _called_names(definition):
                if called in functions and called not in reachable:
                    frontier.append(called)
    return reachable, functions


def _reachable_from_turn() -> tuple[set[str], dict[str, list[ast.FunctionDef]]]:
    return _reachable_from("turn")


def test_no_spaces_gpu_decorator_anywhere_on_turn_path():
    for path in _turn_path_modules():
        hits = _gpu_decorated(_parse(path))
        assert not hits, f"{path.relative_to(REPO_ROOT)} decorates {hits} with a GPU decorator"

    app_hits = _gpu_decorated(_parse(APP))
    assert app_hits in ([], [GPU_PROBE_NAME]), (
        f"app.py carries GPU-decorated functions {app_hits}; only the ZeroGPU startup probe "
        f"{GPU_PROBE_NAME!r} is permitted, and it must stay unreachable from the turn path"
    )

    reachable, _functions = _reachable_from_turn()
    assert GPU_PROBE_NAME not in reachable, (
        f"{GPU_PROBE_NAME} is reachable from blocks.turn: {sorted(reachable)}"
    )
    assert "turn" in reachable and "synthesize" in reachable, sorted(reachable)


@pytest.mark.parametrize("entry", SERVER_ENTRY_POINTS)
def test_gpu_probe_unreachable_from_every_server_function(entry):
    """Plan 02-06: the language server functions are turn-path code too (SC-4 carried forward)."""
    reachable, functions = _reachable_from(entry)
    assert GPU_PROBE_NAME not in reachable, (
        f"{GPU_PROBE_NAME} is reachable from blocks.{entry}: {sorted(reachable)}"
    )
    assert entry in reachable
    # Every server function waits for the language warm-up, so the walk must see it.
    assert "warm_language" in reachable, f"blocks.{entry} does not call warm_language()"
    assert "warm_language" in functions


def test_no_gpu_imports_on_turn_path():
    for path in _turn_path_modules():
        roots = _imports(_parse(path))
        banned = {r for r in roots if r in BANNED_IMPORT_ROOTS or BANNED_IMPORT_PATTERN.match(r)}
        assert not banned, f"{path.relative_to(REPO_ROOT)} imports {sorted(banned)}"

    app_tree = _parse(APP)
    app_roots = _imports(app_tree)
    assert not {r for r in app_roots if r == "torch" or BANNED_IMPORT_PATTERN.match(r)}
    if "spaces" in app_roots:
        defined = {n.name for n in ast.walk(app_tree) if isinstance(n, ast.FunctionDef)}
        assert GPU_PROBE_NAME in defined, (
            "app.py imports spaces without defining the ZeroGPU probe that justifies the import"
        )
        reachable, _functions = _reachable_from_turn()
        assert GPU_PROBE_NAME not in reachable
    # The package must never import the entry point back, or the carve-out would leak inward.
    for path in _turn_path_modules():
        assert "app" not in _imports(_parse(path)), f"{path.name} imports app.py"


def test_disable_gpu_env_var_is_honoured(monkeypatch):
    """With DISABLE_GPU=1 every @spaces.GPU function in the codebase raises RuntimeError.

    There is exactly one such function, the ZeroGPU startup probe in app.py. Off the platform
    the decorator is a no-op wrapper, so calling it runs the body, and the body must refuse.
    If the probe is ever removed this test passes trivially, which is the correct outcome.
    """
    monkeypatch.setenv("DISABLE_GPU", "1")
    monkeypatch.setenv("GRADIO_ANALYTICS_ENABLED", "False")
    monkeypatch.syspath_prepend(str(REPO_ROOT))
    sys.modules.pop("app", None)
    app = importlib.import_module("app")

    assert app.gpu_disabled() is True
    probes = [
        getattr(app, name)
        for name in _gpu_decorated(_parse(APP))
        if callable(getattr(app, name, None))
    ]
    if not probes:
        pytest.skip("no @spaces.GPU function exists; nothing to refuse")
    for probe in probes:
        with pytest.raises(RuntimeError):
            probe()

    monkeypatch.setenv("DISABLE_GPU", "0")
    assert app.gpu_disabled() is False, "gpu_disabled() must read the variable per call"