Spaces:
Running
Running
| """Test theo dõi người dùng: ghi log thô -> suy ra 9 chỉ số (A1-A5, B1-B4). | |
| Cô lập bằng cách trỏ STORAGE_DIR vào thư mục tạm TRƯỚC khi import storage, và | |
| ép DATABASE_URL rỗng để luôn chạy nhánh SQLite (không đụng Postgres thật). | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import tempfile | |
| import time | |
| import uuid | |
| from datetime import datetime, timedelta, timezone | |
| _TEST_STORAGE = tempfile.mkdtemp(prefix="tnt_an_") | |
| os.environ["STORAGE_DIR"] = _TEST_STORAGE | |
| from app.config import settings # noqa: E402 | |
| settings.STORAGE_DIR = _TEST_STORAGE # .env của máy load override -> ép lại | |
| settings.DATABASE_URL = "" # ép SQLite | |
| os.makedirs(_TEST_STORAGE, exist_ok=True) | |
| from app.services.analytics import etl, storage as store # noqa: E402 | |
| TZ = timezone(timedelta(hours=7)) | |
| T0 = time.time() - 3600 | |
| def _ev(off, name, step, *, feature="dub", task="task_1", sess="sess_1", | |
| user="an", lat=None, value="", **props): | |
| """Một dòng log đã chuẩn hoá, đặt tại mốc T0+off giây.""" | |
| ts = T0 + off | |
| row = store.normalize({ | |
| "event_id": uuid.uuid4().hex, | |
| "event_time": datetime.fromtimestamp(ts, TZ).isoformat(timespec="milliseconds"), | |
| "user_id": user, "session_id": sess, "task_id": task, "feature": feature, | |
| "step": step, "event_name": name, "event_value": value, | |
| "latency_ms": lat, "properties": props, | |
| }, ts) | |
| row["event_ts"] = ts # giữ nguyên trục thời gian giả lập | |
| # storage.normalize() trả dạng GHI (properties_json). Khi đọc lại từ DB, | |
| # _row_to_dict() bung thành `properties` dict — ETL dùng dạng đó. Cho dòng | |
| # mang cả hai để test được cả nhánh ghi lẫn nhánh tính chỉ số. | |
| row["properties"] = dict(props) | |
| return row | |
| def _one_full_task(): | |
| """Một task dub hoàn chỉnh: 2 lần máy, người sửa ở giữa, tải về, 👍.""" | |
| return [ | |
| _ev(0, "session_start", "open", feature="", task=""), | |
| _ev(2, "feature_open", "open"), | |
| _ev(6, "upload_done", "prepare"), | |
| _ev(9, "job_queued", "run", job_id="j1"), | |
| _ev(11, "job_started", "run", job_id="j1"), | |
| _ev(41, "job_done", "run", lat=30000, job_id="j1", phase="asr_translate"), | |
| _ev(120, "edit_translation", "review"), | |
| _ev(200, "job_queued", "run", job_id="j2"), | |
| _ev(202, "job_started", "run", job_id="j2"), | |
| _ev(260, "job_done", "run", lat=59000, job_id="j2", phase="tts"), | |
| _ev(275, "output_download", "output", value="mp4", count=2, | |
| success_type="download"), | |
| _ev(277, "thumbs_feedback", "output", value="up"), | |
| _ev(280, "feature_complete", "done", success_type="download"), | |
| ] | |
| # ── Chuẩn hoá dòng log ─────────────────────────────────────────── | |
| def test_normalize_drops_zero_latency(): | |
| """latency_ms chỉ dành cho dòng MÁY chạy: 0/rỗng phải thành None. | |
| Nhét 0 vào là kéo tụt trung bình tốc độ máy (spec §2 "quy tắc vàng"). | |
| """ | |
| assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": 0})["latency_ms"] is None | |
| assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": ""})["latency_ms"] is None | |
| assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": -5})["latency_ms"] is None | |
| assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": 1200})["latency_ms"] == 1200 | |
| def test_normalize_parses_time_and_props(): | |
| r = store.normalize({ | |
| "event_name": "feature_open", "user_id": " AN ", | |
| "event_time": "2026-08-24T09:25:09.300+07:00", | |
| "properties": {"entry": "sidebar"}, | |
| }) | |
| assert r["user_id"] == "an" # chuẩn hoá để gom đúng người | |
| assert abs(r["event_ts"] - 1787538309.3) < 1.0 # 09:25:09.300 +07:00 | |
| assert '"entry"' in r["properties_json"] | |
| # thời gian sai định dạng -> lùi về giờ máy chủ, không làm vỡ luồng | |
| bad = store.normalize({"event_name": "x", "user_id": "a", "event_time": "hôm qua"}) | |
| assert bad["event_ts"] > 0 | |
| # ── Kho lưu ────────────────────────────────────────────────────── | |
| def test_insert_query_and_dedup(): | |
| rows = _one_full_task() | |
| n = store.insert(rows) | |
| assert n == len(rows) | |
| assert store.insert(rows) == 0 # cùng event_id -> không nhân đôi | |
| got = store.query(user_id="an") | |
| assert len(got) >= len(rows) | |
| assert all(g["user_id"] == "an" for g in got) | |
| # dòng thiếu user_id thì bỏ, không rác vào bảng | |
| assert store.insert([store.normalize({"event_name": "x"})]) == 0 | |
| def test_query_newest_first_respects_limit(): | |
| """Cắt `limit` mà xếp tăng dần sẽ lấy nhầm N dòng CŨ NHẤT — màn log thô cần MỚI nhất.""" | |
| old = store.query(limit=3) | |
| new = store.query(limit=3, newest_first=True) | |
| assert len(old) == len(new) == 3 | |
| assert new[0]["server_ts"] >= old[-1]["server_ts"] | |
| assert [r["server_ts"] for r in new] == sorted([r["server_ts"] for r in new], reverse=True) | |
| def test_purge_old_keeps_recent(): | |
| before = store.stats()["rows"] | |
| assert store.purge_old(days=3650) == 0 # chưa có gì cũ tới vậy | |
| assert store.stats()["rows"] == before | |
| # ── ETL: dựng task & chỉ số ────────────────────────────────────── | |
| def test_task_metrics_from_one_session(): | |
| evs = _one_full_task() | |
| tasks = etl.build_tasks(evs) | |
| assert len(tasks) == 1 | |
| t = tasks[0] | |
| assert t["has_output"] and t["success_type"] == "download" | |
| # A1 = từ lúc mở tool tới ĐẦU RA đầu tiên (275 - 2 giây) | |
| assert abs(t["duration_ms"] - 273000) < 50 | |
| # B4 = tổng thời gian máy chạy thật, KHÔNG gồm chờ hàng đợi | |
| assert t["machine_ms"] == 89000 | |
| assert abs(t["wait_ms"] - 4000) < 50 # (11-9) + (202-200) | |
| assert t["attempts"] == 2 # A4: hai lần job_done | |
| assert t["produced"] == 2 # B1: count=2 | |
| assert t["thumbs_up"] == 1 and t["thumbs_down"] == 0 | |
| def test_step_durations_stop_at_done(): | |
| """Bấm 👍 sau khi xong mang step=output; dòng thời gian không được hiện thêm bước sau 'done'.""" | |
| steps = [d["step"] for d in etl.step_durations(etl.build_tasks(_one_full_task())[0])] | |
| assert steps[-1] == "done" | |
| assert steps.count("done") == 1 | |
| def test_abandon_and_completion_rate(): | |
| """A3 = task có đầu ra / tổng task; bỏ dở ghi rõ bước cuối chạm tới.""" | |
| evs = _one_full_task() + [ | |
| _ev(300, "feature_open", "open", feature="shuffle", task="task_2"), | |
| _ev(340, "upload_done", "prepare", feature="shuffle", task="task_2"), | |
| _ev(900, "feature_abandon", "prepare", feature="shuffle", task="task_2", | |
| last_step="prepare", reason="qua_nhieu_clip"), | |
| ] | |
| s = etl.summarize(evs) | |
| assert s["overall"]["tasks"] == 2 | |
| assert s["overall"]["tasks_done"] == 1 | |
| assert abs(s["overall"]["completion_rate"] - 0.5) < 1e-9 | |
| sh = next(f for f in s["features"] if f["feature"] == "shuffle") | |
| assert sh["abandon_by_step"] == {"prepare": 1} | |
| assert sh["label"] == "Xào video" | |
| def test_forward_counts_as_success(): | |
| """Chuyển tiếp sang tool khác cũng là ĐẦU RA hợp lệ (spec §6.2).""" | |
| evs = [ | |
| _ev(0, "feature_open", "open", feature="down_list", task="t_fw"), | |
| _ev(5, "output_forward", "output", feature="down_list", task="t_fw", | |
| value="cut", success_type="forward", to="cut", count=3), | |
| ] | |
| t = etl.build_tasks(evs)[0] | |
| assert t["has_output"] and t["success_type"] == "forward" | |
| prods = etl.summarize(evs)["products"] | |
| assert prods[0]["kind"].startswith("→") # đọc ra "chuyển sang", không lẫn đuôi file | |
| assert prods[0]["count"] == 3 | |
| def test_job_speed_split_by_phase(): | |
| """B2 phải tách tốc độ theo TỪNG BƯỚC máy, không gộp một cục.""" | |
| jobs = etl.summarize(_one_full_task())["jobs"] | |
| phases = {j["phase"]: j["median_ms"] for j in jobs} | |
| assert phases["asr_translate"] == 30000 | |
| assert phases["tts"] == 59000 | |
| def test_contribution_score_ranks_tools(): | |
| """Điểm đóng góp = 40% nhu cầu + 20% ít bấm chạy lại + 40% sản phẩm đầu ra. | |
| Tool A: 2 việc, 4 sản phẩm, mỗi việc bấm 1 lần -> phải đứng trên | |
| Tool B: 1 việc, 1 sản phẩm, bấm 2 lần/việc. | |
| """ | |
| evs = [] | |
| for i in (1, 2): # cut: 2 việc, mỗi việc 1 lần bấm, 2 sản phẩm | |
| evs += [ | |
| _ev(i * 10, "feature_open", "open", feature="cut", task=f"c{i}"), | |
| _ev(i * 10 + 2, "job_done", "run", feature="cut", task=f"c{i}", lat=1000), | |
| _ev(i * 10 + 4, "output_download", "output", feature="cut", task=f"c{i}", | |
| value="mp4", count=2), | |
| ] | |
| evs += [ # shuffle: 1 việc, 2 lần bấm, 1 sản phẩm | |
| _ev(100, "feature_open", "open", feature="shuffle", task="s1"), | |
| _ev(102, "job_done", "run", feature="shuffle", task="s1", lat=1000), | |
| _ev(104, "regenerate", "run", feature="shuffle", task="s1"), | |
| _ev(106, "output_download", "output", feature="shuffle", task="s1", | |
| value="mp4", count=1), | |
| ] | |
| feats = {f["feature"]: f for f in etl.summarize(evs)["features"]} | |
| cut, sh = feats["cut"], feats["shuffle"] | |
| # cut cao nhất ở cả 3 thành phần -> tròn 100 điểm | |
| assert cut["score_demand"] == 100.0 and cut["score_output"] == 100.0 | |
| assert cut["score_rerun"] == 100.0 and cut["score_total"] == 100.0 | |
| # shuffle: nhu cầu 1/2, sản phẩm 1/4, bấm gấp đôi -> 0.4*50 + 0.2*50 + 0.4*25 | |
| assert sh["score_demand"] == 50.0 | |
| assert sh["score_output"] == 25.0 | |
| assert sh["score_rerun"] == 50.0 | |
| assert abs(sh["score_total"] - 40.0) < 0.05 | |
| assert cut["score_total"] > sh["score_total"] | |
| def test_contribution_rerun_zero_when_never_ran(): | |
| """Tool mở ra rồi bỏ (chưa bấm chạy lần nào) KHÔNG được 100đ "ít chạy lại". | |
| Kẹp sàn 1.0 như bản đầu làm cả bảng cùng 100 -> thành phần 20% mất tác dụng. | |
| Chưa chạy = chưa đóng góp -> 0 điểm. | |
| """ | |
| evs = [ | |
| _ev(0, "feature_open", "open", feature="cut", task="c1"), | |
| _ev(2, "job_done", "run", feature="cut", task="c1", lat=1000), | |
| _ev(4, "output_download", "output", feature="cut", task="c1", value="mp4", count=1), | |
| _ev(50, "feature_open", "open", feature="text_video", task="v1"), # mở rồi bỏ | |
| ] | |
| feats = {f["feature"]: f for f in etl.summarize(evs)["features"]} | |
| assert feats["cut"]["score_rerun"] == 100.0 | |
| assert feats["text_video"]["attempts_avg"] == 0 | |
| assert feats["text_video"]["score_rerun"] == 0.0 | |
| assert feats["cut"]["score_total"] > feats["text_video"]["score_total"] | |
| def test_contribution_score_handles_single_tool(): | |
| """Chỉ có 1 tool -> không vỡ, tự là tool cao nhất (100 điểm).""" | |
| f = etl.summarize(_one_full_task())["features"] | |
| assert len(f) == 1 and f[0]["score_total"] == 100.0 | |
| def test_passive_pages_are_not_tasks(): | |
| """page_view (trang chủ/hướng dẫn) không có task_id -> không tính vào A3.""" | |
| evs = _one_full_task() + [ | |
| _ev(400, "page_view", "open", feature="home", task=""), | |
| _ev(410, "page_view", "open", feature="guide", task=""), | |
| ] | |
| s = etl.summarize(evs) | |
| assert s["overall"]["tasks"] == 1 | |
| assert all(f["feature"] not in ("home", "guide") for f in s["features"]) | |
| # ── ETL: phiên làm việc ────────────────────────────────────────── | |
| def test_session_splits_on_long_silence(): | |
| """Im lặng quá ngưỡng -> tách thành hai phiên (spec §6: session_end là mốc mềm).""" | |
| gap = (settings.SESSION_IDLE_MINUTES + 10) * 60 | |
| evs = [ | |
| _ev(0, "session_start", "open", feature="", task="", sess="s"), | |
| _ev(60, "feature_open", "open", task="t1", sess="s"), | |
| _ev(60 + gap, "feature_open", "open", task="t2", sess="s"), | |
| _ev(120 + gap, "upload_done", "prepare", task="t2", sess="s"), | |
| ] | |
| sessions = etl.build_sessions(evs) | |
| assert len(sessions) == 2 | |
| # thời lượng DÙNG THẬT không được cộng cả quãng im lặng | |
| assert sum(s["active_ms"] for s in sessions) < gap * 1000 | |
| assert abs(sessions[0]["active_ms"] - 60000) < 50 | |
| def test_returning_user_needs_two_days(): | |
| """A5 = một user xuất hiện ở >= 2 ngày khác nhau.""" | |
| day = 86400 | |
| evs = [ | |
| _ev(0, "feature_open", "open", task="t1", user="an"), | |
| _ev(-2 * day, "feature_open", "open", task="t0", user="an", sess="s0"), | |
| _ev(0, "feature_open", "open", task="t2", user="binh", sess="sb"), | |
| ] | |
| users = {u["user_id"]: u for u in etl.summarize(evs)["users"]} | |
| assert users["an"]["returning"] is True | |
| assert users["binh"]["returning"] is False | |
| def test_user_detail_lists_sessions_and_tasks(): | |
| d = etl.user_detail(_one_full_task(), "an") | |
| assert d["user_id"] == "an" and len(d["sessions"]) == 1 | |
| task = d["sessions"][0]["tasks"][0] | |
| assert task["label"] == "Dịch / đổi giọng" | |
| assert task["has_output"] and task["machine_ms"] == 89000 | |
| assert [j["latency_ms"] for j in task["jobs"]] == [30000, 59000] | |
| # ── Dịch log sang tiếng Việt + xuất CSV ────────────────────────── | |
| def test_labels_translate_every_column(): | |
| """Mỗi cột log thô phải ra được một câu tiếng Việt, không lọt khoá máy.""" | |
| from app.services.analytics import labels as lb | |
| assert lb.feature_label("dub") == "Dịch / đổi giọng" | |
| assert lb.step_label("run") == "3. Máy chạy" | |
| assert lb.event_label("feature_abandon") == "Bỏ dở" | |
| assert lb.value_label("up").startswith("Hài lòng") | |
| assert lb.value_label("mp4") == "Video MP4" | |
| assert lb.duration_label(53043) == "53,0 giây" | |
| assert lb.duration_label(None) == "" # thao tác NGƯỜI -> để trống, không in "0" | |
| def test_labels_keep_unknown_keys(): | |
| """Tool/thuộc tính mới thêm mà quên cập nhật bảng thì giữ nguyên khoá gốc, | |
| tuyệt đối không được nuốt mất dữ liệu.""" | |
| from app.services.analytics import labels as lb | |
| assert lb.feature_label("tool_moi") == "tool_moi" | |
| assert lb.event_label("su_kien_la") == "su_kien_la" | |
| assert "thuoc_tinh_la: 7" in lb.describe_props({"thuoc_tinh_la": 7}) | |
| def test_describe_props_unpacks_json_blob(): | |
| from app.services.analytics import labels as lb | |
| txt = lb.describe_props({"job_id": "f09c", "kind": "download", "attempt": 1, | |
| "phase": "asr_translate", "dur_ms": 63290}) | |
| # thuộc tính quan trọng lên trước, mili-giây in ra kiểu người đọc | |
| assert txt.startswith("Lần bấm chạy thứ: 1") | |
| assert "Bước máy: Bóc lời & dịch" in txt | |
| assert "Loại job: Tải video" in txt | |
| assert "Kéo dài: 1 phút 3 giây" in txt | |
| def test_export_csv_is_readable_and_raw_keeps_schema(): | |
| """CSV mặc định phải đọc được; `?raw=1` giữ nguyên 11 cột schema v2.""" | |
| import csv as _csv | |
| import io as _io | |
| from app.api import analytics as api | |
| store.insert(_one_full_task()) | |
| api.settings.VIDEO_ADMIN_TOKEN = "tok" | |
| def _read(**kw): | |
| # StreamingResponse bọc iterator đồng bộ thành async generator -> phải | |
| # rút bằng vòng lặp asyncio chứ không list() thẳng được. | |
| import asyncio | |
| resp = api.admin_export_csv(token="tok", days=2, **kw) | |
| async def drain(): | |
| return [c async for c in resp.body_iterator] | |
| text = "".join(c.decode("utf-8") if isinstance(c, bytes) else c | |
| for c in asyncio.run(drain())) | |
| return list(_csv.reader(_io.StringIO(text.lstrip("")))) | |
| rows = _read() | |
| head = rows[0] | |
| assert head[:8] == ["STT", "Ngày", "Giờ", "Người dùng", "Mã người dùng", | |
| "Tool", "Bước", "Hành động"] | |
| body = [r for r in rows[1:] if r] | |
| assert any(r[5] == "Dịch / đổi giọng" and r[7] == "Job chạy xong" | |
| and r[9] == "30,0 giây" for r in body) | |
| assert any(r[7] == "Đánh giá kết quả" and r[8].startswith("Hài lòng") for r in body) | |
| raw = _read(raw=1) | |
| assert raw[0] == ["event_id", "event_time", "user_id", "session_id", "task_id", | |
| "feature", "step", "event_name", "event_value", "latency_ms", | |
| "properties_json"] | |
| # ── Chỉ số mức độ bám app (nhóm "Tình trạng người dùng") ───────── | |
| def test_adoption_counts_ignore_lookers(): | |
| """`used_users` chỉ đếm người có thao tác THẬT; hàng ngày/hàng tuần tính | |
| theo mốc mới nhất TRONG KHOẢNG LỌC, không phải theo giờ chạy báo cáo.""" | |
| day = 86400 | |
| evs = [ | |
| _ev(0, "feature_open", "open", user="an", task="t1"), # làm thật | |
| _ev(-3 * day, "run_click", "run", user="binh", task="t2", sess="sb"), | |
| _ev(-10 * day, "page_view", "open", user="cuong", task="", sess="sc"), | |
| ] | |
| o = etl.summarize(evs)["overall"] | |
| assert o["used_users"] == 2 # cuong chỉ ngó trang -> không tính | |
| assert o["daily_users"] == 1 # trong 24h cuối: an | |
| assert o["weekly_users"] == 2 # trong 7 ngày cuối: an + binh | |
| assert o["session_max_ms"] >= 0 | |
| def test_job_rows_carry_attempt_and_vietnamese_phase(): | |
| evs = _one_full_task() | |
| jobs = {j["phase"]: j for j in etl.summarize(evs)["jobs"]} | |
| assert jobs["tts"]["phase_label"] == "Đọc thành tiếng" | |
| assert jobs["asr_translate"]["runs"] == 1 | |
| assert jobs["asr_translate"]["attempt_avg"] == 1.0 | |