tests/test_app.py

"""The app's state machine: actions fold correctly, and nothing else moves.

`apply_action` is the only way state changes, so these tests are the contract
for every control on the page. They run against the real module with the store
pointed at a temp directory, so no network and no credentials are involved.
"""

from __future__ import annotations

import importlib
import os

import pytest


@pytest.fixture(scope="module")
def app(tmp_path_factory):
    """The app module, booted against an empty local store."""
    os.environ["ARENA_OFFLINE"] = "1"
    os.environ["ARENA_LOCAL_STORE"] = str(tmp_path_factory.mktemp("store"))
    os.environ["ARENA_LOCAL_NAV"] = ""
    import app as app_module

    app_module = importlib.reload(app_module)

    # Enroll the dependency-free baselines so the selection tests exercise real
    # registry entries rather than skipping. They need no weights or network.
    from src import runtime as rt

    for model_id in ("baseline/random-walk", "baseline/drift", "baseline/bootstrap"):
        rt.enroll(app_module.STORE, "baseline", model_id, enrolled_by="tests",
                  registry=app_module.BOOT["registry"])
    app_module.BOOT["registry"] = app_module.STORE.get_registry()
    return app_module


def _state(app, **over):
    state = app.default_state()
    state.update(over)
    return state


# --------------------------------------------------------------------------
# Action folding
# --------------------------------------------------------------------------


def test_an_unknown_action_changes_nothing(app):
    before = _state(app)
    after = app.apply_action(before, "not-an-action:whatever|n1")
    assert after == before


def test_a_malformed_payload_changes_nothing(app):
    before = _state(app)
    for raw in ("", "no-colon", "|", None):
        assert app.apply_action(before, raw) == before


def test_asset_and_timeframe_are_validated(app):
    state = _state(app)
    assert app.apply_action(state, "asset:BTC-USD|n1")["asset"] == "BTC-USD"
    # Off the known list: ignored, not stored.
    assert app.apply_action(state, "asset:DOGE-USD|n2")["asset"] == state["asset"]
    assert app.apply_action(state, "tf:1d|n3")["tf"] == "1d"
    assert app.apply_action(state, "tf:7y|n4")["tf"] == state["tf"]


def test_changing_the_timeframe_resets_the_horizon_to_its_default(app):
    state = app.apply_action(_state(app), "hz:96|n1")
    assert state["horizon"] == 96
    state = app.apply_action(state, "tf:1d|n2")
    assert state["horizon"] == app.config.DEFAULT_HORIZON["1d"]


def test_an_out_of_range_horizon_is_rejected(app):
    state = _state(app, tf="1d")
    assert app.apply_action(state, "hz:100000|n1")["horizon"] == state["horizon"]
    assert app.apply_action(state, "hz:-4|n2")["horizon"] == state["horizon"]
    assert app.apply_action(state, "hz:notanumber|n3")["horizon"] == state["horizon"]


def test_the_same_nonce_twice_folds_once(app):
    """Two delivery paths mean one click can arrive twice."""
    state = _state(app, gear=False)
    once = app.apply_action(state, "gear:toggle|abc")
    twice = app.apply_action(once, "gear:toggle|abc")
    assert once["gear"] is True
    assert twice["gear"] is True, "the duplicate delivery toggled it back"


def test_playground_holds_one_model_and_matchup_holds_up_to_three(app):
    slugs = sorted(app.BOOT["registry"].get("models", {}))
    if len(slugs) < 2:
        pytest.skip("needs at least two enrolled models")

    state = _state(app, mode="matchup", selected=[slugs[0]])
    for slug in slugs[1:5]:
        state = app.apply_action(state, f"pick:{slug}|n{slug}")
    assert len(state["selected"]) <= app.view.MAX_MATCHUP

    state = app.apply_action(state, "mode:playground|nz")
    assert len(state["selected"]) == 1


def test_deselecting_the_last_model_is_refused(app):
    slugs = sorted(app.BOOT["registry"].get("models", {}))
    if not slugs:
        pytest.skip("needs an enrolled model")
    state = _state(app, mode="matchup", selected=[slugs[0]])
    state = app.apply_action(state, f"pick:{slugs[0]}|n1")
    assert state["selected"] == [slugs[0]], "left the UI with no model selected"


def test_picking_an_unenrolled_model_is_ignored(app):
    state = _state(app)
    after = app.apply_action(state, "pick:some-model-nobody-enrolled|n1")
    assert after["selected"] == state["selected"]


def test_standings_class_filter_is_validated(app):
    state = _state(app)
    assert app.apply_action(state, "cls:Crypto|n1")["cls"] == "Crypto"
    assert app.apply_action(state, "cls:Nonsense|n2")["cls"] == state["cls"]


def test_the_enrollment_box_is_length_capped(app):
    state = app.apply_action(_state(app), "hfid:" + "x" * 5000 + "|n1")
    assert len(state["hfid"]) <= 120


# --------------------------------------------------------------------------
# Rendering
# --------------------------------------------------------------------------


