tests/test_scan.py

"""Contract /api/scan (bàn giao 18, 05/08/2026) — fakeredis, KHÔNG cần
Docker/Redis/torch (Redis + model thật chỉ cần cho gate G3/G4 chạy tay).

Worker CV giả trong test chạy ĐÚNG code path transport của worker thật —
`jobqueue.serve_one(jobs_key=SCAN_JOBS_KEY)` — chỉ khác handler: trả reply
canned thay vì YOLO (venv app không torch là QUYẾT ĐỊNH, bàn giao 17; logic
detect+homography đo tay ở gate G3). Handler capture payload để khoá luôn
hợp đồng route → worker: image_b64 round-trip đúng bytes, corners đúng shape.

Route /api/scan không đụng engine/JIT — client không cần env_stub/jit_ready.
"""

from __future__ import annotations

import base64
import threading
import time

import pytest
from fastapi.testclient import TestClient

from app import jobqueue
from app import main as app_main

IMG_BYTES = b"\xff\xd8\xe0fake-jpeg-bytes"          # nội dung không cần là JPEG
CORNERS = "[[10, 20], [640, 22], [600, 300], [40, 310]]"
SCAN_RESULT = {"balls": [
    {"x": 0.45, "y": 0.30, "type": "cue", "conf": 0.98},
    {"x": 0.32, "y": 1.09, "type": "ball", "conf": 0.91},
], "dropped": {"dot": 3, "ngoai_ban": 0, "trung_tam": 0}}


@pytest.fixture(autouse=True)
def _reset_jobqueue():
    """Nếp test_queue: xong test nào transport cũng về inprocess."""
    yield
    jobqueue.teardown()


@pytest.fixture
def client():
    return TestClient(app_main.app)


@pytest.fixture
def fq():
    from test_queue import FakeQueue

    q = FakeQueue()
    jobqueue.setup(client=q.api)
    return q


