yml2213
afa5c815f6
让运营能按订单号查到订单，作为后续资料收集（P2）与皮肤准入（P3/P4）的输入。 order_site 零改动：只读直连、独立极小连接池、可一键降级。 上机核实推翻了规划两处假设（详见 §6.5.1）： - orders 表没有 order_no 列，只有 id 与 platform_order_id → 查询入口用后者； - platform_order_id 非唯一：541 组重复，形态固定为「91kaquan 内部行(已发货) + 真实店铺行(无人管)」。按规划原文的单行判据，这 540 个已发货订单会被误判 成「没人管」而放行重复履约。改为**订单号维度聚合**：任一行已发货即 auto_fulfilled。 真机验证（只读账号 livehub_ro）：三桶判定全对，重复组正确拦下并回显真实 店铺行的中文名（狂沙舞者，而非内部码 suit_sand_dancer），单次查询 3.3ms。 顺带修一个只在真机才暴露的 bug：_get_engine 先置 _INITIALIZED 再读 URL， 一次落空就把模块永久毒化成不可用，之后补上连接串也救不回来。
89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
"""P0 适配层对真实只读库的连通性验证（**手工跑，不进自动回归**）。
|
|
|
|
需要能连到 order_site 的 PostgreSQL（生产机或已打通网络的环境）：
|
|
|
|
ORDER_SITE_DATABASE_URL=postgresql://livehub_ro:...@host:5432/order_site \
|
|
ORDER_SITE_LOOKUP_ENABLED=true \
|
|
.venv/bin/python -m pytest tests/test_external_order_source_live.py -q -s
|
|
|
|
默认会因为连不上而 skip，不会误报失败。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
import web.backend.services.external_order_source as src
|
|
|
|
# 三桶各取一个真实订单号（2026-10-02 实测）。
|
|
ORDER_C_UNHANDLED = "2627501602915345" # 激萌罗小黑，C 桶
|
|
ORDER_B_WORKER = "2627501932004488" # 指挥官密钥4个，B 桶
|
|
ORDER_A_AUTO = "2627501850053080" # 套装-浪漫天命，A 桶
|
|
# 重复订单号：91kaquan 内部行已发货 + 真实店铺行无人管（§6.5.1 的 540 组之一）
|
|
ORDER_DUPLICATE = "2624100167260659"
|
|
ORDER_MISSING = "0000000000000000"
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not os.getenv("ORDER_SITE_DATABASE_URL", "").strip(),
|
|
reason="未配置 ORDER_SITE_DATABASE_URL，跳过真实库验证",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def _configured(monkeypatch):
|
|
monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true")
|
|
src.reset_engine()
|
|
yield
|
|
src.reset_engine()
|
|
|
|
|
|
def test_c_bucket_is_unhandled(_configured):
|
|
result = src.lookup_order(ORDER_C_UNHANDLED, use_cache=False)
|
|
assert result["route"] == src.ROUTE_UNHANDLED
|
|
assert result["sku_name"]
|
|
|
|
|
|
def test_b_bucket_is_worker_platform(_configured):
|
|
assert (
|
|
src.lookup_order(ORDER_B_WORKER, use_cache=False)["route"]
|
|
== src.ROUTE_WORKER_PLATFORM
|
|
)
|
|
|
|
|
|
def test_a_bucket_is_auto_fulfilled(_configured):
|
|
assert (
|
|
src.lookup_order(ORDER_A_AUTO, use_cache=False)["route"]
|
|
== src.ROUTE_AUTO_FULFILLED
|
|
)
|
|
|
|
|
|
def test_duplicate_order_no_is_treated_as_fulfilled(_configured):
|
|
"""核心回归：同一订单号下内部行已发货 → 整单不能放行。
|
|
|
|
按店铺行单独看是 unhandled，放行就会重复履约。
|
|
"""
|
|
result = src.lookup_order(ORDER_DUPLICATE, use_cache=False)
|
|
assert result["route"] == src.ROUTE_AUTO_FULFILLED
|
|
# 回显取真实店铺行的中文名，不是内部行的 suit_sand_dancer
|
|
assert result["sku_name"] == "狂沙舞者"
|
|
|
|
|
|
def test_missing_order_raises(_configured):
|
|
with pytest.raises(src.OrderLookupUnavailable):
|
|
src.lookup_order(ORDER_MISSING, use_cache=False)
|
|
|
|
|
|
def test_lookup_is_fast(_configured):
|
|
"""对对方库的负载必须可忽略（规划验收标准第 6 条：单行索引查询 < 50ms）。"""
|
|
import time
|
|
|
|
src.lookup_order(ORDER_C_UNHANDLED, use_cache=False) # 预热连接
|
|
started = time.perf_counter()
|
|
for _ in range(5):
|
|
src.lookup_order(ORDER_C_UNHANDLED, use_cache=False)
|
|
elapsed_ms = (time.perf_counter() - started) / 5 * 1000
|
|
print(f"\n单次查询耗时：{elapsed_ms:.1f} ms")
|
|
assert elapsed_ms < 50, f"查询耗时 {elapsed_ms:.1f}ms 超过 50ms"
|