tests/test_displays.py

"""Contract TV tại bàn (bàn giao 15, 05/08/2026) — WS /display/{id}, pair,
push sau recommend.

Điểm KHÁC các file test cũ: client ở đây dùng ``with TestClient(...)`` —
context manager cho mọi request HTTP lẫn WS session dùng CHUNG một portal
(một event loop). Không có nó thì mỗi request một loop riêng, và push từ
route ``/api/recommend`` (loop A) sang WS session (loop B) là undefined
behaviour chỉ-có-trong-test. Đổi lại phải vô hiệu ``_boot`` trước khi enter
(context manager chạy lifespan → thread warmup sẽ đụng pooltool stub và set
``boot_error`` giữa chừng — race với monkeypatch).

Engine vẫn mock như mọi test app: kiểm CONTRACT (hello/paired/payload/
status), 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.
"""

from __future__ import annotations

import re

import pytest
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect

from app import displays, jobqueue
from app import main as app_main
from test_api_v2 import BALLS_FULL, BALLS_LAST, make_shot, patch_v2, v2_result
from test_queue import FakeQueue, start_worker


@pytest.fixture(autouse=True)
def _reset():
    """Hub + transport về trạng thái sạch sau MỖI test — registry là state
    module-level, test sau không được thấy display/pair sót lại (nếp
    _reset_jobqueue của test_queue)."""
    yield
    displays.hub.reset()
    jobqueue.teardown()


@pytest.fixture
def client(monkeypatch, env_stub):
    # DATABASE_URL/REDIS_URL của máy dev không được lọt vào lifespan —
    # test displays chạy degraded (không DB, transport inprocess) như Space.
    monkeypatch.delenv("DATABASE_URL", raising=False)
    monkeypatch.delenv("REDIS_URL", raising=False)
    monkeypatch.setattr(app_main, "_boot", lambda: None)
    with TestClient(app_main.app) as c:
        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)
        yield c


# ------------------------------------------------------- WS hello + registry

def test_ws_hello_mang_ma_pair_6_so(client):
    with client.websocket_connect("/display/tv-a") as ws:
        hello = ws.receive_json()
    assert hello["type"] == "hello"
    assert re.fullmatch(r"\d{6}", hello["code"])
    assert hello["paired"] is False


def test_ws_reconnect_cung_display_id_giu_nguyen_ma(client):
    """TV reload giữ display_id trong sessionStorage → reconnect phải thấy
    ĐÚNG mã cũ, người đang gõ dở mã trên điện thoại không bị đổi bài."""
    with client.websocket_connect("/display/tv-a") as ws:
        code1 = ws.receive_json()["code"]
    with client.websocket_connect("/display/tv-a") as ws:
        code2 = ws.receive_json()["code"]
    assert code1 == code2


def test_ws_hai_display_hai_ma_khac_nhau(client):
    with client.websocket_connect("/display/tv-a") as wa:
        with client.websocket_connect("/display/tv-b") as wb:
            ca = wa.receive_json()["code"]
            cb = wb.receive_json()["code"]
    assert ca != cb


def test_ws_rot_giu_entry_va_ma(client):
    """Disconnect chỉ buông ws — entry (mã + pairing) sống tới khi server
    restart, đúng quyết định MVP 'pair mất khi restart là chấp nhận được'."""
    with client.websocket_connect("/display/tv-a") as ws:
        code = ws.receive_json()["code"]
    d = displays.hub._displays["tv-a"]
    assert d.ws is None
    assert d.code == code


# ------------------------------------------------- POST /api/displays/pair

def test_pair_dung_ma_tv_nhan_paired(client):
    with client.websocket_connect("/display/tv-a") as ws:
        code = ws.receive_json()["code"]
        r = client.post("/api/displays/pair",
                        json={"code": code, "session_id": "phien-1"})
        assert r.status_code == 200
        assert r.json() == {"paired": True}
        assert ws.receive_json() == {"type": "paired"}
    assert displays.hub._displays["tv-a"].session_id == "phien-1"


