tests/test_local_mode.py

"""Contract mode "local" + video mẫu bundled (BRIEF 14/08/2026, việc B).

Mode "local" = MỘT tiến trình cho cả app lẫn CV (Space HF): transport chạy
fakeredis trong tiến trình, CV worker là một luồng (`app/localcv.py`), còn
`/api/recommend` GIỮ in-process như bản không queue. Test ở đây khoá đúng
ba mệnh đề đó + đường đọc kết quả từ ĐĨA (thứ làm video mẫu mở được kể cả
khi tầng CV chết).

KHÔNG test luồng CV thật (cần torch/YOLO/ffmpeg) — cái đó nghiệm thu bằng
lần chạy tay trên clip thật, ghi số trong HANDOFF.
"""

from __future__ import annotations

import json

import pytest
from fastapi.testclient import TestClient

from app import analyzer_store as astore
from app import 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():
    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)


@pytest.fixture
def local_mode(monkeypatch):
    monkeypatch.setenv("POOLCOACH_LOCAL_CV", "1")
    monkeypatch.delenv("REDIS_URL", raising=False)
    assert jobqueue.setup() == "local"
    return jobqueue


# ------------------------------------------------------------ mode

def test_env_bat_thi_ra_mode_local(local_mode):
    assert jobqueue.mode() == "local"
    assert jobqueue.cv_enabled() is True
    assert jobqueue.ping() is True          # fakeredis trong tiến trình


@pytest.mark.parametrize("val", ["", "0", "false", "no"])
def test_env_tat_thi_van_inprocess(monkeypatch, val):
    monkeypatch.setenv("POOLCOACH_LOCAL_CV", val)
    monkeypatch.delenv("REDIS_URL", raising=False)
    assert jobqueue.setup() == "inprocess"
    assert jobqueue.cv_enabled() is False


def test_redis_url_van_thang_local(monkeypatch):
    """Có REDIS_URL thì vẫn là mode queue — local chỉ là đường dự phòng cho
    container một tiến trình, không được cướp cấu hình thật."""
    monkeypatch.setenv("POOLCOACH_LOCAL_CV", "1")
    monkeypatch.setenv("REDIS_URL", "redis://localhost:6379/0")
    assert jobqueue.setup() == "queue"


def test_health_khai_mode_local_va_khong_doi_engine_worker(client, local_mode):
    h = client.get("/api/health").json()
    assert h["mode"] == "local"
    # engine chạy IN-PROCESS ở mode local → không có engine worker để chết:
    # null nghĩa "không áp dụng", KHÁC False = "chết" (quy ước bàn giao 14)
    assert h["worker_alive"] is None
    assert h["redis_ok"] is True
    assert h["cv_worker_alive"] is False     # luồng nhúng chưa chạy trong test


def test_recommend_o_mode_local_van_chay_in_process(client, local_mode,
                                                    monkeypatch):
    """Bẫy của việc thêm mode: nếu route đọc "có hàng đợi ⇒ đẩy engine đi
    worker" thì tab Gợi ý trên Space sẽ 503 vì chẳng có engine worker nào."""
    patch_v2(monkeypatch, v2_result([make_shot(1)]))
    called = {}
    orig = app_main.jobqueue.submit

    def boom(*a, **k):        # gọi vào đây là đã đi nhầm đường
        called["queue"] = True
        return orig(*a, **k)

    monkeypatch.setattr(app_main.jobqueue, "submit", boom)
    r = client.post("/api/recommend", json={"balls": BALLS_FULL})
    assert r.status_code == 200
    assert "queue" not in called


# --------------------------------------------------- video mẫu bundled

def _canned_video(base, vid="a" * 32, n=2):
    d = base / vid
    d.mkdir(parents=True)
    shots = [{"idx": i, "t_start_s": i, "t_end_s": i + 2.0,
              "t_onset_s": i + 0.2, "t_settle_s": i + 1.5,
              "status": "queued", "reason": None,
              "analyze_id": astore.shot_analyze_id(vid, i),
              "thumb": f"thumb_{i:02d}.jpg"} for i in range(1, n + 1)]
    astore.write_json(astore.manifest_path(d),
                      {"status": "done", "video_id": vid, "shots": shots,
                       "duration_s": 60.0, "warnings": []})
    for i in range(1, n + 1):
        astore.write_json(astore.shot_json_path(d, i), {
            "metrics": {"v0_mps": 2.0, "phi_deg": 12.3, "n_collisions": 1},
            "spin_class": "stun", "warnings": []})
    return vid, d


