tests/test_db_qr.py

"""Contract DB + QR + logging recommend (BRIEF 04/08/2026, Bước 4).

Toàn bộ chạy SQLite in-memory — pytest KHÔNG cần Docker/Postgres (Postgres
chỉ cần cho dev thật, gate G3 chạy tay). Engine vẫn mock như mọi test app:
kiểm CONTRACT (status/shape/message/row DB), không kiểm vật lý.

Tái dùng `make_shot`/`v2_result`/`BALLS_FULL` của test_api_v2 — một nguồn
duy nhất cho shape ShotFull, hai file không được trôi khác nhau.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
import sqlalchemy as sa
from fastapi.testclient import TestClient

from poolcoach_rl import recommend as rec_pkg

from app import db, qr
from app import main as app_main
from test_api_v2 import BALLS_FULL, make_shot, v2_result


@pytest.fixture
def client(monkeypatch, env_stub):
    monkeypatch.setattr(app_main.state, "env_h", env_stub)
    monkeypatch.setattr(app_main.state, "jit_ready", True)
    monkeypatch.setattr(app_main.state, "boot_error", None)
    return TestClient(app_main.app)


@pytest.fixture
def db_mem():
    """DB SQLite in-memory sạch cho MỖI test; teardown trả module `db` về
    trạng thái tắt để các test degraded/cũ không thấy DB sót lại."""
    db.setup("sqlite:///:memory:")
    db.Base.metadata.create_all(db.get_engine())
    yield db
    db.teardown()


@pytest.fixture
def no_db():
    """Degraded mode tường minh — như Space HF không có DATABASE_URL."""
    db.teardown()
    yield


def _seed_table(name="Bàn 1"):
    with db.session() as s:
        tenant = db.Tenant(name="dev")
        s.add(tenant)
        s.flush()
        tbl = db.Table(tenant_id=tenant.id, name=name)
        s.add(tbl)
        s.flush()
        return tenant.id, tbl.id


def patch_v2(monkeypatch, result):
    monkeypatch.setattr(rec_pkg, "recommend_v2", lambda *a, **k: result)


# ------------------------------------------------------------ /api/table-qr

def test_qr_hop_le_tra_session_va_ban(client, db_mem):
    tenant_id, table_id = _seed_table()
    token = qr.make_table_token(tenant_id, table_id)

    r = client.get(f"/api/table-qr/{token}")
    assert r.status_code == 200
    data = r.json()
    assert set(data) == {"session_id", "table", "expires_at"}
    assert data["table"] == {"id": table_id, "name": "Bàn 1"}
    # session phải nằm trong DB, không phải chỉ là chuỗi bịa trong response
    with db.session() as s:
        gs = s.get(db.GuestSession, data["session_id"])
        assert gs is not None
        assert (gs.tenant_id, gs.table_id) == (tenant_id, table_id)


def test_qr_ttl_24h(client, db_mem):
    tenant_id, table_id = _seed_table()
    r = client.get(f"/api/table-qr/{qr.make_table_token(tenant_id, table_id)}")

    with db.session() as s:
        gs = s.get(db.GuestSession, r.json()["session_id"])
        ttl_h = (gs.expires_at - gs.created_at).total_seconds() / 3600
    assert ttl_h == pytest.approx(24, abs=0.01)

    # expires_at trong response là ISO AWARE UTC — FE không phải đoán múi giờ
    exp = datetime.fromisoformat(r.json()["expires_at"])
    assert exp.tzinfo is not None
    assert exp.utcoffset().total_seconds() == 0
    now = datetime.now(timezone.utc)
    assert (exp - now).total_seconds() == pytest.approx(24 * 3600, abs=60)


def test_qr_moi_lan_quet_mot_session_moi(client, db_mem):
    tenant_id, table_id = _seed_table()
    token = qr.make_table_token(tenant_id, table_id)
    s1 = client.get(f"/api/table-qr/{token}").json()["session_id"]
    s2 = client.get(f"/api/table-qr/{token}").json()["session_id"]
    assert s1 != s2
    with db.session() as s:
        n = s.scalar(sa.select(sa.func.count()).select_from(db.GuestSession))
    assert n == 2


def test_qr_token_rac_400_message_viet(client, db_mem):
    r = client.get("/api/table-qr/khong-phai-token")
    assert r.status_code == 400
    detail = r.json()["detail"]
    assert isinstance(detail, str)
    assert "Mã QR không hợp lệ" in detail


def test_qr_token_ky_secret_khac_400(client, db_mem, monkeypatch):
    """Đổi POOLCOACH_SECRET là mọi token cũ chết ngay (thu hồi QR) — secret
    phải được đọc MỖI request, không cache lúc import."""
    tenant_id, table_id = _seed_table()
    token = qr.make_table_token(tenant_id, table_id)     # ký secret dev
    monkeypatch.setenv("POOLCOACH_SECRET", "secret-moi-sau-khi-thu-hoi")
    assert client.get(f"/api/table-qr/{token}").status_code == 400


def test_qr_ban_khong_ton_tai_404(client, db_mem):
    tenant_id, _ = _seed_table()
    r = client.get(f"/api/table-qr/{qr.make_table_token(tenant_id, 999)}")
    assert r.status_code == 404
    assert "không còn tồn tại" in r.json()["detail"]


def test_qr_ban_cua_tenant_khac_404(client, db_mem):
    """Token ghép tenant A với bàn của tenant B phải 404 — không lộ bàn
    xuyên tenant dù chữ ký hợp lệ."""
    _seed_table()                                        # tenant 1, bàn 1
    with db.session() as s:
        t2 = db.Tenant(name="tenant-khac")
        s.add(t2)
        s.flush()
        t2_id = t2.id
    r = client.get(f"/api/table-qr/{qr.make_table_token(t2_id, 1)}")
    assert r.status_code == 404


def test_qr_khong_db_503(client, no_db):
    tenant_id, table_id = 1, 1
    r = client.get(f"/api/table-qr/{qr.make_table_token(tenant_id, table_id)}")
    assert r.status_code == 503
    assert "chưa cấu hình cơ sở dữ liệu" in r.json()["detail"].lower()


# ------------------------------------------- logging /api/recommend (Bước 3)

def _rows():
    with db.session() as s:
        return s.scalars(sa.select(db.Recommendation)
                         .order_by(db.Recommendation.id)).all()


def test_recommend_co_scan_id_edited_ghi_dung_row(client, db_mem, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    r = client.post("/api/recommend", json={
        "balls": BALLS_FULL, "scan_id": "scan-123", "edited": True})
    assert r.status_code == 200

    rows = _rows()
    assert len(rows) == 1
    row = rows[0]
    assert (row.scan_id, row.edited) == ("scan-123", True)
    assert row.engine_ver == "zoneplanner-v2"
    assert row.latency_ms >= 1
    assert row.balls["cue"] == BALLS_FULL["cue"]
    assert row.shot["rank"] == 1 and row.shot["phi"] == 45.0
    assert len(row.alternatives) == 1 and row.alternatives[0]["rank"] == 2
    # quỹ đạo là dữ liệu vẽ, không vào log (sim tất định dựng lại được)
    assert "trajectories" not in row.shot


def test_recommend_khong_scan_id_van_ghi_row_null(client, db_mem, monkeypatch):
    """Client cũ (không biết scan_id/edited tồn tại) vẫn 200 và vẫn được log
    — hai cột để NULL, không bịa default."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 200
    rows = _rows()
    assert len(rows) == 1
    assert (rows[0].scan_id, rows[0].edited) == (None, None)


