"""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