| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374 |
- """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_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"}},
- ]
- # q1: 2 抖音 + 1 微信 = 3 个 ok 行；q2: 抖音空 + 微信空 = 2 个 ok=0 行 → 共 5
- assert store.insert_search_results(c, "s1", recs, ts=5) == 5
- 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 个结果，q2 有 0 个但被记为搜过（total=2 行）
- s = store.search_summary(c)
- assert s["q1"]["ok"] == 3 and s["q2"]["ok"] == 0 and s["q2"]["total"] == 2
- 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"]
|
