# -*- coding: utf-8 -*- """票 06 运行时专项验证 — D-14/15/16/17/20/21/22。 前置：服务运行中（含票 06 jar，jarcheck NEW-JAR-WITH-T06）；远程库可达。 模式复用 t05-verify.py：Sess(debug token) + 中文 form body；Result data=null 非失败。 Part C：D-14 create/edit 必填 —— create 缺 6 必填逐字段 66001「X不能为空」→ 全字段建主商机 A → edit 改空 66001 → edit 部分更新放行（merged 语义）。 Part A：D-21 attachment bizType —— 缺失/旧形态非法值 66001，OPP 放行。 Part D：D-16 detail 回显 —— followed/lastFollowTime/schemeCardStatus/schemeBudget 四字段。 Part S：D-17 submitCard 回写 —— set-primary 主客户卡提交后 DB 快照列 + detail 回显。 Part O：D-20 oplog 流转留痕 —— pause/resume/release-pool 落 STATUS_FLOW/opp_status 记录。 Part B：D-15 board 视图 —— summary/cards 同 viewType 同口径（MINE 不含公海、POOL 不含在办）。 Part V：D-22 versions DTO —— 三族响应不含 creatorId/updaterId/deleted。 CLEANUP：软删票06商机及子表行 + 停用/删除 e2e-t06 前缀规则与模板。 用法：python t06-verify.py [C|A|D|S|O|B|V|CLEANUP ...] # 无参全量（按序） """ import io, sys from datetime import datetime import requests sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') BASE = 'http://localhost:8080' UID = {'ADMIN': '739564171091247104'} # 罗伟健 管理员 = 商机负责人 TAG = 'e2e-t06-' + datetime.now().strftime('%H%M%S') ST_PATH = '/api/rule/opp-stage-template' PR_PATH = '/api/rule/opp-pool-rule' SC_PATH = '/api/rule/opp-scheme-template' results, fails = [], [] def rec(case, ok, ev=''): v = '✅' if ok else '❌' results.append((case, ok, ev)) if not ok: fails.append((case, ev)) print(f" [{v}] {case}" + (f" —— {ev}" if ev else '')) class Sess: def __init__(self, uid): r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=15) b = r.json() assert b.get('code') == 0, f'debug token 失败 {b}' self.h = {'Authorization': f"Bearer {b['data']}"} def raw(self, method, path, form=None, params=None): headers = dict(self.h) data = None if form is not None: headers['Content-Type'] = 'application/x-www-form-urlencoded' data = form r = requests.request(method, BASE + path, headers=headers, data=data, params=params, timeout=30) r.encoding = 'utf-8' return r def api(self, method, path, form=None, params=None, step=''): try: b = self.raw(method, path, form, params).json() except Exception as e: print(f" [EXC] {step or path}: {e}") return None if b.get('code') != 0: print(f" [ERR] {step or path}: code={b.get('code')} msg={b.get('message')}") return None return b.get('data') def db(): import pymysql return pymysql.connect(host='8.129.84.155', port=3306, user='root', password='Itc@123456', database='crm', charset='utf8mb4', autocommit=True) def db_rows(sql, args=None): conn = db() try: cur = conn.cursor() cur.execute(sql, args or ()) return cur.fetchall() finally: conn.close() S = None # admin Sess A_ID = None # 主商机（推进中，owner=admin） B_ID = None # 公海商机 # 全字段基础 form（票 06 D-14 后必填 6 项：名称/来源/招标形式/省/市/属地） def base_form(**over): f = {'opportunityName': f'票06-{TAG}', 'oppSource': 'opp_source_02', 'industryCode': 'gov', 'localityType': 'locality_type_01', 'bidForm': 'bid_form_01', 'provinceCode': '510000', 'cityCode': '510100', 'partyAClear': 1, 'partyA': '票06甲方', 'remark': '票06验证'} f.update(over) return f def expect_code(method, path, form=None, params=None, code=66001, contains=None, step=''): r = S.raw(method, path, form=form, params=params) b = r.json() msg = b.get('message') or '' ok = b.get('code') == code and (contains is None or contains in msg) return ok, f"code={b.get('code')} msg={msg[:60]}" # ==================== Part C：D-14 create/edit 必填 ==================== def part_c(): global A_ID print(f"\n===== Part C D-14 必填校验 @{TAG} =====") # C1-C6：逐字段缺省 → 66001 + 「X不能为空」 required = [('opportunityName', '商机名称'), ('oppSource', '商机来源'), ('bidForm', '招标形式'), ('provinceCode', '项目区域（省）'), ('cityCode', '项目区域（市）'), ('localityType', '项目属地')] for i, (key, label) in enumerate(required, 1): form = base_form() form[key] = '' ok, ev = expect_code('POST', '/api/opportunity', form=form, contains=f'{label}不能为空', step=f'C{i} 缺{label}') rec(f'C{i} create 缺{label} → 66001', ok, ev) # C7：全字段 → 成功建主商机 A A_ID = S.api('POST', '/api/opportunity', form=base_form(opportunityName=f'票06A-{TAG}'), step='C7 创建主商机A') rec('C7 create 全字段成功（建商机A）', bool(A_ID), f"id={A_ID}") # C8：edit 改空 → 66001（merged 后仍空；DTO 属性真名 oppName，create 入参才叫 opportunityName） ok, ev = expect_code('POST', '/api/opportunity/edit', form={'id': A_ID, 'oppName': ''}, contains='商机名称不能为空', step='C8 edit 改空') rec('C8 edit oppName 改空 → 66001', ok, ev) # C9：edit 部分更新（只传 oppName，其余 null 保留原值）→ 放行且不清库 # （票 06 教训回归断言：实体 "= \"\"" 初始化曾致未传字段被 SET 空串，Y2 实验实证） b = S.raw('POST', '/api/opportunity/edit', form={'id': A_ID, 'oppName': f'票06A改-{TAG}'}).json() row = db_rows('SELECT opp_name, opp_source, bid_form, province_code, city_code, locality_type ' 'FROM opportunity WHERE id=%s', (int(A_ID),))[0] ok = b.get('code') == 0 and row[0] == f'票06A改-{TAG}' \ and row[1] == 'opp_source_02' and row[2] == 'bid_form_01' \ and row[3] == '510000' and row[4] == '510100' and row[5] == 'locality_type_01' rec('C9 edit 部分更新放行且未传字段不清库', ok, f"code={b.get('code')} db={row}") # ==================== Part A：D-21 attachment bizType ==================== def part_a(): print(f"\n===== Part A D-21 bizType 值域 @{TAG} =====") # A1：缺 bizType → 66001 ok, ev = expect_code('POST', '/api/opportunity/attachment/add', form={'oppId': A_ID, 'fileId': '8101'}, contains='bizType', step='A1 缺bizType') rec('A1 attachment/add 缺 bizType → 66001', ok, ev) # A2：旧形态非法值 → 66001 ok, ev = expect_code('POST', '/api/opportunity/attachment/add', form={'oppId': A_ID, 'fileId': '8102', 'bizType': 'biz_type_01'}, contains='bizType', step='A2 非法bizType') rec("A2 bizType='biz_type_01'（非法值）→ 66001", ok, ev) # A3：合法值 OPP → 放行 b = S.raw('POST', '/api/opportunity/attachment/add', form={'oppId': A_ID, 'fileId': '8103', 'bizType': 'OPP', 'remark': '票06'}).json() rec("A3 bizType='OPP' 放行", b.get('code') == 0, f"code={b.get('code')} msg={b.get('message')}") # ==================== Part D：D-16 detail 回显 ==================== def part_d(): print(f"\n===== Part D D-16 detail 四字段回显 @{TAG} =====") d = S.api('GET', '/api/opportunity/detail', params={'id': A_ID}, step='D1 detail') ok = d is not None and d.get('followed') is False and d.get('lastFollowTime') is None \ and d.get('schemeCardStatus') == 0 rec('D1 初始 detail：followed=false lastFollowTime=null schemeCardStatus=0（无卡兜底）', ok, f"followed={d and d.get('followed')} lastFollowTime={d and d.get('lastFollowTime')} " f"cardStatus={d and d.get('schemeCardStatus')}") b = S.raw('POST', '/api/opportunity/follow/add', form={'oppId': A_ID, 'followContent': '票06回显验证', 'followTime': '2026-08-30 09:00:00', 'followWay': 'follow_way_01', 'customerId': '5001'}).json() assert b.get('code') == 0, f'前置 follow/add 失败: {b}' d = S.api('GET', '/api/opportunity/detail', params={'id': A_ID}, step='D2 detail after follow') rec('D2 follow/add 后 lastFollowTime 非空', bool(d and d.get('lastFollowTime')), f"lastFollowTime={d and d.get('lastFollowTime')}") b = S.raw('POST', '/api/opportunity/focus', params={'oppId': A_ID}).json() assert b.get('code') == 0, f'前置 focus 失败: {b}' d = S.api('GET', '/api/opportunity/detail', params={'id': A_ID}, step='D3 detail after focus') rec('D3 focus 后 followed=true', d is not None and d.get('followed') is True, f"followed={d and d.get('followed')}") # ==================== Part S：D-17 submitCard 回写 ==================== def form_fields(items): f = {} for i, n in enumerate(items): for k, v in n.items(): if v is not None: f[f'fields[{i}].{k}'] = v return f def make_card_template(): """现场造发布中方案卡模板（通用 applyScope=1，一显示字段，非必填）。""" defs = S.api('GET', f'{SC_PATH}/field-defs', step='field-defs') assert defs, 'field-defs 失败' fkey = defs[0]['fieldKey'] name = f'{TAG}-卡模板' S.api('POST', f'{SC_PATH}/save-draft', form={'templateName': name, 'applyScope': 1, 'isDefault': 0, 'templateDesc': '票06验证'}, step='卡模板草稿') rows = (S.api('POST', f'{SC_PATH}/page', form={'current': 1, 'size': 50, 'keyword': name}) or {}).get('content') or [] row = next((x for x in rows if x.get('templateName') == name and x.get('status') == 1), None) assert row, '卡模板草稿未找到' S.api('POST', f'{SC_PATH}/publish', form={'id': row['id'], 'templateName': name, 'applyScope': 1, 'isDefault': 0, **form_fields([{'fieldKey': fkey, 'isVisible': 1, 'isRequired': 0}])}, step='卡模板发布') rows = (S.api('POST', f'{SC_PATH}/page', form={'current': 1, 'size': 50, 'keyword': name}) or {}).get('content') or [] pub = next((x for x in rows if x.get('templateName') == name and x.get('status') == 2), None) assert pub, '卡模板未发布' return pub['id'] def part_s(): print(f"\n===== Part S D-17 submitCard 主表回写 @{TAG} =====") # S0：主客户设置 → primary_customer_id b = S.raw('POST', '/api/opportunity/customer/add', form={'oppId': A_ID, 'customerId': '5001', 'customerNameSnapshot': '客户甲', 'customerRole': 'customer_role_01', 'isPrimaryIntended': 0}).json() assert b.get('code') == 0, f'前置 customer/add 失败: {b}' b = S.raw('POST', '/api/opportunity/customer/set-primary', form={'oppId': A_ID, 'customerId': '5001'}).json() assert b.get('code') == 0, f'前置 set-primary 失败: {b}' pc = db_rows('SELECT primary_customer_id FROM opportunity WHERE id=%s', (int(A_ID),))[0][0] rec('S0 set-primary 后 primary_customer_id=5001', str(pc) == '5001', f"primary_customer_id={pc}") if str(pc) != '5001': return tpl_id = make_card_template() b = S.raw('POST', '/api/opportunity/scheme-card/save', form={'oppId': A_ID, 'customerId': '5001', 'templateId': tpl_id, 'biddingForm': 'bid_form_01', 'schemeBudget': '666.66'}).json() assert b.get('code') == 0, f'前置 scheme-card/save 失败: {b}' card_id = b.get('data') b = S.raw('POST', '/api/opportunity/scheme-card/submit', params={'id': card_id}).json() rec('S1 scheme-card/save(666.66)→submit 成功', b.get('code') == 0, f"code={b.get('code')} msg={b.get('message')}") row = db_rows('SELECT scheme_card_status, scheme_budget FROM opportunity WHERE id=%s', (int(A_ID),))[0] ok = row[0] == 2 and row[1] is not None and float(row[1]) == 666.66 rec('S2 DB 快照回写：scheme_card_status=2 scheme_budget=666.66', ok, f"db={row}") d = S.api('GET', '/api/opportunity/detail', params={'id': A_ID}, step='S3 detail') ok = d is not None and d.get('schemeCardStatus') == 2 and d.get('schemeBudget') is not None \ and float(d.get('schemeBudget')) == 666.66 rec('S3 detail 快照回显：schemeCardStatus=2 schemeBudget=666.66', ok, f"cardStatus={d and d.get('schemeCardStatus')} budget={d and d.get('schemeBudget')}") # ==================== Part O：D-20 oplog 流转留痕 ==================== def assert_flow(oid, old, new, source='MANUAL', step=''): rows = db_rows("SELECT old_value, new_value, op_source, op_user_id, op_desc FROM opportunity_oplog " "WHERE opp_id=%s AND log_type='STATUS_FLOW' AND field_name='opp_status' " "AND old_value=%s AND new_value=%s AND deleted=0", (int(oid), str(old), str(new))) if not rows: return False, '无匹配记录' if source and not any(str(r[2]) == source for r in rows): return False, f"op_source={set(r[2] for r in rows)}" return True, f"rows={len(rows)} {rows[0][2]} desc={rows[0][4]}" def part_o(): print(f"\n===== Part O D-20 流转 oplog 留痕 @{TAG} =====") b = S.raw('POST', '/api/opportunity/pause', params={'id': A_ID}).json() assert b.get('code') == 0, f'前置 pause 失败: {b}' ok, ev = assert_flow(A_ID, 2, 3, 'MANUAL') rec('O1 pause(2→3) 落 STATUS_FLOW/opp_status op_source=MANUAL', ok, ev) b = S.raw('POST', '/api/opportunity/resume', params={'id': A_ID}).json() assert b.get('code') == 0, f'前置 resume 失败: {b}' ok, ev = assert_flow(A_ID, 3, 2, 'MANUAL') rec('O2 resume(3→2) 落 STATUS_FLOW/opp_status', ok, ev) # 商机 B：建 → release-pool（2→1），同时为 Part B 备公海数据 global B_ID B_ID = S.api('POST', '/api/opportunity', form=base_form(opportunityName=f'票06B-{TAG}'), step='O3 创建商机B') assert B_ID, '创建商机B失败' b = S.raw('POST', '/api/opportunity/release-pool', form={'id': B_ID, 'poolReason': 'pool_reason_02'}).json() assert b.get('code') == 0, f'前置 release-pool 失败: {b}' ok, ev = assert_flow(B_ID, 2, 1, 'MANUAL') rec('O3 release-pool(2→1) 落 STATUS_FLOW/opp_status', ok, ev) # ==================== Part B：D-15 board 视图过滤 ==================== def part_b(): print(f"\n===== Part B D-15 board 视图过滤 @{TAG} =====") sa = db_rows('SELECT current_stage_id, opp_status, owner_user_id FROM opportunity WHERE id=%s', (int(A_ID),))[0] sb = db_rows('SELECT current_stage_id, opp_status, owner_user_id FROM opportunity WHERE id=%s', (int(B_ID),))[0] rec('B0 前置状态：A 推进中 owner=admin；B 公海 owner=null', sa[1] == 2 and sb[1] == 1 and str(sa[2]) == UID['ADMIN'] and sb[2] is None, f"A(status={sa[1]},owner={sa[2]},stage={sa[0]}) B(status={sb[1]},owner={sb[2]},stage={sb[0]})") # cards：MINE 视图 A 在 B 不在；PUBLIC_POOL 视图 B 在 A 不在（各自 stageId 上交叉断言） def cards_ids(view, stage): d = S.api('POST', '/api/opportunity/board/cards', params={'viewType': view, 'stageId': stage, 'offset': 0, 'limit': 50}, step=f'cards {view}') return set(str(x['id']) for x in (d or [])) mine_a = cards_ids('MINE', sa[0]) pool_a = cards_ids('PUBLIC_POOL', sa[0]) rec('B1 cards(MINE, A_stage) 含 A 不含 B', A_ID in mine_a and B_ID not in mine_a, f"ids={sorted(mine_a)}") rec('B2 cards(PUBLIC_POOL, A_stage) 不含 A（owner 条件排除在办）', A_ID not in pool_a, f"ids={sorted(pool_a)}") if sb[0] != sa[0]: pool_b = cards_ids('PUBLIC_POOL', sb[0]) mine_b = cards_ids('MINE', sb[0]) rec('B3 cards(PUBLIC_POOL, B_stage) 含 B 不含 A', B_ID in pool_b and A_ID not in pool_b, f"ids={sorted(pool_b)}") rec('B4 cards(MINE, B_stage) 不含 B（视图过滤）', B_ID not in mine_b, f"ids={sorted(mine_b)}") # 默认 viewType = MINE（不传参数） d_def = S.api('POST', '/api/opportunity/board/cards', params={'stageId': sa[0]}, step='cards default') ids_def = set(str(x['id']) for x in (d_def or [])) rec('B5 cards 不传 viewType 默认 MINE（=B1 结果）', ids_def == mine_a, f"ids={sorted(ids_def)}") # summary 与 cards 同口径：MINE summary 含 A_stage 行；PUBLIC_POOL summary 不含 A 的归属误计 sm_mine = S.api('POST', '/api/opportunity/board/stage-summary', params={'viewType': 'MINE'}, step='summary MINE') or [] sm_pool = S.api('POST', '/api/opportunity/board/stage-summary', params={'viewType': 'PUBLIC_POOL'}, step='summary POOL') or [] mine_cnt = sum(int(m['count']) for m in sm_mine if str(m['stageId']) == str(sa[0])) pool_cnt = sum(int(m['count']) for m in sm_pool if str(m['stageId']) == str(sa[0])) # A_stage 上：MINE 口径 count≥1（含 A）；POOL 口径的 count 不含 A——若 B 与 A 同 stage， # POOL 该行 count≥1 但来源只能是 B，交叉用 cards 已证；此处断言 MINE 含 A、POOL 口径与 POOL cards 数一致 pool_cards_cnt = len(cards_ids('PUBLIC_POOL', sa[0])) rec('B6 stage-summary MINE 在 A_stage count≥1（含在办）', mine_cnt >= 1, f"count={mine_cnt}") rec('B7 stage-summary PUBLIC_POOL 与 cards 同口径（A_stage 行数一致）', pool_cnt == pool_cards_cnt, f"summary={pool_cnt} cards={pool_cards_cnt}") # ==================== Part V：D-22 versions DTO ==================== def versions_keys(path, table): row = db_rows(f'SELECT id FROM {table} WHERE deleted=0 ORDER BY id DESC LIMIT 1') assert row, f'{table} 无数据' d = S.api('GET', f'{path}/versions', params={'id': row[0][0]}, step=f'{path}/versions') assert d is not None and len(d) > 0, f'{path}/versions 空响应' keys = set(d[0].keys()) leaked = keys & {'creatorId', 'updaterId', 'deleted'} return keys, leaked, len(d) def part_v(): print(f"\n===== Part V D-22 versions DTO 化 @{TAG} =====") for path, table, biz in ((ST_PATH, 'opportunity_stage_template', 'templateCode'), (PR_PATH, 'opportunity_pool_rule', 'ruleCode'), (SC_PATH, 'opportunity_scheme_card_template', 'templateCode')): try: keys, leaked, n = versions_keys(path, table) rec(f"V {path}/versions DTO 化（{n} 行无泄露）", not leaked, f"leaked={leaked or '无'} bizField={'有' if biz in keys else '缺失!'}") except AssertionError as e: rec(f"V {path}/versions DTO 化", False, str(e)) # ==================== Part CLEANUP ==================== def part_cleanup(): print(f"\n===== CLEANUP 票06 验证数据 @ {TAG} =====") # 规则与模板：e2e-t06 前缀，发布中停用、草稿删除 for path, nmkey in ((PR_PATH, 'ruleName'), (SC_PATH, 'templateName')): rows = (S.api('POST', f'{path}/page', form={'current': 1, 'size': 50, 'keyword': 'e2e-t06-'}) or {}).get('content') or [] for x in rows: nm, rid, st = x.get(nmkey) or '', x.get('id'), x.get('status') if not nm.startswith('e2e-t06-'): continue if st == 2: S.api('POST', f'{path}/disable', params={'id': rid}, step=f'disable {nm}') elif st == 1: S.api('POST', f'{path}/delete', params={'id': rid}, step=f'delete {nm}') # 商机软删（票06 前缀 + 本次 TAG） rows = db_rows("SELECT id FROM opportunity WHERE opp_name LIKE %s OR remark LIKE %s", ('票06%', '票06%')) n = 0 for (oid,) in rows: db_rows('UPDATE opportunity SET deleted=1 WHERE id=%s', (oid,)) for tbl in ('opportunity_customer', 'opportunity_team'): # 两表 FK 列名 = opportunity_id db_rows(f'UPDATE {tbl} SET deleted=1 WHERE opportunity_id=%s', (oid,)) for tbl in ('opportunity_follow', 'opportunity_site_survey', 'opportunity_attachment', 'opportunity_scheme_card'): # 四表 FK 列名 = opp_id db_rows(f'UPDATE {tbl} SET deleted=1 WHERE opp_id=%s', (oid,)) db_rows('UPDATE opportunity_oplog SET deleted=1 WHERE opp_id=%s', (oid,)) n += 1 rec(f'CLEANUP 软删 {n} 个票06商机及子表行 + 停用自建规则/模板', True, '') PARTS = {'C': part_c, 'A': part_a, 'D': part_d, 'S': part_s, 'O': part_o, 'B': part_b, 'V': part_v, 'CLEANUP': part_cleanup} if __name__ == '__main__': args = [a.upper() for a in sys.argv[1:]] or ['C', 'A', 'D', 'S', 'O', 'B', 'V', 'CLEANUP'] S = Sess(UID['ADMIN']) for p in args: PARTS[p]() print(f"\n===== 汇总 {len(results)} 项，失败 {len(fails)} =====") for case, ev in fails: print(f" ❌ {case}: {ev}") sys.exit(1 if fails else 0)