def start_cv_worker(fq, reply=None, n_jobs=1, seen=None):
    """Worker CV giả — serve_one trên SCAN_JOBS_KEY như worker thật.

    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 19 route
    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"."""
    jobqueue.beat(client=fq.worker, key=jobqueue.SCAN_HEARTBEAT_KEY)

    def handler(payload):
        if seen is not None:
            seen.update(payload)
        return reply if reply is not None else {"ok": True,
                                                "result": SCAN_RESULT}

    def run():
        done = 0
        tries = 0
        while done < n_jobs and tries < 50:
            tries += 1
            if jobqueue.serve_one(handler, 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 post_scan(client, image=IMG_BYTES, corners=CORNERS):
    return client.post("/api/scan",
                       files={"image": ("ban.jpg", image, "image/jpeg")},
                       data={"corners": corners})


# ------------------------------------------------------------ G2: degraded

def test_khong_redis_url_503_message_ro(client):
    """Không REDIS_URL (mode inprocess) → 503 nói thẳng cần CV worker —
    KHÔNG có đường in-process để fallback (app không torch, bàn giao 17)."""
    jobqueue.teardown()
    r = post_scan(client)
    assert r.status_code == 503
    assert "CV worker" in r.json()["detail"]


def test_redis_chet_503_message(client, fq):
    fq.server.connected = False
    r = post_scan(client)
    assert r.status_code == 503
    assert "Redis" in r.json()["detail"]


def test_worker_chet_timeout_503(client, fq, monkeypatch):
    """Worker không trả lời → 503 sau timeout, message chỉ đích danh
    cv_worker (G3c bản test). Từ bàn giao 19: heartbeat phải CÒN SỐNG để đi
    đúng đường timeout (= worker chết SAU enqueue, trong cửa sổ TTL 15s);
    vắng heartbeat là đường fast-fail 503-ngay — test riêng bên dưới."""
    jobqueue.beat(client=fq.worker, key=jobqueue.SCAN_HEARTBEAT_KEY)
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")
    r = post_scan(client)
    assert r.status_code == 503
    assert "CV worker" in r.json()["detail"]
    assert "cv_worker" in r.json()["detail"]


# ----------------------------------------------------------- happy path

def test_scan_tra_scan_id_va_balls(client, fq):
    seen = {}
    start_cv_worker(fq, seen=seen)
    r = post_scan(client)
    assert r.status_code == 200
    data = r.json()
    # scan_id uuid hex 32 ký tự — chuỗi mờ, không bảng scans (BRIEF)
    assert len(data["scan_id"]) == 32
    assert int(data["scan_id"], 16) is not None
    assert data["balls"] == SCAN_RESULT["balls"]
    # hợp đồng route → worker: ảnh round-trip đúng BYTES, corners đúng số
    assert base64.b64decode(seen["image_b64"]) == IMG_BYTES
    assert seen["corners"] == [[10, 20], [640, 22], [600, 300], [40, 310]]


def test_hai_scan_id_khac_nhau(client, fq):
    """scan_id sinh mỗi request — hai lần scan cùng ảnh vẫn là hai scan."""
    start_cv_worker(fq, n_jobs=2)
    a = post_scan(client).json()["scan_id"]
    b = post_scan(client).json()["scan_id"]
    assert a != b


# ------------------------------------------------- lỗi từ worker / input

def test_worker_validation_422_message_nguyen_van(client, fq):
    msg = "corners phải là 4 điểm [x, y] pixel — chấm đủ 4 góc bàn."
    start_cv_worker(fq, reply={"ok": False, "error": "validation",
                               "message": msg})
    r = post_scan(client)
    assert r.status_code == 422
    assert r.json()["detail"] == msg


def test_worker_loi_noi_bo_500(client, fq):
    start_cv_worker(fq, reply={"ok": False, "error": "internal",
                               "message": "RuntimeError: no bung gia lap"})
    r = post_scan(client)
    assert r.status_code == 500
    assert "CV worker gặp lỗi" in r.json()["detail"]
    assert "RuntimeError" in r.json()["detail"]


@pytest.mark.parametrize("bad", [
    "khong-phai-json",
    "[[1,2],[3,4],[5,6]]",                       # 3 điểm
    "[[1,2],[3,4],[5,6],[7]]",                   # điểm thiếu toạ độ
    '[[1,2],[3,4],[5,6],["a",8]]',               # toạ độ không phải số
    '{"c": 4}',                                  # không phải list
])
def test_corners_hong_422_khong_cham_queue(client, fq, bad):
    """Shape corners route tự chặn — 422 NGAY, không đẩy job rác vào queue."""
    r = post_scan(client, corners=bad)
    assert r.status_code == 422
    assert "corners" in r.json()["detail"]
    assert fq.api.llen(jobqueue.SCAN_JOBS_KEY) == 0


def test_anh_rong_422(client, fq):
    r = post_scan(client, image=b"")
    assert r.status_code == 422
    assert "rỗng" in r.json()["detail"]


def test_anh_qua_lon_422_khong_cham_queue(client, fq):
    """Trần 10MB chặn từ cửa — payload to là chuyện phải dừng hỏi (BRIEF),
    không được để nó thành job Redis rồi mới vỡ."""
    r = post_scan(client, image=b"x" * (app_main.SCAN_MAX_UPLOAD_B + 1))
    assert r.status_code == 422
    assert "quá lớn" in r.json()["detail"]
    assert fq.api.llen(jobqueue.SCAN_JOBS_KEY) == 0


def test_thieu_file_hoac_corners_422(client, fq):
    """Multipart thiếu trường — 422 của FastAPI (detail list là đủ, FE scan
    tự gửi đủ trường; test chỉ khoá status không phải 500)."""
    r = client.post("/api/scan", data={"corners": CORNERS})
    assert r.status_code == 422
    r2 = client.post("/api/scan",
                     files={"image": ("a.jpg", IMG_BYTES, "image/jpeg")})
    assert r2.status_code == 422


# ------------------------------------- số bi BallID (bàn giao 22, G3)

SCAN_RESULT_NUM = {"balls": [
    {"x": 0.45, "y": 0.30, "type": "cue", "conf": 0.98},
    {"x": 0.32, "y": 1.09, "type": "ball", "conf": 0.91,
     "number": 9, "number_conf": 0.83, "wb": True},
    {"x": 0.62, "y": 0.40, "type": "ball", "conf": 0.88,
     "number": None, "number_conf": 0.0, "wb": True},
], "dropped": {"dot": 0, "ngoai_ban": 0, "trung_tam": 0}}


def test_scan_number_truyen_qua_nguyen_ven(client, fq):
    """Worker mới trả number/number_conf/wb → API truyền qua NGUYÊN VẸN,
    kể cả number=null tường minh (FE cần null để biết đường fallback);
    bi cue không gửi ba trường đó thì response cũng không mọc key."""
    start_cv_worker(fq, reply={"ok": True, "result": SCAN_RESULT_NUM})
    r = post_scan(client)
    assert r.status_code == 200
    assert r.json()["balls"] == SCAN_RESULT_NUM["balls"]


def test_worker_cu_khong_number_van_200_khong_truong_la(client, fq):
    """Worker cũ (reply không có number) → 200 và balls giữ NGUYÊN 4 key
    cũ — lùi tương thích hai chiều: client cũ lẫn worker cũ đều sống."""
    start_cv_worker(fq)                      # SCAN_RESULT kiểu cũ ở đầu file
    r = post_scan(client)
    assert r.status_code == 200
    for b in r.json()["balls"]:
        assert set(b) == {"x", "y", "type", "conf"}


# ------------------------------------ fast-fail heartbeat (bàn giao 19)

def test_vang_heartbeat_503_ngay_khong_doi_timeout(client, fq):
    """Vắng heartbeat CV worker → 503 NGAY với 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)."""
    t0 = time.perf_counter()
    r = post_scan(client)
    elapsed = time.perf_counter() - t0
    assert r.status_code == 503
    assert "CV worker không chạy" in r.json()["detail"]
    assert "cv_worker" in r.json()["detail"]
    assert elapsed < 2.0
    assert fq.api.llen(jobqueue.SCAN_JOBS_KEY) == 0   # không đẩy job


def test_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ũ giữ nguyên: 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, key=jobqueue.SCAN_HEARTBEAT_KEY)
    monkeypatch.setenv("POOLCOACH_QUEUE_TIMEOUT_S", "0.2")
    r = post_scan(client)
    assert r.status_code == 503
    assert "không trả lời" in r.json()["detail"]
    assert fq.api.llen(jobqueue.SCAN_JOBS_KEY) == 1   # job đã enqueue