def test_pair_sai_ma_404_message_tieng_viet(client):
    with client.websocket_connect("/display/tv-a") as ws:
        ws.receive_json()
        r = client.post("/api/displays/pair",
                        json={"code": "000000", "session_id": "phien-1"})
    assert r.status_code == 404
    detail = r.json()["detail"]
    assert isinstance(detail, str)          # FE toast chỉ đọc được string
    assert "Mã ghép không đúng" in detail
    assert "TV" in detail


def test_pair_thieu_session_id_422(client):
    r = client.post("/api/displays/pair", json={"code": "123456"})
    assert r.status_code == 422


def test_pair_khi_tv_dang_rot_van_duoc_reconnect_thay_paired(client):
    """TV rớt đúng lúc người chơi gõ mã: entry còn → pair vẫn 200, và hello
    lúc TV reconnect phải mang paired=True (không cần gửi bù message nào)."""
    with client.websocket_connect("/display/tv-a") as ws:
        code = ws.receive_json()["code"]
    r = client.post("/api/displays/pair",
                    json={"code": code, "session_id": "phien-1"})
    assert r.status_code == 200
    with client.websocket_connect("/display/tv-a") as ws:
        hello = ws.receive_json()
    assert hello["code"] == code
    assert hello["paired"] is True


def test_pair_lai_phien_khac_de_len(client):
    """Bàn đổi khách: pair lần nữa với phiên mới — phiên mới thắng."""
    with client.websocket_connect("/display/tv-a") as ws:
        code = ws.receive_json()["code"]
        client.post("/api/displays/pair",
                    json={"code": code, "session_id": "phien-1"})
        ws.receive_json()                       # paired lần 1
        client.post("/api/displays/pair",
                    json={"code": code, "session_id": "phien-2"})
        ws.receive_json()                       # paired lần 2
    assert displays.hub._displays["tv-a"].session_id == "phien-2"


# ------------------------------------------- push sau recommend (Bước 3, G3)

def _pair(client, ws, session_id):
    """Đọc hello + pair + đọc báo paired — WS sạch message, sẵn chờ push."""
    code = ws.receive_json()["code"]
    r = client.post("/api/displays/pair",
                    json={"code": code, "session_id": session_id})
    assert r.status_code == 200
    assert ws.receive_json() == {"type": "paired"}


def test_push_inprocess_tv_nhan_dung_payload(client, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-1")
        r = client.post("/api/recommend",
                        json={"balls": BALLS_FULL, "session_id": "phien-1"})
        assert r.status_code == 200
        msg = ws.receive_json()
    assert msg["type"] == "recommend"
    assert msg["balls"] == BALLS_FULL
    assert msg["result"] == r.json()    # TV thấy ĐÚNG cái điện thoại thấy
    assert msg["ts"]                    # timestamp có mặt (format tự do)


def test_push_qua_queue_transport(client, env_stub, monkeypatch):
    """G3 — push phải chạy Y HỆT khi recommend đi qua worker Redis: push nằm
    ở API layer SAU khi có kết quả, transport nào cũng đi qua nó."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    q = FakeQueue()
    jobqueue.setup(client=q.api)
    start_worker(q, env_stub)
    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-1")
        r = client.post("/api/recommend",
                        json={"balls": BALLS_FULL, "session_id": "phien-1"})
        assert r.status_code == 200
        msg = ws.receive_json()
    assert msg["type"] == "recommend"
    assert msg["result"] == r.json()


def test_g2_response_byte_equal_co_khong_tv(client, monkeypatch):
    """G2 — bất biến recommend: body CÓ TV ghép bằng TỪNG BYTE body KHÔNG TV
    (mock giữ elapsed_s cố định nên so thẳng content, cùng kỹ thuật với bất
    biến hai mode của bàn giao 14)."""
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    body = {"balls": BALLS_FULL, "alternatives": 3, "session_id": "phien-1"}

    khong_tv = client.post("/api/recommend", json=body)
    assert khong_tv.status_code == 200          # session có, TV chưa ghép

    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-1")
        co_tv = client.post("/api/recommend", json=body)
        assert co_tv.status_code == 200
        ws.receive_json()                       # push ĐÃ tới TV
    assert co_tv.content == khong_tv.content


def test_tv_rot_recommend_van_200(client, monkeypatch):
    """G3 — TV rớt giữa chừng (đã ghép, socket đóng): recommend vẫn 200,
    không exception, không chậm — best-effort bỏ qua lặng lẽ."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-1")
    r = client.post("/api/recommend",
                    json={"balls": BALLS_FULL, "session_id": "phien-1"})
    assert r.status_code == 200
    assert len(r.json()["shots"]) == 1