def test_the_page_renders_for_every_mode(app):
    for mode in ("playground", "matchup"):
        html = app.render(_state(app, mode=mode))
        assert "Forecast Arena" in html
        assert len(html) > 5000


def test_the_page_renders_with_no_models_enrolled(app, monkeypatch):
    """An empty registry is the first-boot state, not an error."""
    monkeypatch.setitem(app.BOOT, "registry", {"models": {}})
    html = app.render(_state(app, selected=[]))
    assert "Forecast Arena" in html


def test_the_disclaimer_is_always_on_the_page(app):
    """A forecast must never be presented as advice, on any state of the page."""
    html = app.render(_state(app)).lower()
    assert "nothing here is financial advice" in html
    # Twice: once beside the method note, once in the footer.
    assert html.count("financial advice") >= 2


def test_the_methodology_promises_are_on_the_page(app):
    html = app.render(_state(app))
    for promise in ("frozen at issue", "backfilled entries", "resolved forecasts only"):
        assert promise in html.lower()


def test_on_action_returns_html_and_state(app):
    html, state = app.on_action("cls:Crypto|n1", app.default_state())
    assert isinstance(html, str) and len(html) > 1000
    assert state["cls"] == "Crypto"


# --------------------------------------------------------------------------
# Deep links
# --------------------------------------------------------------------------


def test_a_model_deep_link_preselects_and_badges(app):
    slugs = sorted(app.BOOT["registry"].get("models", {}))
    if not slugs:
        pytest.skip("needs an enrolled model")
    state = app.apply_action(_state(app), f"deeplink:{slugs[0]}|d1")
    assert state["selected"] == [slugs[0]]
    assert state["deep_linked"] is True
    assert state["mode"] == "playground"
    assert "DEEP LINK" in app.render(state)


def test_a_stale_deep_link_opens_the_arena_rather_than_failing(app):
    before = _state(app)
    after = app.apply_action(before, "deeplink:a-model-that-was-retired|d2")
    assert after["selected"] == before["selected"]
    assert not after.get("deep_linked")
    assert "Forecast Arena" in app.render(after)


def test_a_deep_link_cannot_smuggle_a_path(app):
    for hostile in ("../../etc/passwd", "a|b", "<script>"):
        after = app.apply_action(_state(app), f"deeplink:{hostile}|d3")
        assert not after.get("deep_linked")


# --------------------------------------------------------------------------
# The data endpoints
# --------------------------------------------------------------------------


def test_health_reports_what_ci_asserts(app):
    h = app.health()
    assert h["ok"] is True
    assert h["models_enrolled"] >= 1
    assert h["store"] == app.config.STORE_REPO
    assert "models_by_tier" in h and "gpu_available" in h


def test_the_forecast_api_refuses_an_unenrolled_model(app):
    out = app.forecast_api("nobody/nothing")
    assert out["ok"] is False
    assert "not enrolled" in out["error"]


def test_the_forecast_api_validates_asset_and_timeframe(app):
    slugs = sorted(app.BOOT["registry"].get("models", {}))
    if not slugs:
        pytest.skip("needs an enrolled model")
    out = app.forecast_api(slugs[0], asset="DOGE-USD")
    assert out["ok"] is False
    out = app.forecast_api(slugs[0], timeframe="17y")
    assert out["ok"] is False


# --------------------------------------------------------------------------
# Identity
# --------------------------------------------------------------------------


class _Profile:
    username = "someone"


def test_identity_comes_only_from_the_oauth_profile(app):
    """State round-trips through the client, so it cannot promote itself."""
    forged = _state(app, signed_in=True, user="admin")
    folded = app._with_identity(forged, None)
    assert folded["signed_in"] is False
    assert folded["user"] == ""


def test_a_signed_in_profile_unlocks_the_session(app):
    folded = app._with_identity(_state(app), _Profile())
    assert folded["signed_in"] is True
    assert folded["user"] == "someone"
    html = app.render(folded)
    assert "SOMEONE" in html
    assert "ANONYMOUS SESSION" not in html


def test_the_anonymous_header_counts_down_the_session_cap(app):
    state = _state(app, session_forecasts=3)
    html = app.render(app._with_identity(state, None))
    assert "ANONYMOUS SESSION" in html
    assert f"{app.config.ANON_SESSION_CAP - 3} FORECASTS LEFT" in html


def test_a_signed_in_visitor_is_not_blocked_from_gpu_models(app, monkeypatch):
    """The sign-in gate on GPU-tier models, exercised offline.

    The fixture registry holds only baselines, which are all CPU-tier, so this
    used to skip -- and a gate that is never tested is a gate that can rot. A
    synthetic GPU-tier entry makes it run everywhere.
    """
    registry = {"models": dict(app.BOOT["registry"].get("models", {}))}
    registry["models"]["synthetic-gpu"] = {
        "model_slug": "synthetic-gpu", "model_id": "fixture/gpu-model",
        "family": "baseline", "revision": "test", "display": "Synthetic GPU",
        "capabilities": {"output": "ohlcv_paths", "hardware": "gpu",
                         "max_context": 256, "asset_generality": "general"},
    }
    monkeypatch.setitem(app.BOOT, "registry", registry)

    anon = _state(app, selected=["synthetic-gpu"], signed_in=False)
    assert app.build_data(anon)["quota_blocked"] is True

    signed = _state(app, selected=["synthetic-gpu"], signed_in=True)
    blocked = app.build_data(signed)["quota_blocked"]
    if app.gpu_dispatch.available():
        # Signed in on GPU hardware: nothing left to gate on.
        assert blocked is False
    else:
        # No GPU attached, so it stays blocked -- but for the hardware reason,
        # not the sign-in one, and the message has to say which.
        assert blocked is True
        assert "CPU hardware" in app.build_data(signed)["quota_msg"]


