tests/test_queue.py

"""Contract transport queue (bàn giao 14, 04/08/2026) — fakeredis, KHÔNG cần
Docker/Redis thật (Redis thật chỉ cần cho gate G3 chạy tay).

Worker giả trong test chạy ĐÚNG code path worker thật — `jobqueue.serve_one`
+ `engine.handle_job` — trên một FakeServer chung, API và worker mỗi bên một
client riêng như hai process thật. Chỉ khác chỗ boot: env_stub thay
PositionPlayEnv (nếp conftest — test contract, không test vật lý).

Byte-equality hai mode đo ĐÚNG NGHĨA ĐEN bằng `r.content` — cùng kỹ thuật
với bất biến DB on/off của bàn giao 10 (test_db_qr), mở rộng chiều mode.
"""

from __future__ import annotations

import threading
import time

import fakeredis
import pytest
from fastapi.testclient import TestClient

from poolcoach_rl import recommend as rec_pkg

from app import engine, jobqueue
from app import main as app_main
from test_api_v2 import BALLS_FULL, make_shot, patch_v2, v2_result


@pytest.fixture(autouse=True)
def _reset_jobqueue():
    """Mọi test trong file này xong là transport về inprocess — test file
    khác (và test sau trong file) không được thấy queue sót lại (nếp db_mem)."""
    yield
    jobqueue.teardown()


@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)


class FakeQueue:
    """FakeServer chung + hai connection tách nhau (API / worker)."""

    def __init__(self):
        self.server = fakeredis.FakeServer()
        self.api = fakeredis.FakeStrictRedis(server=self.server,
                                             decode_responses=True)
        self.worker = fakeredis.FakeStrictRedis(server=self.server,
                                                decode_responses=True)


@pytest.fixture
def fq():
    q = FakeQueue()
    jobqueue.setup(client=q.api)         # API vào mode queue
    return q


def start_worker(fq: FakeQueue, env_h, n_jobs=1) -> threading.Thread:
    """Worker giả: phục vụ đúng `n_jobs` job rồi dừng — cùng code path worker
    thật (serve_one + handle_job), chỉ khác vòng đời hữu hạn cho test.

    Beat heartbeat TRƯỚC khi phục vụ (đồng bộ, trong thread test — không
    race): worker thật cũng beat trong vòng serve, và từ bàn giao 20 route
    recommend fast-fail ĐỌC heartbeat trước enqueue — worker giả không beat
    thì mọi test đi qua queue rơi nhầm vào đường 503-ngay. Ngữ nghĩa từng
    test không đổi: vẫn là "worker sống phục vụ job" (cùng lý do
    start_cv_worker của test_scan, bàn giao 19)."""
    jobqueue.beat(client=fq.worker)

    def run():
        done = 0
        tries = 0
        while done < n_jobs and tries < 50:
            tries += 1
            if jobqueue.serve_one(lambda p: engine.handle_job(p, env_h),
                                  timeout_s=0.2, client=fq.worker):
                done += 1

    t = threading.Thread(target=run, daemon=True)
    t.start()
    return t


# ------------------------------------------------------------ mode theo env

def test_mode_khong_redis_url_la_inprocess(monkeypatch):
    """Không set REDIS_URL → mode inprocess, chạy Y HỆT hôm nay (cùng nếp
    DATABASE_URL — Space không được vỡ)."""
    monkeypatch.delenv("REDIS_URL", raising=False)
    assert jobqueue.setup() == "inprocess"
    assert jobqueue.mode() == "inprocess"


def test_mode_co_redis_url_la_queue(monkeypatch):
    """Set REDIS_URL → mode queue. Chưa có lệnh nào chạy nên URL không cần
    trỏ tới Redis sống — connect lười của redis-py."""
    monkeypatch.setenv("REDIS_URL", "redis://localhost:6399/9")
    assert jobqueue.setup() == "queue"
    assert jobqueue.mode() == "queue"


# ------------------------------------------------- transport: enqueue/reply

