"""淘宝 deeplink 失效标记 + lookup 过滤 + invalidate 端点测试。 覆盖:按 shopId 标记所有行(幂等)、lookup 过滤失效淘宝候选、失效后新行仍可命中、 invalidate 端点鉴权(503 未配 / 401 头错 / 200)、端到端标记后 lookup MISS、非淘宝 no-op。 """ from __future__ import annotations import pytest from app.core.config import settings from app.db.session import SessionLocal from app.models.store_mapping import StoreMapping from app.repositories import store_mapping as repo _SECRET = "test-internal-secret-only-for-pytest" def _mk_row(db, *, trace_id, name_meituan=None, name_taobao=None, id_taobao=None, deeplink=None, lat=None, lng=None, name_jd=None, id_jd=None, jd_deeplink=None): row = StoreMapping( trace_id=trace_id, business_type="food", name_meituan=name_meituan, name_taobao=name_taobao, id_taobao=id_taobao, taobao_deeplink=deeplink, lat=lat, lng=lng, name_jd=name_jd, id_jd=id_jd, jd_deeplink=jd_deeplink, ) db.add(row) db.commit() db.refresh(row) return row @pytest.fixture() def db(): """每个用例前后清空 store_mapping(表由 conftest 的 create_all 建好,session 级共享)。""" s = SessionLocal() s.query(StoreMapping).delete() s.commit() yield s s.query(StoreMapping).delete() s.commit() s.close() # ---------- repo 层 ---------- def test_mark_invalid_marks_all_rows_with_shop_id(db): # 同一个坏 shopId 散在两行(两次比价),另一行不同 shopId _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_a") _mk_row(db, trace_id="t2", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_b") _mk_row(db, trace_id="t3", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="GOOD9", deeplink="dl_g") # 按 shopId 标记所有行 assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 2 # 幂等:已标记的不重复 assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 0 rows = {r.trace_id: r for r in db.query(StoreMapping).all()} assert rows["t1"].taobao_deeplink_invalid_at is not None assert rows["t2"].taobao_deeplink_invalid_at is not None assert rows["t3"].taobao_deeplink_invalid_at is None # 不同 shopId 不动 def test_lookup_filters_invalid_taobao(db): _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl") # 标记前命中 assert repo.lookup_nearest(db, "meituan", "绝味鸭脖")["taobao"]["shop_id"] == "BAD1" # 标记失效后淘宝候选被过滤 → MISS(仅此一条淘宝) repo.mark_taobao_deeplink_invalid(db, "BAD1") assert "taobao" not in repo.lookup_nearest(db, "meituan", "绝味鸭脖") def test_lookup_picks_new_valid_row_after_invalidate(db): # 失效旧行 + 重搜写的新行(新 shopId,invalid_at=NULL)共存 → lookup 选到新行 _mk_row(db, trace_id="t1", name_meituan="店A", name_taobao="店A", id_taobao="BAD1", deeplink="dl_bad") repo.mark_taobao_deeplink_invalid(db, "BAD1") _mk_row(db, trace_id="t2", name_meituan="店A", name_taobao="店A", id_taobao="NEW2", deeplink="dl_new") assert repo.lookup_nearest(db, "meituan", "店A")["taobao"]["shop_id"] == "NEW2" # ---------- invalidate 端点 ---------- def test_invalidate_endpoint_secret_unset_503(client, monkeypatch): monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "") # 未配 = 端点关闭 r = client.post("/internal/store-mapping/invalidate", json={"platform": "taobao", "shop_id": "X"}) assert r.status_code == 503 def test_invalidate_endpoint_auth(client, monkeypatch): monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) body = {"platform": "taobao", "shop_id": "X"} assert client.post("/internal/store-mapping/invalidate", json=body).status_code == 401 assert client.post("/internal/store-mapping/invalidate", json=body, headers={"X-Internal-Secret": "wrong"}).status_code == 401 r = client.post("/internal/store-mapping/invalidate", json=body, headers={"X-Internal-Secret": _SECRET}) assert r.status_code == 200 and r.json()["ok"] is True def test_invalidate_endpoint_marks_and_lookup_miss(client, db, monkeypatch): monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) _mk_row(db, trace_id="t1", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl") _mk_row(db, trace_id="t2", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl2") r = client.post("/internal/store-mapping/invalidate", json={"platform": "taobao", "shop_id": "BADX"}, headers={"X-Internal-Secret": _SECRET}) assert r.status_code == 200 and r.json()["affected"] == 2 lk = client.get("/internal/store-mapping/lookup", params={"source_platform": "meituan", "name": "店B"}, headers={"X-Internal-Secret": _SECRET}) assert "taobao" not in lk.json() def test_invalidate_endpoint_unsupported_platform_noop(client, monkeypatch): # 当前支持 taobao/jd;其它平台(如 meituan)no-op 返 affected=0 monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) r = client.post("/internal/store-mapping/invalidate", json={"platform": "meituan", "shop_id": "X"}, headers={"X-Internal-Secret": _SECRET}) assert r.status_code == 200 and r.json()["affected"] == 0 # ---------- 京东(对称淘宝)---------- def test_mark_jd_invalid_marks_all_rows_with_store_id(db): _mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_a") _mk_row(db, trace_id="j2", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_b") _mk_row(db, trace_id="j3", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JGOOD", jd_deeplink="dl_g") assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 2 assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 0 # 幂等 rows = {r.trace_id: r for r in db.query(StoreMapping).all()} assert rows["j1"].jd_deeplink_invalid_at is not None assert rows["j2"].jd_deeplink_invalid_at is not None assert rows["j3"].jd_deeplink_invalid_at is None # 不同 storeId 不动 def test_lookup_filters_invalid_jd(db): _mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl") assert repo.lookup_nearest(db, "meituan", "兰州拉面")["jd"]["store_id"] == "JBAD" repo.mark_jd_deeplink_invalid(db, "JBAD") assert "jd" not in repo.lookup_nearest(db, "meituan", "兰州拉面") def test_invalidate_endpoint_jd_marks_and_lookup_miss(client, db, monkeypatch): monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) _mk_row(db, trace_id="j1", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl") _mk_row(db, trace_id="j2", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl2") r = client.post("/internal/store-mapping/invalidate", json={"platform": "jd", "shop_id": "JBADX"}, headers={"X-Internal-Secret": _SECRET}) assert r.status_code == 200 and r.json()["affected"] == 2 lk = client.get("/internal/store-mapping/lookup", params={"source_platform": "meituan", "name": "店J"}, headers={"X-Internal-Secret": _SECRET}) assert "jd" not in lk.json()