test_build_docs_api.py
5.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/env python3
"""
测试 POST /indexer/build-docs 接口：构造请求数据、调用接口、打印完整响应。
用法:
1. 先启动 Indexer 服务: ./scripts/start_indexer.sh (或 uvicorn api.indexer_app:app --port 6004)
2. 执行: python scripts/test_build_docs_api.py
也可指定地址: INDEXER_URL=http://localhost:6004 python scripts/test_build_docs_api.py
"""
import json
import os
import sys
from datetime import datetime, timezone
# 项目根目录
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
# 默认使用 requests 调真实服务；若未安装则回退到 TestClient
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
def build_sample_request():
"""构造一条完整的 build-docs 请求体（对应 shoplazza_product_spu / sku / option 表结构）。"""
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
sample_image_url = os.getenv(
"SAMPLE_IMAGE_URL",
"https://oss.essa.cn/98532128-cf8e-456c-9e30-6f2a5ea0c19f.jpg",
)
return {
"tenant_id": "162",
"items": [
{
"spu": {
"id": 10001,
"tenant_id": "162",
"title": "测试T恤 纯棉短袖",
"brief": "舒适纯棉，多色可选",
"description": "这是一款适合日常穿着的纯棉T恤，透气吸汗。",
"vendor": "测试品牌",
"category": "服装/上衣/T恤",
"category_id": 100,
"category_level": 2,
"category_path": "服装/上衣/T恤",
"fake_sales": 1280,
"image_src": sample_image_url,
"tags": "T恤,纯棉,短袖,夏季",
"create_time": now,
"update_time": now,
},
"skus": [
{
"id": 20001,
"spu_id": 10001,
"price": 99.0,
"compare_at_price": 129.0,
"sku": "SKU-TSHIRT-001",
"inventory_quantity": 50,
"option1": "黑色",
"option2": "M",
"option3": None,
},
{
"id": 20002,
"spu_id": 10001,
"price": 99.0,
"compare_at_price": 129.0,
"sku": "SKU-TSHIRT-002",
"inventory_quantity": 30,
"option1": "白色",
"option2": "L",
"option3": None,
},
],
"options": [
{"id": 1, "position": 1, "name": "颜色"},
{"id": 2, "position": 2, "name": "尺码"},
],
}
],
}
def call_via_http(base_url: str, body: dict):
"""通过 HTTP 调用 build-docs。"""
url = f"{base_url.rstrip('/')}/indexer/build-docs"
r = requests.post(url, json=body, timeout=30)
return r.status_code, r.text, r.json() if r.headers.get("content-type", "").startswith("application/json") else None
def call_via_test_client(body: dict):
"""通过 FastAPI TestClient 调用（不依赖已启动服务，但需 DB/ES 已配置）。"""
from fastapi.testclient import TestClient
import api.indexer_app as indexer_app
with TestClient(indexer_app.app) as client:
r = client.post("/indexer/build-docs", json=body)
return r.status_code, r.text, r.json() if r.headers.get("content-type", "").startswith("application/json") else None
def main():
body = build_sample_request()
print("=" * 60)
print("【请求】POST /indexer/build-docs")
print("=" * 60)
print(json.dumps(body, ensure_ascii=False, indent=2))
base_url = os.getenv("INDEXER_URL", "http://localhost:6004")
use_http = HAS_REQUESTS and (os.getenv("USE_TEST_CLIENT", "").lower() not in ("1", "true", "yes"))
if use_http:
try:
status, raw, data = call_via_http(base_url, body)
except requests.RequestException as e:
print("\n[错误] 无法连接 Indexer 服务:", e)
print("请先启动: ./scripts/start_indexer.sh 或 uvicorn api.indexer_app:app --port 6004")
if HAS_REQUESTS:
print("或使用进程内测试: USE_TEST_CLIENT=1 python scripts/test_build_docs_api.py")
sys.exit(1)
else:
if not use_http and not HAS_REQUESTS:
print("\n[提示] 未安装 requests，使用 TestClient 调用（需配置 DB/ES）。")
else:
print("\n[提示] 使用 TestClient 调用（USE_TEST_CLIENT=1）。")
try:
status, raw, data = call_via_test_client(body)
except Exception as e:
print("\n[错误] TestClient 调用失败:", e)
print("请确保已 source activate.sh 且 DB/ES 环境变量正确，或先启动 Indexer 再用 HTTP 调用。")
sys.exit(1)
print("\n" + "=" * 60)
print("【响应】HTTP status =", status)
print("=" * 60)
if data is not None:
print(json.dumps(data, ensure_ascii=False, indent=2, default=str))
if data.get("docs"):
doc = data["docs"][0]
print("\n" + "=" * 60)
print("【返回 doc 顶层字段】共 {} 个".format(len(doc)))
print("=" * 60)
for k in sorted(doc.keys()):
print(" ", k)
else:
print(raw)
if status != 200:
sys.exit(1)
if __name__ == "__main__":
main()