157 lines
7.5 KiB
Python
157 lines
7.5 KiB
Python
"""淘宝 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()
|