read.markets/tests/test_chat_and_log_gates.py

"""Free-vs-paid gating on /api/chat and the strategic-log read endpoints.

Mirrors the integration-style setup from test_news_window.py: real router
over an in-memory aiosqlite DB, with two seeded users (free + paid)."""
from __future__ import annotations

import asyncio
from datetime import datetime, timezone

import pytest


_TONE = "INTERMEDIATE"  # matches CASSANDRA_TONE default


def _build_app(tmp_path):
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from app import db as db_mod
    from app.auth import sign_session
    from app.db import Base
    from app.models import StrategicLog, User
    from app.routers import api as api_router

    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/gates.db")
    factory = async_sessionmaker(engine, expire_on_commit=False)
    db_mod._engine = engine
    db_mod._session_factory = factory

    # Two logs on the same UTC day: the newer one is at hour 1 (non-boundary)
    # and the older one is at hour 0 (boundary). Free users should see the
    # boundary one; paid users should see the newer one.
    base = datetime(2026, 5, 25, tzinfo=timezone.utc)
    boundary_at = base.replace(hour=0, minute=20)
    newer_at = base.replace(hour=1, minute=20)

    async def _seed():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        async with factory() as s:
            s.add(User(id=1, email="free@x", tier="free"))
            s.add(User(id=2, email="paid@x", tier="paid"))
            s.add(StrategicLog(
                generated_at=boundary_at, model="m", tone=_TONE,
                analysis="DRY", content="boundary-log",
            ))
            s.add(StrategicLog(
                generated_at=newer_at, model="m", tone=_TONE,
                analysis="DRY", content="non-boundary-log",
            ))
            await s.commit()

    asyncio.run(_seed())

    app = FastAPI()
    app.include_router(api_router.router, prefix="/api")
    client = TestClient(app)
    return client, sign_session(1), sign_session(2)


# --- /api/chat gating ------------------------------------------------------


def test_chat_blocks_free_user_with_402(tmp_path):
    client, free_sess, _ = _build_app(tmp_path)
    r = client.post(
        "/api/chat",
        json={"messages": [{"role": "user", "content": "hi"}]},
        cookies={"cassandra_session": free_sess},
    )
    assert r.status_code == 402, r.text
    body = r.json()
    assert body["detail"]["code"] == "paid_required"


def test_chat_lets_paid_user_past_the_gate(tmp_path, monkeypatch):
    """Paid users should clear the tier gate. The next check is the
    OPENROUTER_API_KEY presence — without a key it 503s, which is fine:
    the point is that they got past the paid gate (not 402)."""
    client, _, paid_sess = _build_app(tmp_path)
    # Make sure no real key leaks in from the host environment.
    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
    from app.config import get_settings
    get_settings.cache_clear()  # type: ignore[attr-defined]

    r = client.post(
        "/api/chat",
        json={"messages": [{"role": "user", "content": "hi"}]},
        cookies={"cassandra_session": paid_sess},
    )
    # Paid clears the tier gate. With no API key configured the endpoint
    # then 503s — that's the next check in the handler.
    assert r.status_code in (503, 200), r.text


# --- /api/log/latest free-tier 6-hour throttle -----------------------------


def test_log_latest_free_user_sees_boundary_hour_only(tmp_path):
    client, free_sess, _ = _build_app(tmp_path)
    r = client.get(
        f"/api/log/latest?tone={_TONE}",
        cookies={"cassandra_session": free_sess},
    )
    assert r.status_code == 200, r.text
    assert r.json()["content"] == "boundary-log", (
        "free user must see the 00:20 boundary log, not the newer 01:20 one"
    )


def test_log_latest_paid_user_sees_most_recent(tmp_path):
    client, _, paid_sess = _build_app(tmp_path)
    r = client.get(
        f"/api/log/latest?tone={_TONE}",
        cookies={"cassandra_session": paid_sess},
    )
    assert r.status_code == 200, r.text
    assert r.json()["content"] == "non-boundary-log", (
        "paid user must see the most recent log regardless of generation hour"
    )


# --- /api/log/by-date free-tier filter -------------------------------------


def test_log_by_date_free_user_filters_to_boundary(tmp_path):
    client, free_sess, _ = _build_app(tmp_path)
    r = client.get(
        f"/api/log/by-date/2026-05-25?tone={_TONE}",
        cookies={"cassandra_session": free_sess},
    )
    assert r.status_code == 200, r.text
    assert r.json()["content"] == "boundary-log"


def test_log_by_date_paid_user_sees_most_recent(tmp_path):
    client, _, paid_sess = _build_app(tmp_path)
    r = client.get(
        f"/api/log/by-date/2026-05-25?tone={_TONE}",
        cookies={"cassandra_session": paid_sess},
    )
    assert r.status_code == 200, r.text
    assert r.json()["content"] == "non-boundary-log"