# -*- coding: utf-8 -*- """票 12 运行时专项验证 — 商机详情工作计划 Tab 四端点（精简 A1X，复用 opportunity_work_plan）。 前置：服务运行中（含票 12 jar）；远程库可达；seed 基线在（a2=推进/A、a3=长期暂缓/A）。 模式复用 t06-verify.py：Sess(debug token) + 中文 form body；api_raw 返回完整信封供负路径断言。 Part L：清理 t12- 标记残留（幂等入口） Part A：add 主路径（逾期/临期样例）+ ROW_ADD 日志 + 空内容/超长 66001 Part U：update 部分更新（内容）+ 登记完成 finishTime 回填 + 取消完成清空 + planStatus 值域 66001 + 目标不存在 66002 Part D：delete 软删（list 不可见）+ ROW_DELETE 日志 + 不存在 66002 Part P：暂缓商机（a3）写禁三入口 66003（行由 DB 直插——API 被暂停守卫挡住，属豁免范围） 用法：python t12-verify.py [L|A|U|D|P ...] # 无参全量（按序） """ import io import sys from datetime import datetime, timedelta 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 = {'A': '744842565802524672', 'ADMIN': '739564171091247104'} MARK = 't12-' + datetime.now().strftime('%H%M%S') 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_raw(self, method, path, form=None, params=None): try: return self.raw(method, path, form, params).json() except Exception as e: print(f" [EXC] {path}: {e}") return None def api(self, method, path, form=None, params=None, step=''): b = self.api_raw(method, path, form, params) if b is None or b.get('code') != 0: print(f" [ERR] {step or path}: {b}") 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 find_opp(name_like): conn = db() try: cur = conn.cursor() cur.execute('SELECT id FROM opportunity WHERE opp_name LIKE %s AND deleted=0', (name_like + '%',)) rows = cur.fetchall() finally: conn.close() assert rows, f'未找到商机 {name_like}（seed 基线缺失？）' return rows[0][0] def list_plans(s, opp_id): return s.api('GET', '/api/opportunity/workplan/list', params={'oppId': opp_id}, step='workplan/list') or [] def by_id(opp_id, plan_id): return next((p for p in list_plans(s, opp_id) if str(p.get('id')) == str(plan_id)), None) def fmt(dt): return dt.strftime('%Y-%m-%d %H:%M:%S') plan = {} s = None # main() 注入的全局会话（by_id 依赖） def part_L(s, opp): # 幂等清理：删掉历史 t12- 残留行 for p in list_plans(s, opp): if str(p.get('planContent', '')).startswith('t12-'): s.api('POST', '/api/opportunity/workplan/delete', params={'id': p['id']}, step='cleanup delete') left = [p for p in list_plans(s, opp) if str(p.get('planContent', '')).startswith('t12-')] rec('L1 清理后无 t12- 残留', not left, f'{len(left)} 行残留') def part_A(s, opp): b = s.api_raw('POST', '/api/opportunity/workplan/add', form={'oppId': opp, 'planContent': ' '}) rec('A1 空内容 → 66001', bool(b) and b.get('code') == 66001, f"code={b and b.get('code')} msg={b and b.get('message')}") b = s.api_raw('POST', '/api/opportunity/workplan/add', form={'oppId': opp, 'planContent': '长' * 1001}) rec('A2 内容超 1000 字 → 66001', bool(b) and b.get('code') == 66001, f"code={b and b.get('code')} msg={b and b.get('message')}") b = s.api_raw('POST', '/api/opportunity/workplan/add', form={'oppId': opp, 'planContent': f'{MARK}-逾期样例（未完成已过截止）', 'deadline': fmt(datetime.now() - timedelta(days=2))}) ok = bool(b) and b.get('code') == 0 and b.get('data') rec('A3 add 逾期样例（deadline 过去 2 天）', ok, str(b)) plan['overdue'] = b.get('data') if ok else None b = s.api_raw('POST', '/api/opportunity/workplan/add', form={'oppId': opp, 'planContent': f'{MARK}-临期样例', 'deadline': fmt(datetime.now() + timedelta(days=3))}) ok = bool(b) and b.get('code') == 0 and b.get('data') rec('A4 add 临期样例（deadline 未来 3 天）', ok, str(b)) plan['due'] = b.get('data') if ok else None d = s.api('POST', '/api/opportunity/oplog/page', form={'oppId': opp, 'pageNum': 1, 'pageSize': 20}) hit = any(r.get('opKind') == 'ROW_ADD' and r.get('entityName') == 'opportunity_work_plan' for r in (d or {}).get('content', [])) rec('A5 ROW_ADD 日志（entityName=opportunity_work_plan）', hit, str(d)[:120] if d and not hit else '') def part_U(s, opp): pid = plan.get('due') assert pid, 'U 前置缺失：A 段未产出临期样例行' b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': pid, 'planContent': f'{MARK}-临期样例（改期后）'}) rec('U1 部分更新内容放行', bool(b) and b.get('code') == 0, str(b)) b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': pid, 'planStatus': 1}) row = by_id(opp, pid) ok = bool(b) and b.get('code') == 0 and row and row.get('planStatus') == 1 and row.get('finishTime') rec('U2 登记完成：planStatus=1 服务端回填 finishTime', ok, f"row.planStatus={row and row.get('planStatus')} finishTime={row and row.get('finishTime')}") b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': pid, 'planStatus': 0}) row = by_id(opp, pid) ok = bool(b) and b.get('code') == 0 and row and row.get('planStatus') == 0 and row.get('finishTime') is None rec('U3 取消完成：planStatus=0 清空 finishTime', ok, f"finishTime={row and row.get('finishTime')}") b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': pid, 'planStatus': 2}) rec('U4 planStatus=2 → 66001', bool(b) and b.get('code') == 66001, f"code={b and b.get('code')} msg={b and b.get('message')}") b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': 999, 'planStatus': 1}) rec('U5 编辑目标不存在 → 66002', bool(b) and b.get('code') == 66002, f"code={b and b.get('code')} msg={b and b.get('message')}") def part_D(s, opp): pid = plan.get('overdue') assert pid, 'D 前置缺失：A 段未产出逾期样例行' b = s.api_raw('POST', '/api/opportunity/workplan/delete', params={'id': pid}) left = by_id(opp, pid) ok = bool(b) and b.get('code') == 0 and left is None rec('D1 删除后 list 不可见', ok, f"code={b and b.get('code')} list残留={left is not None}") d = s.api('POST', '/api/opportunity/oplog/page', form={'oppId': opp, 'pageNum': 1, 'pageSize': 20}) hit = any(r.get('opKind') == 'ROW_DELETE' and r.get('entityName') == 'opportunity_work_plan' for r in (d or {}).get('content', [])) rec('D2 ROW_DELETE 日志', hit, '' if hit else str(d)[:120]) b = s.api_raw('POST', '/api/opportunity/workplan/delete', params={'id': pid}) rec('D3 重复删除（已不存在）→ 66002', bool(b) and b.get('code') == 66002, f"code={b and b.get('code')} msg={b and b.get('message')}") def part_P(s, a3): # 暂缓商机 API 进不来（守卫挡 add），行由 DB 直插造出——豁免口径：API 覆盖不到的关联数据 snow = (int(datetime.now().timestamp() * 1000) << 22) + 7 # 表 id 非自增（雪花），显式给值 conn = db() try: cur = conn.cursor() cur.execute( 'INSERT INTO opportunity_work_plan (id, opp_id, plan_content, deadline, plan_status, ' 'finish_time, delete_key, deleted, create_time, update_time, creator_id, updater_id) ' 'VALUES (%s, %s, %s, %s, 0, NULL, 0, 0, NOW(), NOW(), %s, %s)', (snow, a3, f'{MARK}-暂缓禁写样例', fmt(datetime.now() + timedelta(days=1)), '0', '0')) pid = snow finally: conn.close() b = s.api_raw('POST', '/api/opportunity/workplan/add', form={'oppId': a3, 'planContent': f'{MARK}-暂缓期新增'}) rec('P1 暂缓商机 add → 66003', bool(b) and b.get('code') == 66003, f"code={b and b.get('code')} msg={b and b.get('message')}") b = s.api_raw('POST', '/api/opportunity/workplan/update', form={'id': pid, 'planStatus': 1}) rec('P2 暂缓商机 update → 66003', bool(b) and b.get('code') == 66003, f"code={b and b.get('code')} msg={b and b.get('message')}") b = s.api_raw('POST', '/api/opportunity/workplan/delete', params={'id': pid}) rec('P3 暂缓商机 delete → 66003', bool(b) and b.get('code') == 66003, f"code={b and b.get('code')} msg={b and b.get('message')}") # 清理 DB 直插行（硬删，不入软删池） conn = db() try: cur = conn.cursor() cur.execute('DELETE FROM opportunity_work_plan WHERE id=%s', (pid,)) finally: conn.close() print(' [..] P 清理：DB 直插行已硬删') def main(): global s only = [a.upper() for a in sys.argv[1:]] or ['L', 'A', 'U', 'D', 'P'] a2 = find_opp('e2e-A-推进-智慧园区二期') a3 = find_opp('e2e-A-暂缓-数据中心改造') s = Sess(UID['A']) print(f'== 票 12 workplan 验证 | MARK={MARK} | a2={a2} a3={a3} ==') if 'L' in only: print('-- Part L 清理 --'); part_L(s, a2) if 'A' in only: print('-- Part A add --'); part_A(s, a2) if 'U' in only: print('-- Part U update --'); part_U(s, a2) if 'D' in only: print('-- Part D delete --'); part_D(s, a2) if 'P' in only: print('-- Part P 暂缓禁写 --'); part_P(s, a3) print(f'\n== 汇总：{sum(1 for _, ok, _ in results if ok)}/{len(results)} 通过 ==') if fails: print('失败项：') for c, ev in fails: print(f' ✗ {c}: {ev}') sys.exit(1) if __name__ == '__main__': main()