#!/usr/bin/env python3 """生产E2E实证: 成本模型研发归集维度 (cma-rd-dimension-20260905) 链路验证: A. 标准成本 POST project_type=platform_dev → research_flag 自动=1 (研发归集) B. 标准成本 POST 不传归集维度 → 默认 client_delivery/research_flag=0 (录入默认) C. 标准成本列表 ?project_type=platform_dev 过滤 → 只含研发行, 不含客户交付 D. 标准成本列表 ?research_flag=1 过滤 → 只含研发行 E. 实际成本 POST project_type=platform_dev → 落库 + 列表过滤命中 F. 存量迁移保守默认: 两表 internal 行数 > 0 (ALTER 后存量全 internal) G. 回归: 既有列表接口(标准/实际)返回 200 且 data 为列表 H. ORM schema: inspect 两表含 research_flag/project_type 列 + 索引 数据纪律: 临时行 product_code 带 E2E-RD- 前缀 + remark 带标记, 验证完成后 DB 精确清理并回查 COUNT=0。 用法: cd /root/cma-management/backend && venv/bin/python3 scripts/verify_cost_research_dimension_e2e.py """ import json import os import sys import urllib.request import urllib.error from datetime import datetime BASE = "http://127.0.0.1:8010" MARK = "【系统验证临时数据-验证后删除】" PREFIX = "E2E-RD-" STAMP = datetime.now().strftime("%H%M%S") sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) os.environ.setdefault("CMA_DB_HOST", "127.0.0.1") from app.database import get_db, get_engine # noqa: E402 from sqlalchemy import text # noqa: E402 from sqlalchemy import inspect as sa_inspect # noqa: E402 def http(method, path, body=None, headers=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(BASE + path, data=data, method=method) req.add_header("Content-Type", "application/json") for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.loads(r.read().decode()) except urllib.error.HTTPError as e: raw = e.read().decode() try: return e.code, json.loads(raw) except Exception: return e.code, {"detail": raw} def main(): failures = [] def check(name, cond, detail=""): tag = "PASS" if cond else "FAIL" print(f" [{tag}] {name}" + (f" | {detail}" if detail else "")) if not cond: failures.append(name) # ── 0. 登录(admin entity=1) ── st, r = http("POST", "/api/cma/auth/login", {"username": "admin", "password": "admin123", "entity_id": 1}) check("登录 admin entity=1", st == 200, f"http={st}") token = r.get("token") if st == 200 else "" auth = {"Authorization": f"Bearer {token}"} rd_code = f"{PREFIX}RD{STAMP}" cd_code = f"{PREFIX}CD{STAMP}" ac_code = f"{PREFIX}AC{STAMP}" try: # ── A. 标准成本: 平台研发 ── st, r = http("POST", "/api/cma/cost/standard-costs", { "product_code": rd_code, "product_name": "CMA平台化开发E2E", "cost_type": "labor", "item_name": "研发工时E2E", "standard_quantity": 10.0, "standard_price": 80.0, "unit": "小时", "project_type": "platform_dev", "remark": MARK, }, auth) check("POST standard platform_dev 200", st == 200, f"http={st} body={r}") # ── B. 标准成本: 不传归集维度 → 默认 client_delivery ── st, r = http("POST", "/api/cma/cost/standard-costs", { "product_code": cd_code, "product_name": "客户交付E2E", "cost_type": "labor", "item_name": "实施工时E2E", "standard_quantity": 5.0, "standard_price": 60.0, "unit": "小时", "remark": MARK, }, auth) check("POST standard 默认client_delivery 200", st == 200, f"http={st} body={r}") # ── C. 列表过滤: project_type=platform_dev ── st, r = http("GET", f"/api/cma/cost/standard-costs?project_type=platform_dev", None, auth) codes = [i["product_code"] for i in r.get("data", [])] check("GET standard filter=platform_dev 200", st == 200, f"http={st}") check("研发过滤含研发行", rd_code in codes, f"{rd_code} in {codes}") check("研发过滤不含客户交付行", cd_code not in codes, f"{cd_code} not in {codes}") # ── D. 列表过滤: research_flag=1 ── st, r = http("GET", "/api/cma/cost/standard-costs?research_flag=1", None, auth) codes_rf = [i["product_code"] for i in r.get("data", [])] check("GET standard filter=research_flag=1 含研发行", rd_code in codes_rf, f"{rd_code} in {codes_rf}") # ── E. 实际成本: platform_dev ── st, r = http("POST", "/api/cma/cost/actual-costs", { "period": "2026-09", "product_code": ac_code, "product_name": "CMA平台化开发E2E", "cost_type": "labor", "item_name": "研发工时E2E", "actual_quantity": 40.0, "actual_price": 100.0, "project_type": "platform_dev", "remark": MARK, }, auth) check("POST actual platform_dev 200", st == 200, f"http={st} body={r}") st, r = http("GET", "/api/cma/cost/actual-costs?project_type=platform_dev&period=2026-09", None, auth) ac_codes = [i["product_code"] for i in r.get("data", [])] check("GET actual filter=platform_dev 含研发行", ac_code in ac_codes, f"{ac_code} in {ac_codes}") # ── DB 落库校验 ── db = next(get_db()) try: row_s = db.execute(text( "SELECT project_type, research_flag, entity_id FROM standard_costs WHERE product_code = :c" ), {"c": rd_code}).fetchone() check("DB standard 落库 project_type=platform_dev", row_s and row_s[0] == "platform_dev", f"row={row_s}") check("DB standard research_flag 联动=1", row_s and row_s[1] == 1, f"row={row_s}") check("DB standard entity_id=1(token绑定)", row_s and row_s[2] == 1, f"row={row_s}") row_c = db.execute(text( "SELECT project_type, research_flag FROM standard_costs WHERE product_code = :c" ), {"c": cd_code}).fetchone() check("DB standard 默认 project_type=client_delivery", row_c and row_c[0] == "client_delivery", f"row={row_c}") check("DB standard 默认 research_flag=0", row_c and row_c[1] == 0, f"row={row_c}") row_a = db.execute(text( "SELECT project_type, research_flag FROM actual_costs WHERE product_code = :c" ), {"c": ac_code}).fetchone() check("DB actual 落库 platform_dev+flag=1", row_a and row_a[0] == "platform_dev" and row_a[1] == 1, f"row={row_a}") # ── F. 存量保守默认: internal 行仍在(ALTER后存量 internal, 未误归研发) ── n_std_internal = db.execute(text( "SELECT COUNT(*) FROM standard_costs WHERE project_type = 'internal' AND product_code NOT LIKE :p" ), {"p": f"{PREFIX}%"}).scalar() n_act_internal = db.execute(text( "SELECT COUNT(*) FROM actual_costs WHERE project_type = 'internal' AND product_code NOT LIKE :p" ), {"p": f"{PREFIX}%"}).scalar() check("存量 standard internal 保留(29)", n_std_internal >= 29, f"count={n_std_internal}") check("存量 actual internal 保留(134)", n_act_internal >= 134, f"count={n_act_internal}") # ── H. ORM schema: 列+索引存在 ── insp = sa_inspect(get_engine()) for t in ("standard_costs", "actual_costs"): cols = {c["name"] for c in insp.get_columns(t)} idxs = {ix["name"] for ix in insp.get_indexes(t)} check(f"{t} 含 research_flag+project_type 列", {"research_flag", "project_type"} <= cols, f"cols={sorted(cols)}") check(f"{t} 含两索引", {f"ix_{t}_research_flag", f"ix_{t}_project_type"} <= idxs, f"idx={sorted(idxs)}") finally: db.rollback() db.close() # ── G. 回归: 原列表接口正常 ── st, r = http("GET", "/api/cma/cost/standard-costs", None, auth) check("回归 GET standard-costs(全量) 200+list", st == 200 and isinstance(r.get("data"), list), f"http={st} n={len(r.get('data', []))}") st, r = http("GET", "/api/cma/cost/actual-costs?period=2026-09", None, auth) check("回归 GET actual-costs 200+list", st == 200 and isinstance(r.get("data"), list), f"http={st} n={len(r.get('data', []))}") finally: # ── 清理临时数据(DB直删标记行) + 回查 ── db = next(get_db()) try: for t in ("standard_costs", "actual_costs"): db.execute(text(f"DELETE FROM {t} WHERE product_code LIKE :p"), {"p": f"{PREFIX}%"}) db.commit() for t in ("standard_costs", "actual_costs"): n = db.execute(text(f"SELECT COUNT(*) FROM {t} WHERE product_code LIKE :p"), {"p": f"{PREFIX}%"}).scalar() check(f"清理回查 {t} 标记行=0", n == 0, f"count={n}") finally: db.rollback() db.close() print() if failures: print(f"RESULT: FAIL ({len(failures)}): {failures}") sys.exit(1) print("RESULT: ALL PASS") sys.exit(0) if __name__ == "__main__": main()