"""波次6 桃育种领域需求对齐 聚焦验证（TestClient，无需起服务）。 运行: cd d:\dpb\dpb\backend C:\ai\miniconda3\envs\dpb\python.exe -X utf8 scripts/test_bre_wave6_tc.py 覆盖: 6.1 编号体系: combination_code 留空自动生成(YY+3位); tree_no 留空自动生成(组合-序号) 6.2 字典补灌: 规划附录 40 枚举全部落库（含 width 窄/中/宽） 6.3 v_trait_value 视图: 与 bre_trait_observation 聚合结果一致，可被统计查询消费 6.4 数据质量门禁: 数值观测越出 trait.valid_min/max -> 409; 区间内放行 6.5 状态机: tree.status / selection_result.is_selected 只能向前流转, 回退 409 """ import os os.environ["ENVIRONMENT"] = "dev" os.environ["PYTHONUTF8"] = "1" import re import sys sys.path.insert(0, r"d:\dpb\dpb\backend") import main from fastapi.testclient import TestClient create_app = main.create_app 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 call(client, method, url, **kw): r = client.request(method, url, headers=auth(), **kw) try: body = r.json() except Exception: body = {"_raw": r.text[:300]} return r.status_code, body # 规划附录 40 枚举: {dict_type: 期望最少值数} APPENDIX_40 = { "weather": 5, "level_3": 3, "level_4": 4, "level_5": 5, "depth": 3, "width": 3, "brightness": 3, "conc_level": 3, "quality": 3, "pollination": 4, "seed_treatment": 3, "identify_status": 6, "growth_vigor": 3, "growth_type": 3, "flower_type": 2, "pollen_amount": 4, "fruit_load": 3, "maturity_uniform": 3, "fruit_shape": 5, "fruit_apex": 5, "fruit_base_type": 3, "symmetry": 3, "peel_base_color": 5, "blush_area": 5, "blush_pattern": 3, "peel_removal": 3, "flesh_color": 7, "firmness": 5, "flavor": 7, "pit_adherence": 3, "pit_size": 3, "pit_shape": 5, "eval_result": 3, "propagation": 3, "retrial_result": 2, "audit_type": 2, "audit_status": 5, "annual_fee": 3, "blush_uniformity": 3, "peel_color_depth": 3, } def check_dict_appendix(): from sqlalchemy import create_engine, text from app.config.setting import settings eng = create_engine(settings.DB_URI) with eng.connect() as c: rows = c.execute(text( "SELECT dict_type, count(*) FROM sys_dict_data " "WHERE is_deleted = false GROUP BY dict_type" )).all() counts = dict(rows) missing = [dt for dt, n in APPENDIX_40.items() if counts.get(dt, 0) < n] return counts, missing def check_view_consistency(): """v_trait_value 聚合结果与 bre_trait_observation 表一致。""" from sqlalchemy import create_engine, text from app.config.setting import settings eng = create_engine(settings.DB_URI) q = ("SELECT trait_id, count(*), sum(value_numeric::float) " "FROM {src} WHERE is_deleted = false AND value_numeric IS NOT NULL " "GROUP BY trait_id ORDER BY trait_id") with eng.connect() as c: tbl = {r[0]: (r[1], r[2]) for r in c.execute(text(q.format(src="bre_trait_observation")))} vw = {r[0]: (r[1], r[2]) for r in c.execute(text(q.format(src="v_trait_value")))} return tbl == vw def main_(): with TestClient(create_app()) as client: login(client) RUN = __import__("datetime").datetime.now().strftime("%H%M%S") passed = 0 def expect(name, cond, detail=""): nonlocal passed ok = bool(cond) print(f"[{'OK ' if ok else 'FAIL'}] {name}" + (f" ({detail})" if detail and not ok else "")) if not ok: sys.exit(1) passed += 1 print("\n=== 6.2 字典: 规划附录 40 枚举完整性 ===") counts, missing = check_dict_appendix() expect("40 枚举全部落库(值数达标)", not missing, f"缺失/不足: {missing}") print("\n=== 6.3 v_trait_value 视图与源表一致 ===") expect("视图聚合 == bre_trait_observation 聚合", check_view_consistency()) print("\n=== 6.1 编号自动生成 ===") tgt = call(client, "POST", "/api/v1/bre/target/create", json={"target_name": f"波次6目标{RUN}"}) expect("target.create", tgt[0] == 200, tgt[1]) tgt_id = tgt[1]["data"]["id"] comb = call(client, "POST", "/api/v1/bre/cross_combination/create", json={"cross_year": 2026, "bre_target_id": tgt_id, "remark": f"波次6{RUN}"}) expect("组合留空编号 create -> 200", comb[0] == 200, comb[1]) code = comb[1]["data"]["combination_code"] expect(f"自动生成编号格式 YY+3位 ({code})", isinstance(code, str) and re.fullmatch(r"\d{5}", code), code) comb_id = comb[1]["data"]["id"] t1 = call(client, "POST", "/api/v1/bre/tree/create", json={"combination_id": comb_id, "status": "alive", "remark": f"单株1{RUN}"}) expect("单株留空编号 create -> 200", t1[0] == 200, t1[1]) tn1 = t1[1]["data"]["tree_no"] expect(f"自动生成单株编号 = 组合-01 ({tn1})", tn1 == f"{code}-01", tn1) t2 = call(client, "POST", "/api/v1/bre/tree/create", json={"combination_id": comb_id, "status": "alive", "remark": f"单株2{RUN}"}) expect("第二株自动编号 = 组合-02", t2[0] == 200 and t2[1]["data"]["tree_no"] == f"{code}-02", t2[1]) tree1_id = t1[1]["data"]["id"] tree2_id = t2[1]["data"]["id"] print("\n=== 6.4 数值门禁 valid_min/max ===") trait = call(client, "POST", "/api/v1/bre/trait/create", json={"trait_code": f"W6_{RUN}", "trait_name": f"波次6性状{RUN}", "data_type": "numeric", "unit": "cm", "valid_min": 0, "valid_max": 10}) expect("trait.create(valid 0-10)", trait[0] == 200, trait[1]) trait_id = trait[1]["data"]["id"] obs_in = call(client, "POST", "/api/v1/bre/trait_observation/create", json={"combination_id": comb_id, "tree_id": tree1_id, "trait_id": trait_id, "value_numeric": 5, "evaluate_year": 2026}) expect("区间内数值 5 放行", obs_in[0] == 200, obs_in[1]) obs_high = call(client, "POST", "/api/v1/bre/trait_observation/create", json={"combination_id": comb_id, "tree_id": tree2_id, "trait_id": trait_id, "value_numeric": 15, "evaluate_year": 2026}) expect("数值 15 > valid_max=10 -> 409", obs_high[0] == 409, obs_high[1]) obs_low = call(client, "POST", "/api/v1/bre/trait_observation/create", json={"combination_id": comb_id, "tree_id": tree2_id, "trait_id": trait_id, "value_numeric": -1, "evaluate_year": 2026}) expect("数值 -1 < valid_min=0 -> 409", obs_low[0] == 409, obs_low[1]) obs_in_id = obs_in[1]["data"]["id"] upd = call(client, "PUT", f"/api/v1/bre/trait_observation/update/{obs_in_id}", json={"value_numeric": 99}) expect("更新越界 -> 409", upd[0] == 409, upd[1]) upd2 = call(client, "PUT", f"/api/v1/bre/trait_observation/update/{obs_in_id}", json={"value_numeric": 8}) expect("更新区间内 -> 200", upd2[0] == 200, upd2[1]) print("\n=== 6.5 状态机 单株状态 ===") st_fwd = call(client, "PUT", f"/api/v1/bre/tree/update/{tree1_id}", json={"status": "selected"}) expect("存活->入选 放行", st_fwd[0] == 200, st_fwd[1]) st_bwd = call(client, "PUT", f"/api/v1/bre/tree/update/{tree1_id}", json={"status": "alive"}) expect("入选->存活 回退 409", st_bwd[0] == 409, st_bwd[1]) st_end = call(client, "PUT", f"/api/v1/bre/tree/update/{tree1_id}", json={"status": "eliminated"}) expect("入选->淘汰(向前跳转) 放行", st_end[0] == 200, st_end[1]) print("\n=== 6.5 状态机 选育结论 ===") sr = call(client, "POST", "/api/v1/bre/selection_result/create", json={"combination_id": comb_id, "tree_id": tree1_id, "is_selected": "selected", "selection_year": 2026}) expect("选育结论=入选 create", sr[0] == 200, sr[1]) sr_id = sr[1]["data"]["id"] sr_fwd = call(client, "PUT", f"/api/v1/bre/selection_result/update/{sr_id}", json={"is_selected": "primary"}) expect("入选->初选 放行", sr_fwd[0] == 200, sr_fwd[1]) sr_bwd = call(client, "PUT", f"/api/v1/bre/selection_result/update/{sr_id}", json={"is_selected": "unknown"}) expect("初选->未知 回退 409", sr_bwd[0] == 409, sr_bwd[1]) print("\n=== 清理 ===") call(client, "DELETE", "/api/v1/bre/trait_observation/delete", json=[obs_in_id]) call(client, "DELETE", "/api/v1/bre/selection_result/delete", json=[sr_id]) call(client, "DELETE", "/api/v1/bre/tree/delete", json=[tree1_id, tree2_id]) call(client, "DELETE", "/api/v1/bre/cross_combination/delete", json=[comb_id]) call(client, "DELETE", "/api/v1/bre/trait/delete", json=[trait_id]) call(client, "DELETE", "/api/v1/bre/target/delete", json=[tgt_id]) print(f"\nALL {passed} CHECKS PASS \u2705") if __name__ == "__main__": main_()