"""bre 模块全量冒烟 + 评价→观测→统计全链路测试（FastAPI TestClient，无需起服务/端口）。 运行（cwd 任意）: cd d:\dpb\dpb\backend C:\ai\miniconda3\envs\dpb\python.exe -X utf8 scripts/test_bre_full_tc.py 覆盖: 1) 全部 24 个 bre_* CRUD 模块 /list 只读冒烟（改名后列/关系错位会 500） 2) 写链路 site→plot→target→germplasm→personnel→cross_combination→tree→tree_evaluation (验证 bre_target_id / bre_personnel_id 等改名列写入) → trait_observation → 统计四接口 traits/describe/correlation/selection-index 3) 自动清理本次创建数据 依赖: Redis + PG 正常（TestClient 走真实 lifespan 初始化）。 """ import os os.environ["ENVIRONMENT"] = "dev" os.environ["PYTHONUTF8"] = "1" import sys, json sys.path.insert(0, r"d:\dpb\dpb\backend") import main from fastapi.testclient import TestClient create_app = main.create_app DOMAINS = [ "germplasm", "cross_combination", "target", "trial", "trial_study", "trial_study_entry", "treatment", "group", "pedigree", "plan", "site", "plot", "personnel", "pollination", "seed_treatment", "seedling", "planting", "tree", "tree_evaluation", "trait_observation", "tree_photo", "trait", "selection_rule", "selection_result", ] TOKEN = None def login(client): global TOKEN d = {"username": "super", "password": "123456", "grant_type": "password", "login_type": "PC端"} r = client.post("/api/v1/system/auth/login", data=d) b = r.json() if r.status_code == 200 and b.get("code") == 0: TOKEN = b["data"]["access_token"]; return key = client.get("/api/v1/system/auth/captcha/get").json()["data"]["key"] client.post("/api/v1/system/auth/captcha/slider/complete", json={"captcha_key": key}) d["captcha_key"] = key r = client.post("/api/v1/system/auth/login", data=d) b = r.json() assert r.status_code == 200 and b.get("code") == 0, f"LOGIN FAIL {r.status_code} {b}" TOKEN = b["data"]["access_token"] def auth(): return {"Authorization": f"Bearer {TOKEN}"} def check(name, r, expect=200): ok = r.status_code == expect print(f"[{'OK ' if ok else 'FAIL'}] {name} -> HTTP {r.status_code}") if not ok: print(" BODY:", str(r.text)[:500]) sys.exit(1) return r.json() def main_(): with TestClient(create_app()) as client: login(client) print("\n=== 1) 全模块 LIST 冒烟 ===") for d in DOMAINS: r = client.get(f"/api/v1/bre/{d}/list", headers=auth(), params={"page": 1, "page_size": 5}) check(f"GET /bre/{d}/list", r) print("\n=== 2) 评价→观测→统计 全链路 ===") RUN = __import__("datetime").datetime.now().strftime("%H%M%S") site = check("create site", client.post("/api/v1/bre/site/create", json={"site_name": f"TC基地{RUN}"}, headers=auth()))["data"] plot = check("create plot", client.post("/api/v1/bre/plot/create", json={"site_id": site["id"], "plot_code": f"TC{RUN}"}, headers=auth()))["data"] target = check("create target", client.post("/api/v1/bre/target/create", json={"target_name": f"TC目标{RUN}"}, headers=auth()))["data"] germ = check("create germplasm", client.post("/api/v1/bre/germplasm/create", json={"cultivar_name": f"TC亲本{RUN}"}, headers=auth()))["data"] personnel = check("create personnel", client.post("/api/v1/bre/personnel/create", json={"name": f"TC人员{RUN}"}, headers=auth()))["data"] combo = check("create cross_combination", client.post("/api/v1/bre/cross_combination/create", json={ "combination_code": f"TC-CC-{RUN}", "bre_target_id": target["id"], "female_parent_id": germ["id"], "male_parent_id": germ["id"], "cross_method": "artificial", "cross_year": 2026}, headers=auth()))["data"] tree = check("create tree", client.post("/api/v1/bre/tree/create", json={ "combination_id": combo["id"], "plot_id": plot["id"], "tree_no": f"T{RUN}", "row_no": 1, "col_no": 1, "planted_date": "2026-03-15", "status": "alive", "bre_personnel_id": personnel["id"]}, headers=auth()))["data"] # 波次2.5 验收: 有单株的 plot 跨基地迁移必须被明确拒绝(409) siteB = check("create siteB", client.post("/api/v1/bre/site/create", json={"site_name": f"TC基地B{RUN}"}, headers=auth()))["data"] rb = client.put(f"/api/v1/bre/plot/update/{plot['id']}", headers=auth(), json={"site_id": siteB["id"], "plot_code": plot["plot_code"]}) assert rb.status_code == 409, f"plot cross-base migration w/ tree should 409, got {rb.status_code} {rb.text[:200]}" ev = check("create tree_evaluation", client.post("/api/v1/bre/tree_evaluation/create", json={ "combination_id": combo["id"], "tree_id": tree["id"], "evaluate_date": "2026-07-31", "evaluate_year": 2026, "bre_personnel_id": personnel["id"], "remark": f"TC自测{RUN}"}, headers=auth()))["data"] traits = check("GET traits", client.get("/api/v1/bre/statistics/traits", headers=auth(), params={"group_by": "combination"}))["data"] core = [t for t in traits if t.get("is_core")] print(f" traits total={len(traits)}, core={len(core)}") assert len(core) >= 1, "无核心性状可观测" vals = {"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} obs_ids = [] for t in core[:3]: code = t["trait_code"] if code not in vals: continue r = check(f"create trait_observation {code}", client.post("/api/v1/bre/trait_observation/create", json={ "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": vals[code]}, headers=auth())) obs_ids.append(r["data"]["id"]) qs = "&".join(f"trait_codes={c}" for c in ["max_fruit_weight", "avg_fruit_weight"]) check("GET describe", client.get(f"/api/v1/bre/statistics/describe?{qs}&group_by=combination", headers=auth())) check("GET correlation", client.get(f"/api/v1/bre/statistics/correlation?{qs}", headers=auth())) check("POST selection-index", client.post("/api/v1/bre/statistics/selection-index", json={"weights": {"max_fruit_weight": 1.0}, "year": 2026, "top_n": 10}, headers=auth())) print("\n=== 3) 清理 ===") if obs_ids: check("delete trait_observation", client.request("DELETE", f"/api/v1/bre/trait_observation/delete", headers=auth(), json=obs_ids)) for dom, obj in [("tree_evaluation", ev), ("tree", tree), ("cross_combination", combo), ("germplasm", germ), ("personnel", personnel), ("target", target), ("plot", plot), ("site", site), ("site", siteB)]: check(f"delete {dom}", client.request("DELETE", f"/api/v1/bre/{dom}/delete", headers=auth(), json=[obj["id"]])) print("\nALL PASS \u2705") if __name__ == "__main__": main_()