def test_seed_demo_chep_va_khong_de_ban_dang_xem(tmp_path, monkeypatch):
    src = tmp_path / "src"
    vid, _ = _canned_video(src)
    monkeypatch.setattr(astore, "DEMO_SRC_DIR", src)
    base = tmp_path / "out"
    assert astore.seed_demo(base) == [vid]
    assert astore.manifest_path(base / vid).exists()

    # lần hai KHÔNG đè: người dùng có thể đang xem dở
    mine = base / vid / "shot_01.json"
    mine.write_text('{"metrics": {"v0_mps": 99.0}, "warnings": []}',
                    encoding="utf-8")
    astore.seed_demo(base)
    assert json.loads(mine.read_text(encoding="utf-8"))["metrics"]["v0_mps"] \
        == 99.0


def test_seed_demo_thieu_thu_muc_khong_nem(tmp_path, monkeypatch):
    monkeypatch.setattr(astore, "DEMO_SRC_DIR", tmp_path / "khong-co")
    assert astore.seed_demo(tmp_path / "out") == []


def test_endpoint_demo_tra_id_bundled(client, tmp_path, monkeypatch):
    src = tmp_path / "src"
    vid, _ = _canned_video(src)
    monkeypatch.setattr(astore, "DEMO_SRC_DIR", src)
    assert client.get("/api/analyzer/demo").json()["id"] == vid


def test_endpoint_demo_tra_null_khi_khong_dong_goi(client, tmp_path,
                                                   monkeypatch):
    monkeypatch.setattr(astore, "DEMO_SRC_DIR", tmp_path / "khong-co")
    assert client.get("/api/analyzer/demo").json() == {"id": None}


def test_bo_thu_muc_demo_that_trong_repo_doc_duoc():
    """Bộ mẫu đóng gói phải THẬT SỰ đọc được (manifest + đủ JSON/overlay
    per cú) — mất file lúc đóng gói thì Space mở ra là bảng trống, mà đó
    lại đúng thứ người mở link thấy đầu tiên."""
    ids = astore.demo_ids()
    if not ids:
        pytest.skip("bản này không đóng gói video mẫu")
    for vid in ids:
        d = astore.DEMO_SRC_DIR / vid
        man = astore.read_json(astore.manifest_path(d))
        assert man and man.get("status") == "done" and man.get("shots")
        for sh in man["shots"]:
            if sh.get("status") == "error":
                continue
            i = int(sh["idx"])
            assert astore.shot_json_path(d, i).exists(), (vid, i)
            assert astore.overlay_path(d, i).exists(), (vid, i)
            assert astore.thumb_path(d, i).exists(), (vid, i)


# ------------------------------------- đọc kết quả từ ĐĨA khi không queue

def test_video_co_manifest_doc_duoc_khi_khong_co_hang_doi(client, tmp_path,
                                                          monkeypatch):
    """Đĩa là nguồn sự thật (nếp BG29b/BG31): video đã phân tích xong phải
    mở được kể cả khi tầng CV tắt hẳn — đó là điều giữ cho video mẫu sống
    trên một bản deploy không có worker."""
    base = tmp_path / "an"
    vid, _ = _canned_video(base, n=3)
    monkeypatch.setattr(app_main, "_analyzer_video_dir", lambda v: base / v)
    jobqueue.teardown()                       # mode inprocess
    r = client.get("/api/analyzer/shots", params={"video": vid})
    assert r.status_code == 200
    body = r.json()
    assert body["n_total"] == 3 and body["n_done"] == 3
    assert body["shots"][0]["phi_deg"] == 12.3
    # export dùng chung nguồn → cũng phải sống
    assert client.get(f"/api/analyzer/videos/{vid}/export.csv").status_code \
        == 200


def test_khong_manifest_va_khong_hang_doi_van_503(client, tmp_path,
                                                  monkeypatch):
    """Không nới lỏng ca cũ: chưa có gì trên đĩa thì phải có hàng đợi mới
    trả lời được video đang ở đâu."""
    base = tmp_path / "an"
    monkeypatch.setattr(app_main, "_analyzer_video_dir", lambda v: base / v)
    jobqueue.teardown()
    r = client.get("/api/analyzer/shots", params={"video": "b" * 32})
    assert r.status_code == 503


def test_fe_tu_nap_video_mau_luc_mo_app():
    js = (app_main.STATIC_DIR / "app.js").read_text(encoding="utf-8")
    assert "/api/analyzer/demo" in js
    assert "loadAnvDemo()" in js


def test_fe_canh_bao_cham_chi_o_ban_cpu():
    """Cảnh báo chậm phải bám tín hiệu SERVER (health.mode == "local"), chứ
    không hiện vô điều kiện — máy Danh chạy GPU qua mode queue."""
    js = (app_main.STATIC_DIR / "app.js").read_text(encoding="utf-8")
    html = (app_main.STATIC_DIR / "index.html").read_text(encoding="utf-8")
    assert 'id="an-cpu-warn"' in html
    assert 'h.mode === "local"' in js
    assert 'S.cpuOnly' in js