def test_submit_roundtrip_qua_worker_gia(fq, env_stub, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    start_worker(fq, env_stub)
    reply = jobqueue.submit({"balls": BALLS_FULL, "alternatives": 3})
    assert reply["ok"] is True
    assert reply["result"]["shots"][0]["phi"] == 45.0
    assert reply["result"]["search"]["engine"] == "zone"


def test_result_key_co_ttl_don_rac(fq, env_stub, monkeypatch):
    """Reply worker phải mang TTL — API timeout bỏ đi thì kết quả mồ côi tự
    bốc hơi, không rác vĩnh viễn trong Redis."""
    import json

    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    fq.api.lpush(jobqueue.JOBS_KEY,
                 json.dumps({"job_id": "j-ttl",
                             "payload": {"balls": BALLS_FULL,
                                         "alternatives": 3}}))
    assert jobqueue.serve_one(lambda p: engine.handle_job(p, env_stub),
                              timeout_s=0.2, client=fq.worker) is True
    key = jobqueue.RESULT_KEY.format(job_id="j-ttl")
    ttl = fq.api.ttl(key)
    assert 0 < ttl <= jobqueue.RESULT_TTL_S
    assert json.loads(fq.api.lindex(key, 0))["ok"] is True


def test_submit_khong_worker_nem_queue_timeout(fq, monkeypatch):
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")
    with pytest.raises(jobqueue.QueueTimeout):
        jobqueue.submit({"balls": BALLS_FULL, "alternatives": 3})


def test_submit_redis_chet_nem_queue_down(fq):
    fq.server.connected = False
    with pytest.raises(jobqueue.QueueDown):
        jobqueue.submit({"balls": BALLS_FULL, "alternatives": 3})


# --------------------------------------------------- API qua queue: happy path

def test_api_recommend_qua_queue_dung_shape(client, fq, env_stub, monkeypatch):
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    start_worker(fq, env_stub)
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 200
    data = r.json()
    assert len(data["shots"]) == 2
    assert data["shots"][0]["describe"].startswith("ĐÁNH BI 1 VÀO LỖ")
    assert data["search"] == {"engine": "zone", "n_sim": 99, "n_pot": 2,
                              "elapsed_s": 1.46}


def test_api_hai_mode_bang_nhau_tung_byte(client, env_stub, monkeypatch):
    """Bất biến trung tâm của bàn giao 14: cùng thế bàn + engine tất định →
    body inprocess và body queue BẰNG NHAU TỪNG BYTE (mock giữ elapsed_s cố
    định — ngoài đời elapsed_s là đồng hồ, gate G3 so phần còn lại)."""
    patch_v2(monkeypatch, v2_result([make_shot(1), make_shot(2)]))
    body = {"balls": BALLS_FULL, "alternatives": 3}

    jobqueue.teardown()                          # mode inprocess
    inproc = client.post("/api/recommend", json=body)
    assert inproc.status_code == 200

    q = FakeQueue()                              # mode queue + worker giả
    jobqueue.setup(client=q.api)
    start_worker(q, env_stub)
    queued = client.post("/api/recommend", json=body)
    assert queued.status_code == 200

    assert queued.content == inproc.content


def test_api_validation_422_cung_byte_hai_mode(client, env_stub, monkeypatch):
    """ValueError của validate_full đi qua worker phải ra ĐÚNG 422 + message
    nguyên văn như in-process — cả body lỗi cũng trùng từng byte."""
    def boom(*a, **k):
        raise ValueError("cue và 1 chồng lên nhau (< 2R)")

    monkeypatch.setattr(rec_pkg, "recommend_v2", boom)
    body = {"balls": BALLS_FULL}

    jobqueue.teardown()
    inproc = client.post("/api/recommend", json=body)
    assert inproc.status_code == 422

    q = FakeQueue()
    jobqueue.setup(client=q.api)
    start_worker(q, env_stub)
    queued = client.post("/api/recommend", json=body)
    assert queued.status_code == 422
    assert queued.content == inproc.content
    assert queued.json()["detail"] == "cue và 1 chồng lên nhau (< 2R)"


def test_api_worker_loi_noi_bo_500_worker_song_tiep(client, fq, env_stub,
                                                    monkeypatch):
    """Job hỏng → 500 message rõ, nhưng worker KHÔNG chết — request kế được
    phục vụ bình thường (handle_job nuốt exception thành reply lỗi)."""
    calls = {"n": 0}

    def flaky(*a, **k):
        calls["n"] += 1
        if calls["n"] == 1:
            raise RuntimeError("no bung gia lap")
        return v2_result([make_shot(1)])

    monkeypatch.setattr(rec_pkg, "recommend_v2", flaky)
    start_worker(fq, env_stub, n_jobs=2)

    r1 = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r1.status_code == 500
    assert "Engine gặp lỗi khi tính cú" in r1.json()["detail"]
    assert "RuntimeError" in r1.json()["detail"]

    r2 = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r2.status_code == 200


# ------------------------------------------- API qua queue: 503, không fallback

def test_api_khong_worker_503_va_khong_fallback(client, fq, monkeypatch):
    """Worker chết SAU enqueue (heartbeat còn sống trong cửa sổ TTL 15s) →
    503 message tiếng Việt trong ~timeout. TUYỆT ĐỐI không âm thầm fallback
    in-process (BRIEF 1.4) — engine local mà bị gọi là test này nổ ngay bằng
    AssertionError. Từ bàn giao 20 phải đặt heartbeat để đi đúng đường
    timeout-sau-enqueue nó vốn kiểm; vắng heartbeat là đường fast-fail
    503-ngay — test riêng bên dưới."""
    def phai_khong_duoc_goi(*a, **k):
        raise AssertionError("mode queue mà engine in-process bị gọi — "
                             "fallback lén là bug")

    monkeypatch.setattr(rec_pkg, "recommend_v2", phai_khong_duoc_goi)
    jobqueue.beat(client=fq.worker)
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")

    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 503
    detail = r.json()["detail"]
    assert "Engine không trả lời" in detail
    assert "worker" in detail


def test_api_redis_chet_503_message(client, fq, monkeypatch):
    """Redis chết → 503 message "Redis". Từ bàn giao 20 đường đi là
    `worker_alive_or_raise` ném QueueDown ngay ở check fast-fail (trước cả
    submit) — vẫn CÙNG nhánh message "Redis", không đổ oan cho worker."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    fq.server.connected = False
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 503
    assert "Redis" in r.json()["detail"]


def test_api_worker_song_lai_request_ke_ok(client, fq, env_stub, monkeypatch):
    """G3(c) bản test: worker chết → 503; worker sống lại → request kế 200
    mà KHÔNG restart API. Job kẹt lại từ lúc chết được worker mới dọn (reply
    mồ côi tự hết hạn TTL) rồi mới tới job mới. Từ bàn giao 20 phải đặt
    heartbeat trước request đầu — không thì job #1 không bao giờ được enqueue
    (fast-fail) và "job kẹt được dọn" thành kiểm chuyện không xảy ra."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    jobqueue.beat(client=fq.worker)
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")

    r1 = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r1.status_code == 503                 # job #1 kẹt trong list

    monkeypatch.delenv("POOLCOACH_QUEUE_TIMEOUT_S")
    start_worker(fq, env_stub, n_jobs=2)         # dọn job kẹt + job mới
    r2 = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r2.status_code == 200


# ------------------------------------ fast-fail heartbeat (bàn giao 20)
# Đối xứng /api/scan (test_scan, bàn giao 19): 3 đường sau fast-fail —
# (a) vắng heartbeat → 503-ngay không enqueue; (b) heartbeat sống nhưng
# worker kẹt → đường timeout cũ; (c) worker sống → byte-equal (đã khoá ở
# test_api_hai_mode_bang_nhau_tung_byte, start_worker beat sẵn heartbeat).

def test_recommend_vang_heartbeat_503_ngay_khong_doi_timeout(client, fq,
                                                             monkeypatch):
    """Vắng heartbeat engine worker → 503 NGAY message riêng, KHÔNG enqueue.

    Timeout giữ MẶC ĐỊNH 30s (cố ý không rút ngắn): nếu route lỡ đi đường
    submit thì test này mất trọn 30s — đồng hồ < 2s là bằng chứng đo được
    của "503-ngay, không đợi timeout" (gate G2). Engine in-process cũng
    không được gọi — fast-fail không phải cửa fallback lén."""
    def phai_khong_duoc_goi(*a, **k):
        raise AssertionError("fast-fail mà engine in-process bị gọi — "
                             "fallback lén là bug")

    monkeypatch.setattr(rec_pkg, "recommend_v2", phai_khong_duoc_goi)
    t0 = time.perf_counter()
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    elapsed = time.perf_counter() - t0
    assert r.status_code == 503
    assert "Engine worker không chạy" in r.json()["detail"]
    assert "engine_worker" in r.json()["detail"]
    assert elapsed < 2.0
    assert fq.api.llen(jobqueue.JOBS_KEY) == 0    # không đẩy job


def test_recommend_co_heartbeat_worker_ket_van_duong_timeout(client, fq,
                                                             monkeypatch):
    """Heartbeat sống nhưng worker không phục vụ (= chết SAU enqueue, trong
    cửa sổ TTL 15s) → vẫn đường timeout cũ: job ĐÃ vào queue, 503 message
    "không trả lời" — fast-fail không nuốt mất ca này."""
    jobqueue.beat(client=fq.worker)
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 503
    assert "không trả lời" in r.json()["detail"]
    assert fq.api.llen(jobqueue.JOBS_KEY) == 1    # job đã enqueue


# ------------------------------------------------------- health 3 trạng thái

def test_health_queue_worker_song(client, fq):
    jobqueue.beat(client=fq.worker)
    data = client.get("/api/health").json()
    assert data["mode"] == "queue"
    assert data["redis_ok"] is True
    assert data["worker_alive"] is True
    assert data["status"] == "ok"                # trường cũ không đổi


def test_health_queue_worker_chet(client, fq):
    data = client.get("/api/health").json()
    assert data["mode"] == "queue"
    assert data["redis_ok"] is True
    assert data["worker_alive"] is False


def test_health_queue_worker_thoat_sach_chet_ngay(client, fq):
    """Ctrl+C của worker xoá heartbeat — health thấy chết NGAY, không phải
    đợi TTL 15s trôi."""
    jobqueue.beat(client=fq.worker)
    jobqueue.clear_heartbeat(client=fq.worker)
    assert client.get("/api/health").json()["worker_alive"] is False


def test_health_queue_redis_chet_khong_500(client, fq):
    """Redis chết thì health TRẢ LỜI chuyện đó (redis_ok False), không 500."""
    fq.server.connected = False
    r = client.get("/api/health")
    assert r.status_code == 200
    data = r.json()
    assert data["mode"] == "queue"
    assert data["redis_ok"] is False
    assert data["worker_alive"] is False


# ------------------------------------------- queue scan (bàn giao 18, 05/08)
# Queue thứ hai TRÊN CÙNG transport: các hàm nhận jobs_key/key tuỳ chọn,
# mặc định giữ queue recommend — test khoá cả hai vế: đi đúng list mới VÀ
# không lọt sang list cũ. Worker giả trả reply canned (CV thật cần torch —
# venv app không có, đúng thiết kế 2 venv; logic detect đo ở gate G3 tay).

SCAN_REPLY = {"ok": True, "result": {"balls": [
    {"x": 0.45, "y": 0.30, "type": "cue", "conf": 0.98}]}}


def start_scan_worker(fq: FakeQueue, reply=None, n_jobs=1) -> threading.Thread:
    """Worker scan giả: cùng serve_one worker thật, chỉ khác handler canned."""
    def run():
        done = 0
        tries = 0
        while done < n_jobs and tries < 50:
            tries += 1
            if jobqueue.serve_one(lambda p: reply or SCAN_REPLY,
                                  timeout_s=0.2, client=fq.worker,
                                  jobs_key=jobqueue.SCAN_JOBS_KEY):
                done += 1

    t = threading.Thread(target=run, daemon=True)
    t.start()
    return t


def test_scan_queue_la_list_rieng(fq):
    """Job scan nằm ở pc:jobs:scan — worker recommend KHÔNG thấy nó, worker
    scan thấy và reply về result key chung giao thức."""
    import json

    fq.api.lpush(jobqueue.SCAN_JOBS_KEY,
                 json.dumps({"job_id": "j-scan", "payload": {"n": 1}}))
    assert jobqueue.serve_one(lambda p: {"ok": True}, timeout_s=0.1,
                              client=fq.worker) is False
    assert jobqueue.serve_one(lambda p: SCAN_REPLY, timeout_s=0.1,
                              client=fq.worker,
                              jobs_key=jobqueue.SCAN_JOBS_KEY) is True
    key = jobqueue.RESULT_KEY.format(job_id="j-scan")
    reply = json.loads(fq.api.lindex(key, 0))
    assert reply["ok"] is True
    assert 0 < fq.api.ttl(key) <= jobqueue.RESULT_TTL_S


def test_submit_jobs_key_scan_roundtrip(fq):
    """submit(jobs_key=SCAN_JOBS_KEY) đẩy đúng queue scan và nhận reply."""
    start_scan_worker(fq)
    reply = jobqueue.submit({"image_b64": "x", "corners": []},
                            jobs_key=jobqueue.SCAN_JOBS_KEY)
    assert reply == SCAN_REPLY


def test_heartbeat_scan_doc_lap_voi_engine(fq):
    """Heartbeat CV worker là khoá RIÊNG — beat/clear bên scan không đụng
    worker_alive() của engine và ngược lại."""
    jobqueue.beat(client=fq.worker, key=jobqueue.SCAN_HEARTBEAT_KEY)
    assert jobqueue.worker_alive() is False
    assert jobqueue.worker_alive(key=jobqueue.SCAN_HEARTBEAT_KEY) is True
    jobqueue.beat(client=fq.worker)
    jobqueue.clear_heartbeat(client=fq.worker,
                             key=jobqueue.SCAN_HEARTBEAT_KEY)
    assert jobqueue.worker_alive(key=jobqueue.SCAN_HEARTBEAT_KEY) is False
    assert jobqueue.worker_alive() is True


# ---------------------------------- health cv_worker_alive (bàn giao 19)
# CHỈ THÊM trường (nếp bàn giao 14) — shape đầy đủ khoá bằng so TẬP ở
# test_api_v2.test_g_health_shape_sau_don_dep; ở đây khoá GIÁ TRỊ theo
# trạng thái heartbeat scan, và tính độc lập với worker_alive (engine).

def test_health_cv_worker_alive_doc_lap_hai_heartbeat(client, fq):
    """cv_worker_alive đọc heartbeat SCAN — chỉ CV sống thì cv=True/engine
    =False, và ngược lại (hai worker chết/sống độc lập, bàn giao 18)."""
    jobqueue.beat(client=fq.worker, key=jobqueue.SCAN_HEARTBEAT_KEY)
    data = client.get("/api/health").json()
    assert data["cv_worker_alive"] is True
    assert data["worker_alive"] is False
    jobqueue.clear_heartbeat(client=fq.worker,
                             key=jobqueue.SCAN_HEARTBEAT_KEY)
    jobqueue.beat(client=fq.worker)
    data = client.get("/api/health").json()
    assert data["cv_worker_alive"] is False
    assert data["worker_alive"] is True


def test_health_inprocess_cv_worker_alive_null(client):
    """Mode inprocess → cv_worker_alive null "không áp dụng" (cùng quy ước
    redis_ok/worker_alive bàn giao 14), KHÔNG phải False "chết"."""
    jobqueue.teardown()
    assert client.get("/api/health").json()["cv_worker_alive"] is None


def test_health_redis_chet_cv_worker_alive_false(client, fq):
    """Redis chết → cv_worker_alive False như worker_alive, health vẫn 200
    (không 500 — redis_ok=False là câu trả lời)."""
    fq.server.connected = False
    r = client.get("/api/health")
    assert r.status_code == 200
    data = r.json()
    assert data["redis_ok"] is False
    assert data["cv_worker_alive"] is False