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 ``blocks.turn`` is walked to show the probe is
unreachable from it.
"""

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"}
BANNED_IMPORT_PATTERN = re.compile(r"^cuda")


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_turn() -> tuple[set[str], dict[str, 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. Over-approximating
    reachability is the safe direction for this guard.
    """
    functions: dict[str, 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, node)
    assert "turn" in functions, "blocks.turn is not defined"

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


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)


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"