def test_recommend_het_duong_log_shot_null(client, db_mem, monkeypatch):
    patch_v2(monkeypatch, v2_result([], n_legal_pot=0))
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 200
    rows = _rows()
    assert rows[0].shot is None and rows[0].alternatives == []


def test_recommend_response_khong_doi_mot_byte_khi_co_db(client, db_mem,
                                                         monkeypatch):
    """Cùng request, DB bật hay tắt — body trả về phải BẰNG NHAU TỪNG BYTE.
    Đây là câu 'Response KHÔNG đổi một byte' của BRIEF, đo đúng nghĩa đen."""
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    body = {"balls": BALLS_FULL, "scan_id": "scan-123", "edited": False}
    with_db = client.post("/api/recommend", json=body).content
    db.teardown()
    without_db = client.post("/api/recommend", json=body).content
    assert with_db == without_db


def test_recommend_khong_db_van_200_khong_log(client, no_db, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend", json={
        "balls": BALLS_FULL, "scan_id": "scan-123", "edited": True})
    assert r.status_code == 200
    assert r.json()["shots"][0]["phi"] == 45.0


def test_recommend_db_hong_van_200(client, db_mem, monkeypatch, capfd):
    """Best-effort đúng nghĩa: DB nổ giữa chừng cũng chỉ được phép thành
    warning, response vẫn 200 nguyên vẹn."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))

    def boom():
        raise RuntimeError("DB chet giua chung")

    monkeypatch.setattr(db, "session", boom)
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 200
    assert "khong ghi duoc log recommend" in capfd.readouterr().out


# ------------------------------- session_id trong recommend (bàn giao 12)
# `session_id` là mở rộng thuần phần GHI LOG như scan_id/edited: session
# sống → row recommendations mang tenant_id của quán; session rác/hết hạn
# → bỏ qua + warning; response KHÔNG đổi một byte trong mọi nhánh.

def _seed_session(hours=1.0):
    """Tenant + bàn + guest session còn `hours` giờ hạn (âm = đã hết hạn)."""
    tenant_id, table_id = _seed_table()
    with db.session() as s:
        gs = db.GuestSession(tenant_id=tenant_id, table_id=table_id,
                             expires_at=db.utcnow() + timedelta(hours=hours))
        s.add(gs)
        s.flush()
        return tenant_id, gs.id


def test_recommend_session_that_dien_tenant_vao_row(client, db_mem,
                                                    monkeypatch):
    tenant_id, session_id = _seed_session()
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend",
                    json={"balls": BALLS_FULL, "session_id": session_id})
    assert r.status_code == 200
    rows = _rows()
    assert len(rows) == 1
    assert rows[0].tenant_id == tenant_id


def test_recommend_khong_session_row_tenant_null(client, db_mem, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    assert client.post("/api/recommend",
                       json={"balls": BALLS_FULL}).status_code == 200
    assert _rows()[0].tenant_id is None


def test_recommend_session_rac_van_200_row_nac_danh(client, db_mem,
                                                    monkeypatch, capfd):
    """Session không tồn tại KHÔNG chặn request, KHÔNG mất row — chỉ mất
    tenant (NULL) và một warning cho dev."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend",
                    json={"balls": BALLS_FULL, "session_id": "khong-co-that"})
    assert r.status_code == 200
    rows = _rows()
    assert len(rows) == 1 and rows[0].tenant_id is None
    assert "session_id la/het han" in capfd.readouterr().out