def test_push_chi_toi_phien_da_ghep(client, monkeypatch):
    """Recommend của phiên KHÁC không được lọt sang TV. Chứng minh không cần
    chờ timeout: bắn phiên lạ trước, phiên mình sau — message ĐẦU TIÊN TV
    nhận phải là của phiên mình (thế bàn khác nhau làm chứng)."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-A")
        client.post("/api/recommend",
                    json={"balls": BALLS_FULL, "session_id": "phien-B"})
        client.post("/api/recommend",
                    json={"balls": BALLS_LAST, "session_id": "phien-A"})
        msg = ws.receive_json()
    assert msg["balls"] == BALLS_LAST


# ------------------------------------------- cap + TTL registry (bàn giao 20)

def _clock(monkeypatch):
    """Đồng hồ giả tiêm vào ``displays._now`` — thời gian đi bằng monkeypatch,
    không sleep (BRIEF 3.4). Registry stamp ``last_seen_at`` qua ``_now()``
    lúc gọi nên mọi mốc trong test đều đọc từ đây."""
    state = {"t": 1_000_000.0}
    monkeypatch.setattr(displays, "_now", lambda: state["t"])
    return state


def test_qua_ttl_mat_ma_va_ghep_lai_duoc(client, monkeypatch):
    """TV rớt quá TTL → purge lazy ở register kế tiếp: reconnect được cấp
    entry MỚI (pairing cũ mất — "TV rớt quá 1h mất mã" là hành vi chấp nhận)
    và ghép lại được NGAY bằng mã mới."""
    clock = _clock(monkeypatch)
    with client.websocket_connect("/display/tv-a") as ws:
        code1 = ws.receive_json()["code"]
        r = client.post("/api/displays/pair",
                        json={"code": code1, "session_id": "phien-1"})
        assert r.status_code == 200
        ws.receive_json()                            # báo paired
    clock["t"] += displays.DISPLAY_TTL_S + 1         # rớt quá 60 phút
    with client.websocket_connect("/display/tv-a") as ws:
        hello = ws.receive_json()
        assert hello["paired"] is False              # entry cũ đã bị purge
        r = client.post("/api/displays/pair",
                        json={"code": hello["code"], "session_id": "phien-2"})
        assert r.status_code == 200
        assert ws.receive_json() == {"type": "paired"}
    assert displays.hub._displays["tv-a"].session_id == "phien-2"


def test_duoi_ttl_entry_rot_khong_bi_purge(client, monkeypatch):
    """Rớt CHƯA quá TTL → entry (mã + pairing) còn nguyên qua lần register
    kế — hành vi giữ-mã của bàn giao 15 không đổi trong hạn."""
    clock = _clock(monkeypatch)
    with client.websocket_connect("/display/tv-a") as ws:
        code1 = ws.receive_json()["code"]
    clock["t"] += displays.DISPLAY_TTL_S - 60        # 59 phút — còn hạn
    with client.websocket_connect("/display/tv-b") as ws:   # trigger purge
        ws.receive_json()
    d = displays.hub._displays.get("tv-a")
    assert d is not None
    assert d.code == code1


def test_entry_dang_ghep_dang_ket_noi_khong_bao_gio_purge(client, monkeypatch):
    """Entry đang ghép + đang KẾT NỐI không bao giờ bị purge dù bao lâu trôi
    qua — TTL chỉ áp cho entry rớt (ws=None); TV nối liền nhiều giờ vẫn nhận
    push bình thường."""
    clock = _clock(monkeypatch)
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    with client.websocket_connect("/display/tv-a") as ws:
        _pair(client, ws, "phien-1")
        clock["t"] += displays.DISPLAY_TTL_S * 10    # 10 tiếng trôi qua
        with client.websocket_connect("/display/tv-b") as wb:  # trigger purge
            wb.receive_json()
        d = displays.hub._displays.get("tv-a")
        assert d is not None
        assert d.session_id == "phien-1"
        r = client.post("/api/recommend",
                        json={"balls": BALLS_FULL, "session_id": "phien-1"})
        assert r.status_code == 200
        assert ws.receive_json()["type"] == "recommend"


def test_day_cap_toan_song_tu_choi_sach(client, monkeypatch):
    """Đầy MAX_DISPLAYS toàn kết nối sống → TV mới bị accept-rồi-đóng
    (1013 Try Again Later), KHÔNG cấp entry; TV đang nối không suy suyển.
    Cap thu nhỏ bằng monkeypatch cho test nhẹ — register đọc hằng số module
    lúc gọi nên setattr là đủ."""
    monkeypatch.setattr(displays, "MAX_DISPLAYS", 2)
    with client.websocket_connect("/display/tv-a") as wa:
        wa.receive_json()
        with client.websocket_connect("/display/tv-b") as wb:
            wb.receive_json()
            with client.websocket_connect("/display/tv-c") as wc:
                with pytest.raises(WebSocketDisconnect) as exc:
                    wc.receive_json()            # không hello — đóng ngay
            assert exc.value.code == 1013
            assert set(displays.hub._displays) == {"tv-a", "tv-b"}


def test_day_cap_xoa_entry_rot_cu_nhat(client, monkeypatch):
    """Đầy cap nhưng còn entry rớt → TV mới được chỗ bằng cách xoá entry
    ws=None có last_seen_at CŨ NHẤT — đúng con cũ nhất, không phải con rớt
    gần đây."""
    clock = _clock(monkeypatch)
    monkeypatch.setattr(displays, "MAX_DISPLAYS", 3)
    with client.websocket_connect("/display/tv-old") as ws:
        ws.receive_json()
    clock["t"] += 100
    with client.websocket_connect("/display/tv-mid") as ws:
        ws.receive_json()
    clock["t"] += 100
    with client.websocket_connect("/display/tv-live") as wl:
        wl.receive_json()
        with client.websocket_connect("/display/tv-new") as wn:
            assert wn.receive_json()["type"] == "hello"   # được entry thường
            assert "tv-old" not in displays.hub._displays
            assert "tv-mid" in displays.hub._displays
            assert "tv-live" in displays.hub._displays


def test_reconnect_khi_day_cap_van_giu_ma(client, monkeypatch):
    """TV ĐÃ CÓ entry mở connection mới lúc registry đầy toàn sống → vẫn vào
    bình thường và giữ nguyên mã (chỉ TV MỚI mới cần xin chỗ)."""
    monkeypatch.setattr(displays, "MAX_DISPLAYS", 2)
    with client.websocket_connect("/display/tv-a") as wa:
        code_a = wa.receive_json()["code"]
        with client.websocket_connect("/display/tv-b") as wb:
            wb.receive_json()
            with client.websocket_connect("/display/tv-a") as wa2:
                assert wa2.receive_json()["code"] == code_a


# --------------------------------------------------- trang /tv (Bước 4)

def test_trang_tv_duoc_serve_va_khong_cache(client):
    """`/tv` đi qua mount static html=True (redirect sang `/tv/` rồi trả
    `tv/index.html`) và PHẢI mang no-cache như mọi html/js của app — bug
    cache 30/07 không được tái diễn trên trang mới."""
    r = client.get("/tv")           # TestClient tự follow redirect
    assert r.status_code == 200
    assert "PoolCoach TV" in r.text
    assert r.headers["cache-control"] == "no-cache"