def test_a_cpu_tier_model_is_never_quota_blocked(app):
    cpu_models = [s for s, e in app.BOOT["registry"].get("models", {}).items()
                  if e.get("capabilities", {}).get("hardware") == "cpu"]
    assert cpu_models, "the fixture registry should hold CPU-tier models"
    state = _state(app, selected=[cpu_models[0]], signed_in=False)
    assert app.build_data(state)["quota_blocked"] is False


# --------------------------------------------------------------------------
# Backfill contribution
# --------------------------------------------------------------------------


def test_backfill_requires_sign_in(app):
    """It spends real compute, so it comes out of the contributor's quota."""
    state = _state(app, signed_in=False)
    after = app.apply_action(state, "backfill:run|b1")
    assert after["error"]["kind"] == "quota"
    assert "your own quota" in after["error"]["message"]
    assert not after.get("backfill_requested")


def test_backfill_reports_what_it_actually_did(app, monkeypatch):
    """The button must not claim work it did not do.

    It used to flip a label to "Backfill queued" and run nothing at all, which
    is exactly the placeholder this codebase refuses to ship.
    """
    calls = []

    class _Run:
        archived_rows = 24

    def fake_run_forecast(store, slug, asset, tf, **kw):
        calls.append(kw.get("as_of"))
        assert kw.get("backfilled") is True, "contributed rows must be labelled"
        assert kw.get("as_of") is not None, "a backfill must be issued as-of"
        return _Run()

    monkeypatch.setattr(app.runtime, "run_forecast", fake_run_forecast)
    monkeypatch.setattr(app, "_flush_archive", lambda: None)

    slugs = sorted(app.BOOT["registry"].get("models", {}))
    if not slugs:
        pytest.skip("needs an enrolled model")

    state = _state(app, signed_in=True, selected=[slugs[0]])
    after = app.apply_action(state, "backfill:run|b2")

    if not calls:
        # No price history in the fixture store: it must say so, not pretend.
        assert after.get("backfill_note")
        return

    assert len(calls) <= app.BACKFILL_CHUNK, "a click must stay bounded"
    assert len(set(calls)) == len(calls), "the same window was issued twice"
    assert after["backfill_note"]
    assert str(len(calls)) in after["backfill_note"]


def test_the_backfill_note_reaches_the_page(app):
    state = _state(app, backfill_note="7 window(s) added")
    assert "7 window(s) added" in app.render(state)


def test_enrollment_requires_sign_in(app):
    """A public registry row should carry a name, and the smoke test a quota."""
    state = _state(app, signed_in=False, hfid="amazon/chronos-bolt-tiny",
                   family="Chronos · quantile line")
    after = app.apply_action(state, "enroll:run|e1")
    assert after["enroll_ok"] is False
    assert "Sign in to enroll" in after["enroll_note"]
    assert "Sign in to enroll" in app.render(after)


def test_a_signed_in_enrollment_reaches_the_registry_path(app, monkeypatch):
    seen = {}

    class _Outcome:
        ok, already, message, model_slug, entry = True, False, "enrolled", "x", {}

    def fake_enroll(store, family, model_id, **kw):
        seen.update(family=family, model_id=model_id, by=kw.get("enrolled_by"))
        return _Outcome()

    monkeypatch.setattr(app.runtime, "enroll", fake_enroll)
    monkeypatch.setattr(app, "_flush_archive", lambda: None)

    state = _state(app, signed_in=True, user="someone",
                   hfid="amazon/chronos-bolt-tiny",
                   family="Chronos · quantile line")
    after = app.apply_action(state, "enroll:run|e2")

    assert after["enroll_ok"] is True
    assert seen["family"] == "chronos", "the label must map back to a family slug"
    assert seen["by"] == "someone", "the registry must record who enrolled it"


def test_an_absent_cached_forecast_is_not_memoised(app, monkeypatch):
    """A series with nothing archived today may have something tomorrow."""
    app._CACHED_RUNS.clear()
    calls = []

    def fake(store, slug, asset, tf, registry=None):
        calls.append(slug)
        return None

    monkeypatch.setattr(app.runtime, "cached_run", fake)
    app._cached_run("m", "BTC-USD", "1h")
    app._cached_run("m", "BTC-USD", "1h")
    assert len(calls) == 2, "a miss was cached and never re-checked"