def test_recommend_session_het_han_bo_qua_nhu_rac(client, db_mem,
                                                  monkeypatch, capfd):
    """Hết hạn TTL 24h là hết tư cách gắn tenant — cùng nhánh với session
    rác, không phải lỗi 4xx (phiên chỉ thuộc phần log)."""
    _, session_id = _seed_session(hours=-1)
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend",
                    json={"balls": BALLS_FULL, "session_id": session_id})
    assert r.status_code == 200
    assert _rows()[0].tenant_id is None
    assert "session_id la/het han" in capfd.readouterr().out


def test_recommend_response_khong_doi_mot_byte_vi_session(client, db_mem,
                                                          monkeypatch):
    """Bất biến bàn giao 10 mở rộng cho session_id: cùng thế bàn, có phiên
    hay không — body trả về BẰNG NHAU TỪNG BYTE."""
    _, session_id = _seed_session()
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    with_s = client.post("/api/recommend", json={
        "balls": BALLS_FULL, "session_id": session_id}).content
    without_s = client.post("/api/recommend",
                            json={"balls": BALLS_FULL}).content
    assert with_s == without_s


def test_recommend_khong_db_mang_session_van_200(client, no_db, monkeypatch):
    """Degraded (Space không DATABASE_URL): client vẫn gửi session_id —
    request phải 200 y hệt, không log, không lỗi."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    r = client.post("/api/recommend", json={
        "balls": BALLS_FULL, "session_id": "phien-nao-do"})
    assert r.status_code == 200
    assert r.json()["shots"][0]["phi"] == 45.0