"""票 05 运行时验证 — D-07 暂缓态写禁 + D-09 禁领接线。 前置：服务运行中（含票 05 jar，jarcheck NEW-JAR-WITH-T05）；远程库可达。 模式复用 t04-verify.py / e2e-rules.py：Sess(debug token) + 中文 form body；Result data=null 非失败。 Part P：D-07 暂缓写禁 —— 建商机造全量子表数据 → claim → pause → 14 个写端点逐个断言 66003 → 只读放行抽查 → resume → edit 恢复成功。 Part R：D-09 禁领 —— owner_dept 非空断言（票03修复前提）→ 部门专用 allowFreeClaim=0 发布 → claim/claim-batch 全量预检 66014 且无部分领取 → 部门专用 allowFreeClaim=1 顶替发布 → claim/claim-batch 放行。 CLEANUP：软删票05商机及子表行 + 停用/删除 e2e-t05 前缀规则与模板。 用法：python t05-verify.py [P|R|CLEANUP ...] # 无参全量 """ import io, sys, time from datetime import datetime import requests sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') BASE = 'http://localhost:8080' UID = {'ADMIN': '739564171091247104', # 罗伟健 管理员 = 商机负责人 'A': '744842565802524672', # 赖永利 'B': '744842566742048768', # 肖琴 'C': '744842318024015872'} # 曾偲青 TAG = 'e2e-t05-' + datetime.now().strftime('%H%M%S') SC_PATH = '/api/rule/opp-scheme-template' PR_PATH = '/api/rule/opp-pool-rule' 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 def ensure_opp(name, remark): oid = S.api('POST', '/api/opportunity', form={'opportunityName': name, 'oppSource': 'opp_source_02', 'industryCode': 'gov', 'localityType': 'locality_type_01', 'bidForm': 'bid_form_01', 'partyAClear': 1, 'partyA': '票05甲方', 'remark': remark}, step=f'创建 {name}') assert oid, f'创建商机失败: {name}' return str(oid) def expect_code(path, form=None, params=None, code=66003, step=''): r = S.raw('POST', path, form=form, params=params) b = r.json() return b.get('code') == code, f"code={b.get('code')} msg={(b.get('message') or '')[:50]}" 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': '票05验证'}, 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'] # ==================== Part P：D-07 暂缓态写禁 ==================== def part_p(): print(f"\n===== Part P 暂缓态写禁（D-07）@{TAG} =====") z = ensure_opp(f'票05暂缓Z-{TAG}', '票05 D-07 验证主商机') # ---- pause 前造数据 ---- r = S.raw('POST', '/api/opportunity/customer/add', form={'oppId': z, 'customerId': '5001', 'customerNameSnapshot': '客户甲', 'customerRole': 'customer_role_01', 'isPrimaryIntended': 0}) assert r.json().get('code') == 0, '前置 customer/add 失败' r = S.raw('POST', '/api/opportunity/team/add', form={'oppId': z, 'userIds': [UID['A'], UID['B']], 'projectRole': 'project_role_02', 'duty': '票05验证'}) b = r.json() assert b.get('code') == 0 and len(b.get('data') or []) == 2, '前置 team/add 失败' team_list = S.api('GET', '/api/opportunity/team/list', params={'oppId': z}, step='team/list') or [] row_a = next(x for x in team_list if str(x['userId']) == UID['A']) row_b = next(x for x in team_list if str(x['userId']) == UID['B']) r = S.raw('POST', '/api/opportunity/follow/add', form={'oppId': z, 'followContent': '暂停前进跟进', 'followTime': '2026-08-30 09:00:00', 'followWay': 'follow_way_01', 'customerId': '5001'}) assert r.json().get('code') == 0, '前置 follow/add 失败' fid = r.json().get('data') r = S.raw('POST', '/api/opportunity/site-survey/add', form={'oppId': z, 'surveyDate': '2026-08-30', 'engineerUserId': UID['A'], 'applyWay': 'apply_way_01', 'surveySeq': 'T05-S1'}) assert r.json().get('code') == 0, '前置 site-survey/add 失败' sid = r.json().get('data') r = S.raw('POST', '/api/opportunity/attachment/add', form={'oppId': z, 'fileId': '8001', 'bizType': 'biz_type_01', 'remark': '票05'}) assert r.json().get('code') == 0, '前置 attachment/add 失败' aid = r.json().get('data') tpl_id = make_card_template() r = S.raw('POST', '/api/opportunity/scheme-card/save', form={'oppId': z, 'customerId': '5001', 'templateId': tpl_id, 'biddingForm': 'bid_form_01'}) b = r.json() assert b.get('code') == 0, f'前置 scheme-card/save 失败: {b}' card_id = b.get('data') rec('P 前置数据就绪（客户/团队2/跟进/勘察/附件/卡）', True, f"follow={fid} survey={sid} att={aid} card={card_id}") # ---- pause（创建即推进中，2→3）---- b = S.raw('POST', '/api/opportunity/pause', params={'id': z}).json() assert b.get('code') == 0, f'pause 失败: {b}' st = db_rows('SELECT opp_status FROM opportunity WHERE id=%s', (int(z),))[0][0] rec('P0 claim→pause 后状态=3 暂缓中', st == 3, f"opp_status={st}") if st != 3: return # ---- 14 写端点逐个断言 66003 ---- cases = [ ('P1 edit 主表', '/api/opportunity/edit', {'id': z, 'opportunityName': '票05暂缓Z改名', 'oppSource': 'opp_source_02', 'industryCode': 'gov', 'localityType': 'locality_type_01', 'bidForm': 'bid_form_01', 'partyAClear': 1, 'partyA': '票05甲方'}), ('P2 follow/add', '/api/opportunity/follow/add', {'oppId': z, 'followContent': '暂缓期跟进应被拒', 'followTime': '2026-08-30 10:00:00', 'followWay': 'follow_way_01', 'customerId': '5001'}), ('P3 follow/delete', '/api/opportunity/follow/delete', None, {'id': fid}), ('P4 site-survey/add', '/api/opportunity/site-survey/add', {'oppId': z, 'surveyDate': '2026-08-31', 'engineerUserId': UID['B'], 'applyWay': 'apply_way_01', 'surveySeq': 'T05-S2'}), ('P5 site-survey/delete', '/api/opportunity/site-survey/delete', None, {'id': sid}), ('P6 attachment/add', '/api/opportunity/attachment/add', {'oppId': z, 'fileId': '8002', 'bizType': 'biz_type_01'}), ('P7 attachment/delete', '/api/opportunity/attachment/delete', None, {'id': aid}), ('P8 customer/add', '/api/opportunity/customer/add', {'oppId': z, 'customerId': '5002', 'customerNameSnapshot': '客户乙', 'customerRole': 'customer_role_02', 'isPrimaryIntended': 0}), ('P9 customer/set-primary', '/api/opportunity/customer/set-primary', {'oppId': z, 'customerId': '5001'}), ('P10 team/add', '/api/opportunity/team/add', {'oppId': z, 'userIds': [UID['C']], 'projectRole': 'project_role_02'}), ('P11 team/update', '/api/opportunity/team/update', {'id': row_b['id'], 'duty': '暂缓期改职责'}), ('P12 team/delete', '/api/opportunity/team/delete', None, {'id': row_a['id']}), ('P13 scheme-card/save', '/api/opportunity/scheme-card/save', {'id': card_id, 'oppId': z, 'customerId': '5001', 'templateId': tpl_id, 'schemeBudget': '888.88'}), ('P14 scheme-card/submit', '/api/opportunity/scheme-card/submit', None, {'id': card_id}), ] for c in cases: name, path = c[0], c[1] form = c[2] if len(c) > 2 else None params = c[3] if len(c) > 3 else None ok, ev = expect_code(path, form, params, code=66003, step=name) rec(name + ' → 66003', ok, ev) # ---- 拦截后无副作用抽查（团队行数不变、跟进未增）---- n_team = db_rows('SELECT COUNT(*) FROM opportunity_team WHERE opportunity_id=%s AND deleted=0 AND delete_key=0', (int(z),))[0][0] n_follow = db_rows('SELECT COUNT(*) FROM opportunity_follow WHERE opp_id=%s AND deleted=0 AND delete_key=0', (int(z),))[0][0] rec('P15 拦截无副作用（团队仍 3 行、跟进仍 1 条）', n_team == 3 and n_follow == 1, f"team={n_team} follow={n_follow}") # ---- 只读放行 + resume 恢复 ---- d = S.api('GET', '/api/opportunity/customer/list', params={'oppId': z}, step='只读 customer/list') rec('P16 只读放行：customer/list 正常', d is not None and len(d) == 1, f"rows={len(d or [])}") b = S.raw('POST', '/api/opportunity/resume', params={'id': z}).json() rec('P17 resume 成功（3→2）', b.get('code') == 0, f"code={b.get('code')}") b = S.raw('POST', '/api/opportunity/edit', form={'id': z, 'opportunityName': '票05暂缓Z恢复名', 'oppSource': 'opp_source_02', 'industryCode': 'gov', 'localityType': 'locality_type_01', 'bidForm': 'bid_form_01', 'partyAClear': 1, 'partyA': '票05甲方'}).json() rec('P18 resume 后 edit 恢复成功', b.get('code') == 0, f"code={b.get('code')} msg={b.get('message')}") # ==================== Part R：D-09 禁领接线 ==================== def to_pool(name): """建商机（创建即推进中）→ release-pool → 返回待领取商机 id。""" oid = ensure_opp(name, '票05 D-09 验证') b = S.raw('POST', '/api/opportunity/release-pool', form={'id': oid, 'poolReason': 'pool_reason_02'}).json() assert b.get('code') == 0, f'前置 release-pool 失败: {oid} {b}' return oid def pub_dept_rule(name, dept_id, allow): """发布部门专用规则（同部门重复发布会顶替停用旧版）。返回发布中行 id。""" S.api('POST', f'{PR_PATH}/save-draft', form={'ruleName': name, 'applyScope': 2, 'isDefault': 0, 'deptIds': dept_id, 'allowManualPool': 1, 'allowFreeClaim': allow, 'autoRecycleEnabled': 0, 'recycleRemindEnabled': 0}, step=f'{name} 草稿') rows = (S.api('POST', f'{PR_PATH}/page', form={'current': 1, 'size': 50, 'keyword': name}) or {}).get('content') or [] row = next((x for x in rows if x.get('ruleName') == name and x.get('status') == 1), None) assert row, f'规则草稿未找到: {name}' S.api('POST', f'{PR_PATH}/publish', form={'id': row['id'], 'ruleName': name, 'applyScope': 2, 'isDefault': 0, 'deptIds': dept_id, 'allowManualPool': 1, 'allowFreeClaim': allow, 'autoRecycleEnabled': 0, 'recycleRemindEnabled': 0}, step=f'{name} 发布') rows = (S.api('POST', f'{PR_PATH}/page', form={'current': 1, 'size': 50, 'keyword': name}) or {}).get('content') or [] pub = next((x for x in rows if x.get('ruleName') == name and x.get('status') == 2), None) assert pub, f'规则未发布: {name}' return pub['id'] def part_r(): print(f"\n===== Part R 禁领接线（D-09）@{TAG} =====") r1 = to_pool(f'票05禁领R1-{TAG}') dep = db_rows('SELECT owner_dept_id, opp_status FROM opportunity WHERE id=%s', (int(r1),))[0] owner_dept = str(dep[0]) if dep[0] else None rec('R0 owner_dept_id 非空（票03修复前提）', dep[1] == 1 and owner_dept is not None, f"owner_dept={owner_dept} status={dep[1]}") if not owner_dept: return # ---- 部门专用禁领（allowFreeClaim=0）---- pub0 = pub_dept_rule(f'{TAG}-禁领版', owner_dept, 0) ok, ev = expect_code('/api/opportunity/claim', None, {'id': r1}, code=66014) rec('R1 部门专用禁领 → claim 66014', ok, ev) r2 = to_pool(f'票05禁领R2-{TAG}') b = S.raw('POST', '/api/opportunity/claim-batch', params={'ids': [r1, r2]}).json() st = dict((str(x[0]), x[1]) for x in db_rows( 'SELECT id, opp_status FROM opportunity WHERE id IN (%s,%s)', (int(r1), int(r2)))) rec('R2 claim-batch 全量预检 → 66014 且两单均未被领取', b.get('code') == 66014 and st.get(str(r1)) == 1 and st.get(str(r2)) == 1, f"code={b.get('code')} status={st}") # ---- 部门专用放行（先停用禁领版——部门覆盖唯一约束 64012 硬拒同部门双发布）---- S.api('POST', f'{PR_PATH}/disable', params={'id': pub0}, step='停用禁领版') pub1 = pub_dept_rule(f'{TAG}-放行版', owner_dept, 1) b = S.raw('POST', '/api/opportunity/claim', params={'id': r1}).json() st1 = db_rows('SELECT opp_status FROM opportunity WHERE id=%s', (int(r1),))[0][0] rec('R3 部门专用放行 → claim 成功（status 1→2）', b.get('code') == 0 and st1 == 2, f"code={b.get('code')} status={st1}") b = S.raw('POST', '/api/opportunity/claim-batch', params={'ids': [r2]}).json() st2 = db_rows('SELECT opp_status FROM opportunity WHERE id=%s', (int(r2),))[0][0] rec('R4 claim-batch 放行 → 成功（status 1→2）', b.get('code') == 0 and st2 == 2, f"code={b.get('code')} status={st2}") # ==================== Part CLEANUP ==================== def part_cleanup(): print(f"\n===== CLEANUP 票05 验证数据 @ {TAG} =====") # 规则与模板：e2e-t05 前缀，发布中停用、草稿删除 for path, nmkey in ((PR_PATH, 'ruleName'), (SC_PATH, 'templateName')): rows = (S.api('POST', f'{path}/page', form={'current': 1, 'size': 50, 'keyword': 'e2e-t05-'}) 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-t05-'): 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}') # 商机软删（票05 前缀 + 本次 TAG） rows = db_rows("SELECT id FROM opportunity WHERE opp_name LIKE %s OR remark LIKE %s", ('票05%', '票05%')) 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} 个票05商机及子表行 + 停用自建规则/模板', True, '') PARTS = {'P': part_p, 'R': part_r, 'CLEANUP': part_cleanup} if __name__ == '__main__': args = [a.upper() for a in sys.argv[1:]] or ['P', 'R', '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)