308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""桃育种后端「单株评价 → 性状观测 → 统计分析」部署自测脚本（仅标准库 urllib）。
|
||
|
||
用途：一键验证 breeding 模块评价与统计全链路是否打通，可部署后回归自测。
|
||
链路：登录 → 建基础数据(基地/试验地/育种目标/杂交组合/单株)
|
||
→ 建评价(tree_evaluation) → 批量建观测(trait_observation)
|
||
→ 跑统计(describe / correlation / traits / selection-index)
|
||
→ 断言统计结果确实来自 trait_observation（均值精确回读）
|
||
|
||
特点：
|
||
· 登录自适应验证码（dev 开启时自动 captcha/get → slider/complete → 带 key 登录）。
|
||
· 双单元：①精确校验单元(单组合单株, 固定值, 断言 describe 均值精确回读)
|
||
②批量链路单元(多组合多单株, 仅验证 HTTP 200 与计数)。
|
||
· 默认跑完自动清理，库不留脏数据（CLEANUP=True）。
|
||
· 任意非 200 或断言失败立即退出码 1；全过退出码 0。
|
||
|
||
用法：
|
||
cd d:\\dpb\\dpb\\backend
|
||
$env:ENVIRONMENT='dev'
|
||
C:\\ai\\miniconda3\\envs\\dpb\\python.exe scripts/e2e_eval.py
|
||
"""
|
||
import datetime
|
||
import json
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
BASE = "http://localhost:5667"
|
||
RUN = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
|
||
CLEANUP = True # 部署自测默认清理，不留脏数据
|
||
N_COMBOS = 3 # 批量单元：组合数
|
||
N_TREES = 4 # 批量单元：每组合单株数
|
||
CORE = ["max_fruit_weight", "avg_fruit_weight", "longitudinal_dia",
|
||
"transverse_dia", "lateral_dia", "flesh_thickness", "ssc"]
|
||
|
||
HEADERS = {"Content-Type": "application/json"}
|
||
_COLLECT = {"obs": [], "eval": [], "tree": [], "combo": [], "plot": [], "site": [], "target": []}
|
||
TOKEN = None
|
||
|
||
|
||
def track(kind, obj):
|
||
if isinstance(obj, list):
|
||
_COLLECT[kind].extend(obj)
|
||
else:
|
||
_COLLECT[kind].append(obj)
|
||
return obj
|
||
|
||
|
||
def _req(method, url, *, data=None, headers=None, form=False):
|
||
h = dict(HEADERS)
|
||
if headers:
|
||
h.update(headers)
|
||
body = None
|
||
if data is not None:
|
||
if form:
|
||
body = urllib.parse.urlencode(data).encode()
|
||
h["Content-Type"] = "application/x-www-form-urlencoded"
|
||
else:
|
||
body = json.dumps(data).encode()
|
||
req = urllib.request.Request(url, data=body, method=method, headers=h)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as r:
|
||
return r.status, json.loads(r.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return e.code, json.loads(e.read().decode())
|
||
except Exception: # noqa: BLE001
|
||
return e.code, {"raw": e.read().decode()[:400]}
|
||
|
||
|
||
def step(name, status, body, *, expect=200):
|
||
if status != expect:
|
||
print(f"[FAIL] {name} -> HTTP {status}")
|
||
print(" BODY:", str(body)[:600])
|
||
sys.exit(1)
|
||
print(f"[OK ] {name} -> HTTP {status}")
|
||
return body
|
||
|
||
|
||
def assert_(cond, msg):
|
||
if not cond:
|
||
print(f"[FAIL] ASSERT: {msg}")
|
||
sys.exit(1)
|
||
print(f"[OK ] ASSERT: {msg}")
|
||
|
||
|
||
def login():
|
||
data = {"username": "super", "password": "123456", "grant_type": "password", "login_type": "PC端"}
|
||
st, b = _req("POST", f"{BASE}/api/v1/system/auth/login", data=data, form=True)
|
||
if st == 200 and b.get("code") == 0:
|
||
return b["data"]["access_token"]
|
||
# dev 开启验证码：拿 key -> 标记 verified -> 带 key 登录
|
||
key = _req("GET", f"{BASE}/api/v1/system/auth/captcha/get")[1]["data"]["key"]
|
||
_req("POST", f"{BASE}/api/v1/system/auth/captcha/slider/complete", data={"captcha_key": key})
|
||
data["captcha_key"] = key
|
||
st, b = _req("POST", f"{BASE}/api/v1/system/auth/login", data=data, form=True)
|
||
if st != 200 or b.get("code") != 0:
|
||
print("LOGIN FAILED", st, b)
|
||
sys.exit(1)
|
||
return b["data"]["access_token"]
|
||
|
||
|
||
def create(domain, payload):
|
||
st, b = _req("POST", f"{BASE}/api/v1/bre/{domain}/create", data=payload,
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
return step(f"POST /bre/{domain}/create", st, b)["data"]
|
||
|
||
|
||
def gget(domain, params=None):
|
||
url = f"{BASE}/api/v1/bre/{domain}/list"
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode(params)
|
||
st, b = _req("GET", url, headers={"Authorization": f"Bearer {TOKEN}"})
|
||
return step(f"GET /bre/{domain}/list", st, b)
|
||
|
||
|
||
def _lq(name, items):
|
||
"""构造 FastAPI list[str] 查询串：trait_codes=a&trait_codes=b（逗号串会被当单值）。"""
|
||
return "&".join(f"{name}={urllib.parse.quote(str(i))}" for i in items)
|
||
|
||
|
||
def gen_value(code, ci, ti):
|
||
"""批量单元确定性造数（落在合理量程，仅用于链路验证，非真实育种值）。"""
|
||
bases = {"max_fruit_weight": 80, "avg_fruit_weight": 60, "longitudinal_dia": 50,
|
||
"transverse_dia": 55, "lateral_dia": 52, "flesh_thickness": 8, "ssc": 8}
|
||
off = (ci * 7 + ti * 1.7 + CORE.index(code) * 0.9) % 30
|
||
return round(bases[code] + off, 1)
|
||
|
||
|
||
def precise_unit(core_traits, gid, pid):
|
||
"""精确均值校验：单组合单株 + 7 固定值观测，断言 describe 回读精确均值。"""
|
||
print("\n--- 单元① 精确均值校验 (单组合/单株, 固定值) ---")
|
||
site = create("site", {"site_name": f"E2E基地{RUN}-P", "remark": "自测自动创建"}, )
|
||
track("site", site["id"])
|
||
plot = create("plot", {"site_id": site["id"], "plot_code": f"P{RUN}-P"})
|
||
track("plot", plot["id"])
|
||
target = create("target", {"target_name": f"E2E目标{RUN}-P"})
|
||
track("target", target["id"])
|
||
combo = create("cross_combination", {
|
||
"combination_code": f"E2E-CC-{RUN}-P",
|
||
"bre_target_id": target["id"], "female_parent_id": gid,
|
||
"male_parent_id": gid, "cross_method": "人工杂交", "cross_year": 2026,
|
||
})
|
||
track("combo", combo["id"])
|
||
tree = create("tree", {
|
||
"combination_id": combo["id"], "plot_id": plot["id"], "tree_no": f"T{RUN}-P",
|
||
"row_no": 1, "col_no": 1, "planted_date": "2026-03-15", "status": "1",
|
||
"bre_personnel_id": pid,
|
||
})
|
||
track("tree", tree["id"])
|
||
|
||
ev = create("tree_evaluation", {
|
||
"combination_id": combo["id"], "tree_id": tree["id"], "evaluate_date": "2026-07-29",
|
||
"evaluate_year": 2026, "bre_personnel_id": pid, "remark": f"E2E自测{RUN}-P",
|
||
})
|
||
track("eval", ev["id"])
|
||
|
||
values = {"max_fruit_weight": 120.5, "avg_fruit_weight": 98.2, "longitudinal_dia": 65.1,
|
||
"transverse_dia": 70.3, "lateral_dia": 68.0, "flesh_thickness": 12.4, "ssc": 12.8}
|
||
for t in core_traits:
|
||
code = t["trait_code"]
|
||
ob = create("trait_observation", {
|
||
"evaluation_id": ev["id"], "tree_id": tree["id"], "combination_id": combo["id"],
|
||
"trait_id": t["id"], "category": t.get("category"), "trait_code": code,
|
||
"trait_name": t.get("trait_name"), "data_type": t.get("data_type"),
|
||
"evaluate_year": 2026, "value_numeric": values.get(code),
|
||
})
|
||
track("obs", ob["id"])
|
||
|
||
# describe?group_by=combination：该组合 stats 的 mean 应精确等于插入值
|
||
qs = _lq("trait_codes", CORE)
|
||
st, b = _req("GET", f"{BASE}/api/v1/bre/statistics/describe?{qs}&group_by=combination",
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
de = step("GET /bre/statistics/describe(group_by=combination)", st, b)
|
||
groups = de["data"].get("groups", [])
|
||
grp = next((g for g in groups if g["group"] == combo["id"]), None)
|
||
assert_(grp is not None, f"describe 回读包含本组合 {combo['id']}")
|
||
stat_map = {s["trait"]: s for s in grp["stats"]}
|
||
for code, val in values.items():
|
||
assert_(code in stat_map, f"describe 含性状 {code}")
|
||
assert_(abs(stat_map[code]["mean"] - val) < 1e-6, f"{code} 均值精确回读={val} (got {stat_map[code]['mean']})")
|
||
assert_(stat_map[code]["n"] == 1, f"{code} 覆盖单株数 n==1")
|
||
assert_(de["data"]["total"] == 1, "describe total==1 (单株)")
|
||
return 1 # 贡献的单株数
|
||
|
||
|
||
def batch_unit(n_combos, n_trees, core_traits, gid, pid):
|
||
"""批量链路单元：多组合多单株，仅验证接口不报错且计数正确。"""
|
||
print(f"\n--- 单元② 批量链路 (组合×单株 = {n_combos}×{n_trees}) ---")
|
||
site = create("site", {"site_name": f"E2E基地{RUN}-B", "remark": "自测自动创建"})
|
||
track("site", site["id"])
|
||
plot = create("plot", {"site_id": site["id"], "plot_code": f"P{RUN}-B"})
|
||
track("plot", plot["id"])
|
||
target = create("target", {"target_name": f"E2E目标{RUN}-B"})
|
||
track("target", target["id"])
|
||
|
||
total_trees = 0
|
||
for ci in range(n_combos):
|
||
combo = create("cross_combination", {
|
||
"combination_code": f"E2E-CC-{RUN}-B{ci}",
|
||
"bre_target_id": target["id"], "female_parent_id": gid,
|
||
"male_parent_id": gid, "cross_method": "人工杂交", "cross_year": 2026,
|
||
})
|
||
track("combo", combo["id"])
|
||
for ti in range(n_trees):
|
||
tree = create("tree", {
|
||
"combination_id": combo["id"], "plot_id": plot["id"],
|
||
"tree_no": f"T{RUN}-B{ci}-{ti}", "row_no": ti + 1, "col_no": 1,
|
||
"planted_date": "2026-03-15", "status": "1", "bre_personnel_id": pid,
|
||
})
|
||
track("tree", tree["id"])
|
||
ev = create("tree_evaluation", {
|
||
"combination_id": combo["id"], "tree_id": tree["id"], "evaluate_date": "2026-07-29",
|
||
"evaluate_year": 2026, "bre_personnel_id": pid, "remark": f"E2E自测{RUN}-B{ci}-{ti}",
|
||
})
|
||
track("eval", ev["id"])
|
||
for t in core_traits:
|
||
code = t["trait_code"]
|
||
ob = create("trait_observation", {
|
||
"evaluation_id": ev["id"], "tree_id": tree["id"], "combination_id": combo["id"],
|
||
"trait_id": t["id"], "category": t.get("category"), "trait_code": code,
|
||
"trait_name": t.get("trait_name"), "data_type": t.get("data_type"),
|
||
"evaluate_year": 2026, "value_numeric": gen_value(code, ci, ti),
|
||
})
|
||
track("obs", ob["id"])
|
||
total_trees += 1
|
||
return total_trees
|
||
|
||
|
||
def stat_checks(precise_trees, batch_trees):
|
||
print("\n--- 统计接口断言 ---")
|
||
# traits 核心性状数
|
||
st, b = _req("GET", f"{BASE}/api/v1/bre/statistics/traits?core_only=1",
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
tr = step("GET /bre/statistics/traits?core_only=1", st, b)
|
||
assert_(len(tr["data"]) == 7, f"核心性状数==7 (got {len(tr['data'])})")
|
||
|
||
# describe 全局 total
|
||
qs = _lq("trait_codes", CORE)
|
||
st, b = _req("GET", f"{BASE}/api/v1/bre/statistics/describe?{qs}&group_by=combination",
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
de = step("GET /bre/statistics/describe", st, b)
|
||
expected_total = precise_trees + batch_trees
|
||
assert_(de["data"]["total"] == expected_total,
|
||
f"describe total=={expected_total} (got {de['data']['total']})")
|
||
|
||
# correlation 矩阵维度与对角
|
||
cqs = _lq("trait_codes", ["avg_fruit_weight", "ssc"])
|
||
st, b = _req("GET", f"{BASE}/api/v1/bre/statistics/correlation?{cqs}",
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
co = step("GET /bre/statistics/correlation", st, b)
|
||
assert_(co["data"]["n"] >= 1 and len(co["data"]["matrix"]) == 2, "correlation 矩阵 2x2")
|
||
assert_(co["data"]["matrix"][0][0] == 1.0 and co["data"]["matrix"][1][1] == 1.0, "correlation 对角线==1.0")
|
||
|
||
# selection-index 返回排名
|
||
st, b = _req("POST", f"{BASE}/api/v1/bre/statistics/selection-index",
|
||
data={"weights": {c: 1.0 for c in CORE}, "year": 2026, "top_n": 10},
|
||
headers={"Authorization": f"Bearer {TOKEN}"})
|
||
si = step("POST /bre/statistics/selection-index", st, b)
|
||
top = si["data"].get("top") if isinstance(si.get("data"), dict) else None
|
||
assert_(isinstance(top, list) and len(top) >= 1,
|
||
f"selection-index 返回非空排名 (n={len(top) if isinstance(top, list) else '?'})")
|
||
|
||
|
||
def cleanup():
|
||
if not CLEANUP:
|
||
print("\n（CLEANUP=False，测试数据保留；再次运行自动用新 RUN 后缀避免冲突）")
|
||
return
|
||
print("\n清理测试数据...")
|
||
h = {"Authorization": f"Bearer {TOKEN}"}
|
||
order = [("trait_observation", "obs"), ("tree_evaluation", "eval"), ("tree", "tree"),
|
||
("cross_combination", "combo"), ("plot", "plot"), ("site", "site"), ("target", "target")]
|
||
for domain, key in order:
|
||
ids = _COLLECT[key]
|
||
if not ids:
|
||
continue
|
||
st, _ = _req("DELETE", f"{BASE}/api/v1/bre/{domain}/delete", data=ids, headers=h)
|
||
print(f" DELETE /bre/{domain}/delete {len(ids)} 条 -> HTTP {st}")
|
||
|
||
|
||
def main():
|
||
global TOKEN
|
||
TOKEN = login()
|
||
print(f"登录成功，token 前缀: {TOKEN[:12]}...\n")
|
||
|
||
tr = gget("trait", {"page": 1, "page_size": 100})
|
||
traits = tr["data"]["items"]
|
||
core_traits = [t for t in traits if t["trait_code"] in set(CORE) and t.get("data_type") == "numeric"]
|
||
assert_(len(core_traits) == 7, f"核心数值性状命中 7/7 (got {len(core_traits)})")
|
||
print(f"核心数值性状: {[t['trait_code'] for t in core_traits]}\n")
|
||
|
||
germ = gget("germplasm", {"page": 1, "page_size": 5})
|
||
gid = germ["data"]["items"][0]["id"]
|
||
pers = gget("personnel", {"page": 1, "page_size": 5})
|
||
pid = pers["data"]["items"][0]["id"]
|
||
|
||
try:
|
||
p = precise_unit(core_traits, gid, pid)
|
||
b = batch_unit(N_COMBOS, N_TREES, core_traits, gid, pid)
|
||
stat_checks(p, b)
|
||
print("\n=== 部署自测全部通过 ✅ ===")
|
||
finally:
|
||
cleanup()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|