# ------------------------------------ probe FE #scan-go (bàn giao 26)

def test_scan_go_khong_muon_class_btn_go():
    """Probe BG26 bước 1.3 (Cowork duyệt từ câu hỏi HANDOFF 25 #3): #scan-go
    phải theo nếp .an-cta của 9fa4c18 — app.js bind recommend() cho MỌI
    .btn-go và syncButtons ép label/disabled, nút nào mượn class đó là
    "Quét thế bi" gọi ké engine chạy nền. Khoá luôn 2 nút Analyzer (9fa4c18)
    và danh sách .btn-go hợp lệ còn lại (3 nút recommend thật)."""
    import re
    from pathlib import Path

    html = (Path(__file__).resolve().parents[1]
            / "app" / "static" / "index.html").read_text(encoding="utf-8")

    def classes_of(btn_id: str) -> set[str]:
        m = re.search(rf'<button[^>]*\bid="{btn_id}"[^>]*>', html)
        assert m, f"không thấy #{btn_id} trong index.html"
        cls = re.search(r'class="([^"]*)"', m.group(0))
        return set(cls.group(1).split()) if cls else set()

    for bid in ("scan-go", "anv-pick", "an-go"):
        assert "btn-go" not in classes_of(bid), f"#{bid} mượn .btn-go"
        assert "an-cta" in classes_of(bid), f"#{bid} thiếu .an-cta"

    # .btn-go còn lại đúng 3 nút recommend thật — mọc thêm là có đứa mượn
    got = {m2.group(1) for m2 in re.finditer(
        r'<button[^>]*\bid="([^"]+)"[^>]*\bclass="[^"]*\bbtn-go\b[^"]*"', html)}
    got |= {m2.group(2) for m2 in re.finditer(
        r'<button[^>]*\bclass="([^"]*\bbtn-go\b[^"]*)"[^>]*\bid="([^"]+)"',
        html)}
    assert got == {"btn-go", "qb-go", "drill-suggest"}, got