378 lines
20 KiB
Python
378 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""近交衰退正式 tc 套件：TestClient 走真实 API + 方向感知/深层系谱判别（纯计算不落库）。
|
||
|
||
fixture（镜像 e2e_inbreeding_depression）：6 基源种质（双作），3 组合；12 树
|
||
（前 6 非自交 dam≠sire → F=0；后 6 自交 dam==sire → F=0.5）。观测：
|
||
- tD 衰退型 y=100−60F+ε（desc） - tN 无衰退型 y≈100 常数
|
||
- tA 病指型 y=10+40F（asc：F↑→指标↑=恶化，方向盲目的 slope<0 判据会误报「未检出」）
|
||
经真实 HTTP 端点 POST /api/v1/bre/statistics/inbreeding-depression 断言：
|
||
[1] 衰退：slope<-40、t<-3、r²>0.9、has_depression=true、depression_rate<0、
|
||
max_F≈0.5、mean_F≈0.25、direction=desc、high_inbreeding 前6=自交(F=0.5)后4=F=0
|
||
[2] 无衰退：slope≈0、has_depression=false、文案「未检出」
|
||
[3] 样本不足：min_n=20 → sufficient=false + reason 含样本量/阈值、无回归
|
||
[4] asc 方向：slope>30、t>3、has_depression=true、文案含「上升」（方向感知）
|
||
[5] 深层系谱（P→G7/G8 升谱，候选树 dam=G7 sire=G8 半同胞近交 F=0.125 不低估）
|
||
[6] 性状不存在 → 409
|
||
依赖: Redis + PG 正常（TestClient 走真实 lifespan）。运行后自动清理。
|
||
"""
|
||
import os
|
||
os.environ["ENVIRONMENT"] = "dev"
|
||
os.environ["PYTHONUTF8"] = "1"
|
||
|
||
import sys, asyncio # noqa: E402
|
||
sys.path.insert(0, r"d:\dpb\dpb\backend")
|
||
|
||
import main # noqa: E402
|
||
from fastapi.testclient import TestClient # noqa: E402
|
||
from sqlalchemy import delete, select # noqa: E402
|
||
|
||
from app.core.database import create_async_engine_and_session # noqa: E402
|
||
from app.api.v1.module_system.user.model import UserModel # noqa: E402 (注册 mapper)
|
||
from app.api.v1.module_bre.target.model import TargetModel # noqa: E402
|
||
from app.api.v1.module_bre.trait.model import TraitModel # noqa: E402
|
||
from app.api.v1.module_bre.germplasm.model import BreedingGermplasmModel # noqa: E402
|
||
from app.api.v1.module_bre.cross_combination.model import CrossCombinationModel # noqa: E402
|
||
from app.api.v1.module_bre.tree.model import TreeModel # noqa: E402
|
||
from app.api.v1.module_bre.trait_observation.model import TraitObservationModel # noqa: E402
|
||
|
||
create_app = main.create_app
|
||
TOKEN = None
|
||
ok, fail = 0, 0
|
||
|
||
PREFIX = "TCIBRD"
|
||
tokens: dict[str, list[int]] = {
|
||
"trait": [], "tree": [], "combo": [], "germ": [], "target": [], "obs": [],
|
||
}
|
||
FIX: dict = {}
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
global ok, fail
|
||
if cond:
|
||
ok += 1
|
||
print(f" [ok] {name} {detail}")
|
||
else:
|
||
fail += 1
|
||
print(f" [FAIL] {name} {detail}")
|
||
|
||
|
||
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}"}
|
||
|
||
|
||
async def _wipe() -> None:
|
||
"""启动前清理本前缀残留（防上次进程被杀/建 fixture 中途失败留下的脏数据）。"""
|
||
engine, sf = create_async_engine_and_session()
|
||
try:
|
||
async with sf() as db:
|
||
trait_ids = list((await db.execute(select(TraitModel.id).where(
|
||
TraitModel.trait_code.like(f"%_{PREFIX}")))).scalars())
|
||
if trait_ids:
|
||
obs = list((await db.execute(select(TraitObservationModel.id).where(
|
||
TraitObservationModel.trait_id.in_(trait_ids)))).scalars())
|
||
if obs:
|
||
await db.execute(delete(TraitObservationModel).where(
|
||
TraitObservationModel.id.in_(obs)))
|
||
tree_ids = list((await db.execute(select(TreeModel.id).where(
|
||
TreeModel.tree_no.like(f"{PREFIX}-%")))).scalars())
|
||
if tree_ids:
|
||
await db.execute(delete(TreeModel).where(TreeModel.id.in_(tree_ids)))
|
||
combo_ids = list((await db.execute(select(CrossCombinationModel.id).where(
|
||
CrossCombinationModel.combination_code.like(f"C%_{PREFIX}")))).scalars())
|
||
if combo_ids:
|
||
await db.execute(delete(CrossCombinationModel).where(
|
||
CrossCombinationModel.id.in_(combo_ids)))
|
||
germ_ids = list((await db.execute(select(BreedingGermplasmModel.id).where(
|
||
BreedingGermplasmModel.cultivar_name.like(f"%_{PREFIX}")))).scalars())
|
||
if germ_ids:
|
||
await db.execute(delete(BreedingGermplasmModel).where(
|
||
BreedingGermplasmModel.id.in_(germ_ids)))
|
||
if trait_ids:
|
||
await db.execute(delete(TraitModel).where(TraitModel.id.in_(trait_ids)))
|
||
await db.execute(delete(TargetModel).where(TargetModel.target_name == f"目标{PREFIX}"))
|
||
await db.commit()
|
||
print(f"[preclean] inbreeding tc 前缀残留已清（trait={len(trait_ids)} tree={len(tree_ids)}）")
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _build_fixture() -> None:
|
||
engine, sf = create_async_engine_and_session()
|
||
try:
|
||
async with sf() as db:
|
||
tD = TraitModel(trait_code=f"dD_{PREFIX}", trait_name=f"衰退型{PREFIX}", data_type="numeric",
|
||
unit="g", is_core="1", direction="desc", into_ebv="1",
|
||
default_h2=0.5, stage="evaluation", created_id=1)
|
||
tN = TraitModel(trait_code=f"dN_{PREFIX}", trait_name=f"无衰退型{PREFIX}", data_type="numeric",
|
||
unit="g", is_core="1", direction="desc", into_ebv="1",
|
||
default_h2=0.5, stage="evaluation", created_id=1)
|
||
tA = TraitModel(trait_code=f"dA_{PREFIX}", trait_name=f"病指型{PREFIX}", data_type="numeric",
|
||
unit="%", is_core="1", direction="asc", into_ebv="1",
|
||
default_h2=0.5, stage="evaluation", created_id=1)
|
||
db.add_all([tD, tN, tA])
|
||
await db.flush()
|
||
tokens["trait"] += [tD.id, tN.id, tA.id]
|
||
FIX["D"], FIX["N"], FIX["A"] = tD, tN, tA
|
||
|
||
target = TargetModel(target_name=f"目标{PREFIX}", created_id=1)
|
||
db.add(target)
|
||
await db.flush()
|
||
tokens["target"].append(target.id)
|
||
|
||
germs = []
|
||
for i in range(6):
|
||
g = BreedingGermplasmModel(cultivar_name=f"G{i}_{PREFIX}",
|
||
can_be_female=True, can_be_male=True, created_id=1)
|
||
db.add(g)
|
||
germs.append(g)
|
||
await db.flush()
|
||
tokens["germ"] += [g.id for g in germs]
|
||
|
||
combos = []
|
||
for i in range(3):
|
||
c = CrossCombinationModel(
|
||
combination_code=f"C{i}_{PREFIX}", bre_target_id=target.id,
|
||
female_parent_id=germs[2 * i].id, male_parent_id=germs[2 * i + 1].id,
|
||
design_type="full_diallel", created_id=1,
|
||
)
|
||
db.add(c)
|
||
combos.append(c)
|
||
await db.flush()
|
||
tokens["combo"] += [c.id for c in combos]
|
||
|
||
trees: list[TreeModel] = []
|
||
for i in range(6):
|
||
gA, gB = germs[2 * (i % 3)].id, germs[2 * (i % 3) + 1].id
|
||
t = TreeModel(combination_id=combos[i % 3].id, tree_no=f"{PREFIX}-N{i:02d}",
|
||
status="alive", stage="evaluation", generation="F1",
|
||
dam_id=gA, sire_id=gB, created_id=1)
|
||
db.add(t)
|
||
trees.append(t)
|
||
for i in range(6):
|
||
g = germs[i % 3].id # dam==sire → 自交，F=0.5
|
||
t = TreeModel(combination_id=combos[i % 3].id, tree_no=f"{PREFIX}-S{i:02d}",
|
||
status="alive", stage="evaluation", generation="F1",
|
||
dam_id=g, sire_id=g, created_id=1)
|
||
db.add(t)
|
||
trees.append(t)
|
||
await db.flush()
|
||
tokens["tree"] += [t.id for t in trees]
|
||
|
||
obs_rows = []
|
||
for i, t in enumerate(trees):
|
||
f = 0.5 if i >= 6 else 0.0
|
||
oD = TraitObservationModel(tree_id=t.id, combination_id=combos[i % 3].id,
|
||
trait_id=tD.id, value_numeric=100.0 - 60.0 * f + (i % 5) * 0.3,
|
||
evaluate_year=2025, created_id=1)
|
||
oN = TraitObservationModel(tree_id=t.id, combination_id=combos[i % 3].id,
|
||
trait_id=tN.id, value_numeric=100.0 + (i % 3) * 0.1,
|
||
evaluate_year=2025, created_id=1)
|
||
oA = TraitObservationModel(tree_id=t.id, combination_id=combos[i % 3].id,
|
||
trait_id=tA.id, value_numeric=10.0 + 40.0 * f,
|
||
evaluate_year=2025, created_id=1)
|
||
db.add_all([oD, oN, oA])
|
||
obs_rows += [oD, oN, oA]
|
||
await db.flush()
|
||
tokens["obs"] += [o.id for o in obs_rows]
|
||
|
||
# ---- [5] 深层系谱：P→G7/G8（tree FK 升谱），候选树 dam=G7 sire=G8 半同胞近交 F=0.125 ----
|
||
gP = BreedingGermplasmModel(cultivar_name=f"GP_{PREFIX}", can_be_female=True, can_be_male=True, created_id=1)
|
||
g7 = BreedingGermplasmModel(cultivar_name=f"G7_{PREFIX}", can_be_female=True, can_be_male=True, created_id=1)
|
||
g8 = BreedingGermplasmModel(cultivar_name=f"G8_{PREFIX}", can_be_female=True, can_be_male=True, created_id=1)
|
||
db.add_all([gP, g7, g8])
|
||
await db.flush()
|
||
tokens["germ"] += [gP.id, g7.id, g8.id]
|
||
eX = TreeModel(combination_id=combos[0].id, tree_no=f"{PREFIX}-X01", status="alive",
|
||
stage="evaluation", generation="F1", germplasm_id=g7.id, dam_id=gP.id, created_id=1)
|
||
eY = TreeModel(combination_id=combos[0].id, tree_no=f"{PREFIX}-X02", status="alive",
|
||
stage="evaluation", generation="F1", germplasm_id=g8.id, dam_id=gP.id, created_id=1)
|
||
tD2 = TraitModel(trait_code=f"dD2_{PREFIX}", trait_name=f"深层系谱衰退型{PREFIX}", data_type="numeric",
|
||
unit="g", is_core="1", direction="desc", into_ebv="1",
|
||
default_h2=0.5, stage="evaluation", created_id=1)
|
||
db.add_all([eX, eY, tD2])
|
||
await db.flush()
|
||
tokens["tree"] += [eX.id, eY.id]
|
||
tokens["trait"].append(tD2.id)
|
||
FIX["D2"] = tD2
|
||
deep_trees: list[TreeModel] = []
|
||
for i in range(6):
|
||
t = TreeModel(combination_id=combos[0].id, tree_no=f"{PREFIX}-H{i:02d}", status="alive",
|
||
stage="evaluation", generation="F1", dam_id=g7.id, sire_id=g8.id, created_id=1)
|
||
db.add(t)
|
||
deep_trees.append(t)
|
||
for i in range(6):
|
||
t = TreeModel(combination_id=combos[0].id, tree_no=f"{PREFIX}-C{i:02d}", status="alive",
|
||
stage="evaluation", generation="F1", dam_id=germs[0].id, sire_id=germs[1].id, created_id=1)
|
||
db.add(t)
|
||
deep_trees.append(t)
|
||
await db.flush()
|
||
tokens["tree"] += [t.id for t in deep_trees]
|
||
obs2 = []
|
||
for i, t in enumerate(deep_trees):
|
||
f = 0.125 if i < 6 else 0.0
|
||
o = TraitObservationModel(tree_id=t.id, trait_id=tD2.id, value_numeric=100.0 - 60.0 * f + (i % 4) * 0.2,
|
||
evaluate_year=2025, created_id=1)
|
||
db.add(o)
|
||
obs2.append(o)
|
||
await db.flush()
|
||
tokens["obs"] += [o.id for o in obs2]
|
||
await db.commit()
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
async def _verify_and_cleanup() -> None:
|
||
engine, sf = create_async_engine_and_session()
|
||
try:
|
||
async with sf() as db:
|
||
if tokens["obs"]:
|
||
await db.execute(delete(TraitObservationModel).where(
|
||
TraitObservationModel.id.in_(tokens["obs"])))
|
||
if tokens["tree"]:
|
||
await db.execute(delete(TreeModel).where(TreeModel.id.in_(tokens["tree"])))
|
||
if tokens["combo"]:
|
||
await db.execute(delete(CrossCombinationModel).where(
|
||
CrossCombinationModel.id.in_(tokens["combo"])))
|
||
if tokens["germ"]:
|
||
await db.execute(delete(BreedingGermplasmModel).where(
|
||
BreedingGermplasmModel.id.in_(tokens["germ"])))
|
||
if tokens["trait"]:
|
||
await db.execute(delete(TraitModel).where(TraitModel.id.in_(tokens["trait"])))
|
||
if tokens["target"]:
|
||
await db.execute(delete(TargetModel).where(TargetModel.id.in_(tokens["target"])))
|
||
await db.commit()
|
||
print(f"[cleanup] inbreeding tc 数据已清（trait={len(tokens['trait'])} tree={len(tokens['tree'])} "
|
||
f"obs={len(tokens['obs'])}）")
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
def main_() -> None:
|
||
asyncio.run(_wipe())
|
||
asyncio.run(_build_fixture())
|
||
try:
|
||
with TestClient(create_app()) as client:
|
||
login(client)
|
||
H = auth()
|
||
code = {k: FIX[k].trait_code for k in ("D", "N", "A", "D2")}
|
||
|
||
def run(tid, tcode, min_n=10):
|
||
r = client.post("/api/v1/bre/statistics/inbreeding-depression",
|
||
json={"trait_id": tid, "trait_code": tcode,
|
||
"year": None, "trial_study_id": None, "min_n": min_n},
|
||
headers=H)
|
||
check("[HTTP] inbreeding-depression 200", r.status_code == 200,
|
||
f"{r.status_code} {str(r.text)[:120]}")
|
||
b = r.json()
|
||
return b.get("data") if b.get("code") == 0 else None
|
||
|
||
# ---- [1] 衰退性状 ----
|
||
rD = run(FIX["D"].id, code["D"])
|
||
check("[1] sufficient=True", rD is not None and rD.get("sufficient") is True,
|
||
f"{rD and rD.get('reason')}")
|
||
if rD:
|
||
reg = rD["regression"]
|
||
check("[1] n_obs=12", rD.get("n_obs") == 12, f"{rD.get('n_obs')}")
|
||
check("[1] slope<-40（≈−60）", reg["slope"] < -40, f"slope={reg['slope']:.2f}")
|
||
check("[1] r²>0.9", reg["r2"] > 0.9, f"r²={reg['r2']:.3f}")
|
||
check("[1] has_depression=true", reg["has_depression"] is True)
|
||
check("[1] t_slope<-3", reg["t_slope"] < -3, f"t={reg['t_slope']:.2f}")
|
||
check("[1] 衰退率<0", reg.get("depression_rate") is not None
|
||
and reg["depression_rate"] < 0, f"{reg.get('depression_rate')}")
|
||
check("[1] max_F≈0.5", abs(rD["max_F"] - 0.5) < 0.001, f"max_F={rD['max_F']}")
|
||
check("[1] mean_F≈0.25", 0.2 < rD["mean_F"] < 0.3, f"mean_F={rD['mean_F']}")
|
||
check("[1] direction=desc", rD["direction"] == "desc", f"{rD['direction']}")
|
||
check("[1] 文案含近交衰退", "近交衰退" in reg["interpretation"],
|
||
f"{reg['interpretation']}")
|
||
hi = rD["high_inbreeding"]
|
||
check("[1] high_inbreeding 10 条", len(hi) == 10, f"{len(hi)}")
|
||
check("[1] 前6=自交(F=0.5)+tree_no", all(abs(x["F"] - 0.5) < 0.001 for x in hi[:6])
|
||
and all(x.get("tree_no", "").startswith(f"{PREFIX}-S") for x in hi[:6]),
|
||
f"{[x.get('F') for x in hi[:6]]}")
|
||
check("[1] 后4=F=0", all(abs(x["F"]) < 1e-6 for x in hi[6:]))
|
||
|
||
# ---- [2] 无衰退性状 ----
|
||
rN = run(FIX["N"].id, code["N"])
|
||
check("[2] sufficient=True", rN is not None and rN.get("sufficient") is True)
|
||
if rN:
|
||
regN = rN["regression"]
|
||
check("[2] slope≈0", abs(regN["slope"]) < 5, f"slope={regN['slope']:.2f}")
|
||
check("[2] has_depression=false", regN["has_depression"] is False)
|
||
check("[2] t>-1.645", regN["t_slope"] > -1.645, f"t={regN['t_slope']:.2f}")
|
||
check("[2] 文案含未检出", "未检出" in regN["interpretation"], f"{regN['interpretation']}")
|
||
|
||
# ---- [3] 样本不足 ----
|
||
rLow = run(FIX["D"].id, code["D"], min_n=20)
|
||
check("[3] sufficient=false", rLow is not None and rLow.get("sufficient") is False)
|
||
if rLow:
|
||
check("[3] reason 含样本量+阈值", "12 棵" in rLow["reason"] and "≥20" in rLow["reason"],
|
||
f"{rLow['reason']}")
|
||
check("[3] 无回归/清单", rLow.get("regression") is None and rLow.get("high_inbreeding") == [])
|
||
|
||
# ---- [4] asc 方向（病指，越低越好）：F↑→指标↑=恶化 ----
|
||
rA = run(FIX["A"].id, code["A"])
|
||
check("[4] sufficient=True", rA is not None and rA.get("sufficient") is True)
|
||
if rA:
|
||
regA = rA["regression"]
|
||
check("[4] direction=asc", rA["direction"] == "asc", f"{rA['direction']}")
|
||
check("[4] slope>30（≈+40）", regA["slope"] > 30, f"slope={regA['slope']:.2f}")
|
||
check("[4] t>3", regA["t_slope"] > 3, f"t={regA['t_slope']:.2f}")
|
||
check("[4] has_depression=true（方向感知）", regA["has_depression"] is True)
|
||
check("[4] 文案含上升", "上升" in regA["interpretation"], f"{regA['interpretation']}")
|
||
check("[4] max_F>0.4", rA["max_F"] > 0.4, f"max_F={rA['max_F']}")
|
||
check("[4] 前6=F=0.5", all(abs(x["F"] - 0.5) < 0.001 for x in rA["high_inbreeding"][:6]))
|
||
|
||
# ---- [5] 深层系谱近交系数（多世代不低估 F） ----
|
||
rD2 = run(FIX["D2"].id, code["D2"])
|
||
check("[5] sufficient=True", rD2 is not None and rD2.get("sufficient") is True)
|
||
if rD2:
|
||
check("[5] n_obs=12", rD2.get("n_obs") == 12, f"{rD2.get('n_obs')}")
|
||
check("[5] max_F≈0.125（半同胞近交，浅层=0 低估）", abs(rD2["max_F"] - 0.125) < 0.001,
|
||
f"max_F={rD2['max_F']}")
|
||
check("[5] mean_F≈0.0625", abs(rD2["mean_F"] - 0.0625) < 0.001, f"mean_F={rD2['mean_F']}")
|
||
check("[5] slope<-40", rD2["regression"]["slope"] < -40,
|
||
f"slope={rD2['regression']['slope']:.2f}")
|
||
check("[5] has_depression=true", rD2["regression"]["has_depression"] is True)
|
||
hi2 = rD2["high_inbreeding"]
|
||
check("[5] 前6=F=0.125+H株", all(abs(x["F"] - 0.125) < 0.001 for x in hi2[:6])
|
||
and all(x.get("tree_no", "").startswith(f"{PREFIX}-H") for x in hi2[:6]),
|
||
f"{[x.get('F') for x in hi2[:6]]}")
|
||
|
||
# ---- [6] 校验 409 ----
|
||
r = client.post("/api/v1/bre/statistics/inbreeding-depression",
|
||
json={"trait_id": 99999999, "trait_code": f"dZ_{PREFIX}",
|
||
"year": None, "trial_study_id": None, "min_n": 10},
|
||
headers=H)
|
||
check("[6] 性状不存在 → 409", r.status_code == 409, f"{r.status_code}")
|
||
r = client.post("/api/v1/bre/statistics/inbreeding-depression",
|
||
json={"trait_id": FIX["D"].id, "trait_code": code["D"],
|
||
"year": None, "trial_study_id": None, "min_n": 1},
|
||
headers=H)
|
||
check("[6] min_n<2 → 422", r.status_code == 422, f"{r.status_code}")
|
||
finally:
|
||
asyncio.run(_verify_and_cleanup())
|
||
|
||
print(f"\n===== inbreeding tc 套件：ok={ok} fail={fail} =====")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main_()
|
||
sys.exit(1 if fail else 0)
|