Create-knowledge-find-decode/tests/test_store.py

"""SQLite 数据层离线测：写入幂等 + 拍平 + 分页筛选（临时库，不碰 data/app.db）。"""
from __future__ import annotations

from acquisition import store


def _db(tmp_path):
    return store.connect(tmp_path / "t.db")


def test_connect_uses_ck_sqlite_path(tmp_path, monkeypatch):
    db = tmp_path / "legacy_data" / "app.db"
    monkeypatch.setenv("CK_SQLITE_PATH", str(db))

    c = store.connect()
    try:
        assert db.exists()
        assert store.list_runs(c) == []
    finally:
        c.close()


def test_insert_queries_idempotent_and_axes(tmp_path):
    c = _db(tmp_path)
    items = [{"query": "q1", "实质": "军人", "阶段": "灵感"}, {"query": "q2", "形式": "排比"}]
    assert store.insert_queries(c, "r1", "多轴", items, ts=100) == 2
    # 重导同 (run,method) → 覆盖不叠加
    assert store.insert_queries(c, "r1", "多轴", items, ts=100) == 2
    res = store.list_queries(c, method="多轴")
    assert res["total"] == 2
    assert res["items"][0]["axes"] == {"实质": "军人", "阶段": "灵感"}   # query 不进 axes
    assert "query" not in res["items"][0]["axes"]


def test_insert_queries_skips_empty_query(tmp_path):
    c = _db(tmp_path)
    n = store.insert_queries(c, "r1", "m", [{"query": ""}, {"实质": "x"}, {"query": "ok"}], ts=1)
    assert n == 1


def test_search_flatten_topk_and_ok(tmp_path):
    c = _db(tmp_path)
    recs = [
        {"method": "M", "query": "q1",
         "douyin": {"ok": [{"title": "t", "url": "u", "cover": "/c.jpg", "video": "/v.mp4"},
                           {"title": "t2", "url": "u2", "cover": "/c2.jpg", "video": "/v2.mp4"}],
                    "error": None},
         "weixin": {"ok": [{"title": "w", "url": "wu", "cover": "/wc.jpg", "nick": "号"}], "error": None}},
        {"method": "M", "query": "q2",
         "douyin": {"ok": [], "error": "未搜到"},
         "weixin": {"ok": [], "error": "搜索失败: x"},
         "xiaohongshu": {"ok": [{"title": "x", "url": "xu", "cover": "/xc.jpg", "nick": "薯"}], "error": None}},
    ]
    # q1(无 xhs 键，向后兼容跳过): 2 抖音 + 1 微信 = 3；q2: 抖音空 + 微信空 + 1 小红书 = 3 → 共 6
    assert store.insert_search_results(c, "s1", recs, ts=5) == 6
    xhs = store.list_search(c, platform="xiaohongshu", ok=True)
    assert xhs["total"] == 1 and xhs["items"][0]["extra"]["nick"] == "薯"
    ok_dy = store.list_search(c, platform="douyin", ok=True)
    assert ok_dy["total"] == 2                                   # q1 的两个视频
    assert {i["video"] for i in ok_dy["items"]} == {"/v.mp4", "/v2.mp4"}
    # 搜过但无结果 → ok=0 行带 error（按 query 取得到，按钮才显示「搜过·无结果」）
    bad = store.list_search(c, platform="weixin", ok=False)
    assert bad["total"] == 1 and "失败" in bad["items"][0]["extra"]["error"]
    good_wx = store.list_search(c, platform="weixin", ok=True)
    assert good_wx["items"][0]["extra"]["nick"] == "号"
    # summary：q1 有 3 个结果(抖音2+微信1)；q2 抖音/微信空 + 小红书 1 → ok=1, total=3 行
    s = store.search_summary(c)
    assert s["q1"]["ok"] == 3 and s["q2"]["ok"] == 1 and s["q2"]["total"] == 3


def test_pagination(tmp_path):
    c = _db(tmp_path)
    store.insert_queries(c, "r", "m", [{"query": f"q{i}"} for i in range(25)], ts=1)
    p1 = store.list_queries(c, page=1, size=10)
    p3 = store.list_queries(c, page=3, size=10)
    assert p1["total"] == 25 and len(p1["items"]) == 10
    assert len(p3["items"]) == 5            # 余 5 条
    assert p1["items"][0]["query"] != p3["items"][0]["query"]


def test_runs_and_methods(tmp_path):
    c = _db(tmp_path)
    store.insert_queries(c, "rq", "方法A", [{"query": "a"}], ts=2)
    store.insert_search_results(c, "rs", [{"method": "方法B", "query": "b",
                                           "douyin": None, "weixin": None}], ts=3)
    runs = {r["run_id"]: r["kind"] for r in store.list_runs(c)}
    assert runs == {"rq": "queries", "rs": "search"}
    assert store.list_methods(c, "queries") == ["方法A"]
    assert store.list_methods(c, "search_results") == ["方法B"]