You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
904 lines
48 KiB
904 lines
48 KiB
# -*- coding: utf-8 -*-
|
|
"""e2e-heavy.py — 客户重流程 API 级实测（票 05，F-13 交割 / F-14 导入 / F-15 提醒规则 / F-16 查重消费侧）
|
|
|
|
产出：report-heavy.md + specimens-heavy.json（票 06 合并素材）+ e2e-heavy-checks.json
|
|
输入：seed-ids.json（票 03）；e2e-core.py 同款框架（票 04 已验证基建）
|
|
纪律：flat 实路径；Result<Void> 判 code；401 重取 token；缺陷从 D-04 起；
|
|
判定依据 原型 > spec > 问用户；库检每步新连接 autocommit=True。
|
|
"""
|
|
import sys, io, json, time, re, os, random, subprocess, datetime
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
|
import requests
|
|
import pymysql
|
|
|
|
BASE = 'http://localhost:8080'
|
|
ADMIN = '739564171091247104' # 罗伟健（根部门「广东保伦」，data_scope=4）
|
|
BUDDY = '744842318024015872' # 曾偲青（无角色；f13 交割发起人）
|
|
IDS = json.load(open('.scratch/customer-e2e/seed-ids.json', encoding='utf-8'))
|
|
C = {k: v['id'] for k, v in IDS['customers'].items()}
|
|
NO = {k: v['no'] for k, v in IDS['customers'].items()}
|
|
OPP = {k: v['id'] for k, v in IDS['opportunities'].items()}
|
|
|
|
DEPT_SEED = 744700334353416192 # 华南销售部(seed)：f13 接收总监推导部门
|
|
DEPT_BUDDY_ORIG = 744841292483133440 # 职员部：BUDDY 原部门（teardown 还原）
|
|
SALES_ID = '760000000000000001' # e2c-销售甲 固定 id（幂等复用；字符串便于与 API/DB 比对）
|
|
SALES_NAME = 'e2c-销售甲'
|
|
JAVA_EXE = r'D:\IntelliJ IDEA 2025.1.3\jbr\bin\java.exe' # corretto-17 起服 >120s（第三轮实测），JBR 实测 20-40s
|
|
PFX = 'e2c-'
|
|
TS = datetime.datetime.now().strftime('%H%M%S')
|
|
|
|
checks, defects, specimens = [], [], {}
|
|
_skips = set()
|
|
|
|
|
|
def check(flow, case, verdict, detail=''):
|
|
checks.append({'flow': flow, 'case': case,
|
|
'verdict': {'pass': '✅', 'warn': '⚠', 'fail': '❌'}[verdict], 'detail': detail})
|
|
print(f" {verdict} [{flow}] {case}" + (f' — {detail}' if detail else ''))
|
|
|
|
|
|
def defect(did, severity, title, detail):
|
|
defects.append({'id': did, 'severity': severity, 'title': title, 'detail': detail})
|
|
print(f" ⚑ 登记 {did}({severity}): {title}")
|
|
|
|
|
|
def db():
|
|
return pymysql.connect(host='8.129.84.155', port=3306, user='root', password='Itc@123456',
|
|
database='crm', charset='utf8mb4', autocommit=True,
|
|
connect_timeout=10, cursorclass=pymysql.cursors.DictCursor)
|
|
|
|
|
|
def dbq(sql, args=None):
|
|
with db() as conn, conn.cursor() as cur:
|
|
cur.execute(sql, args)
|
|
return cur.fetchall()
|
|
|
|
|
|
def dbx(sql, args=None):
|
|
with db() as conn, conn.cursor() as cur:
|
|
cur.execute(sql, args)
|
|
return cur.rowcount
|
|
|
|
|
|
def get_token(uid, tries=3):
|
|
# debug/token 偶发瞬时故障（Redis Connection reset）→ 重试 + fail fast（票 04 教训）。
|
|
last = None
|
|
for _ in range(tries):
|
|
try:
|
|
r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=30)
|
|
if r.status_code == 200:
|
|
d = r.json().get('data')
|
|
tok = d if isinstance(d, str) else (d or {}).get('token')
|
|
if tok:
|
|
return tok
|
|
last = f'body={r.text[:120]}'
|
|
else:
|
|
last = f'HTTP {r.status_code} {r.text[:120]}'
|
|
except Exception as e:
|
|
last = str(e)
|
|
time.sleep(2)
|
|
raise RuntimeError(f'debug/token 获取失败 userId={uid}: {last}')
|
|
|
|
|
|
S = requests.Session()
|
|
S.headers['Authorization'] = f'Bearer {get_token(ADMIN)}'
|
|
B = requests.Session()
|
|
B.headers['Authorization'] = f'Bearer {get_token(BUDDY)}'
|
|
|
|
|
|
def _capture(method, path, req, resp_body):
|
|
path = re.sub(r'/\d{15,}', '/{id}', path) # 路径 id 归一化：票 06 按端点归档
|
|
key = f'{method} {path}'
|
|
if key in _skips:
|
|
return
|
|
_skips.add(key)
|
|
is_json = isinstance(req, (dict, list)) and req is not None and method != 'FORM'
|
|
specimens[key] = {
|
|
'request': {'contentType': 'application/json' if is_json else 'application/x-www-form-urlencoded',
|
|
'body': req if is_json else req},
|
|
'response': resp_body,
|
|
}
|
|
|
|
|
|
def api(sess, method, path, form=None, json_body=None, params=None, step='', retry401=True):
|
|
url = BASE + path
|
|
try:
|
|
if json_body is not None:
|
|
r = sess.post(url, json=json_body, params=params, timeout=30) if method == 'POST' \
|
|
else sess.put(url, json=json_body, params=params, timeout=30)
|
|
elif method == 'POST':
|
|
r = sess.post(url, data=form, params=params, timeout=30)
|
|
elif method == 'PUT':
|
|
r = sess.put(url, data=form, params=params, timeout=30)
|
|
elif method == 'DELETE':
|
|
r = sess.delete(url, params=params, timeout=30)
|
|
else:
|
|
r = sess.request(method, url, params=params, timeout=30)
|
|
except Exception as e:
|
|
print(f' ✘ {step}: 网络异常 {e}')
|
|
return None
|
|
if r.status_code == 401 and retry401:
|
|
uid = BUDDY if sess is B else ADMIN
|
|
sess.headers['Authorization'] = f'Bearer {get_token(uid)}'
|
|
return api(sess, method, path, form, json_body, params, step, retry401=False)
|
|
if r.status_code != 200:
|
|
_capture(method, path, json_body if json_body is not None else form,
|
|
{'_http': r.status_code, '_raw': r.text[:400]})
|
|
print(f' ✘ {step}: HTTP {r.status_code} {r.text[:140]}')
|
|
return {'_http': r.status_code, 'code': r.status_code, 'message': r.text[:200]}
|
|
try:
|
|
body = r.json()
|
|
except ValueError:
|
|
_capture(method, path, json_body if json_body is not None else form, {'_raw': r.text[:400]})
|
|
print(f' ✘ {step}: 响应非 JSON')
|
|
return None
|
|
_capture(method, path, json_body if json_body is not None else form, body)
|
|
return body
|
|
|
|
|
|
def api_raw(sess, method, path, **kw):
|
|
"""不经 _capture 的裸请求（二进制/表单文件上传/自定义头检查用）。"""
|
|
return sess.request(method, BASE + path, timeout=60, **kw)
|
|
|
|
|
|
def data_of(resp):
|
|
return resp.get('data') if isinstance(resp, dict) else None
|
|
|
|
|
|
def code_of(resp):
|
|
return resp.get('code') if isinstance(resp, dict) else None
|
|
|
|
|
|
def expect_code(resp, want, flow, case):
|
|
got = code_of(resp)
|
|
if got is not None and int(got) == want:
|
|
check(flow, case, 'pass', f'code={got} 如预期')
|
|
return True
|
|
check(flow, case, 'fail', f'期望 code={want}，实际 code={got} msg={str(resp.get("message") if isinstance(resp, dict) else resp)[:120]}')
|
|
return False
|
|
|
|
|
|
def dict_one(group, parent_id=None):
|
|
try:
|
|
if parent_id:
|
|
rows = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
|
|
"WHERE g.code=%s AND i.deleted=0 AND i.parent_id=%s ORDER BY i.sort_no LIMIT 1",
|
|
(group, parent_id))
|
|
else:
|
|
rows = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
|
|
"WHERE g.code=%s AND i.deleted=0 AND i.parent_id IS NULL ORDER BY i.sort_no LIMIT 1",
|
|
(group,))
|
|
return rows[0]['c'] if rows else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
CTYPE = dict_one('customer_type') or 'customer_type_01'
|
|
GOV = dict_one('industry') or 'gov'
|
|
FOLLOW_WAY = dict_one('follow_way') or 'phone'
|
|
|
|
|
|
# ==================== setup / teardown ====================
|
|
|
|
_g = {'sales_id': SALES_ID, 'buddy_leftover': None, 'transfer_id': None, 'transfer_no': None,
|
|
'sales_assignable': None}
|
|
|
|
|
|
def setup_heavy():
|
|
"""f13 前提：华南销售部 leader=admin；seed 销售甲；BUDDY 划入销售部；甲乙丙(+BUDDY 残留)圈定。"""
|
|
print('== setup_heavy ==')
|
|
# 1) 华南销售部 leader=admin（resolveDirector 数据前提，teardown 还原 NULL）
|
|
dbx("UPDATE sys_dept SET leader_user_id=%s WHERE id=%s", (int(ADMIN), DEPT_SEED))
|
|
check('SETUP', '华南销售部 leader=admin', 'pass')
|
|
# 2) seed 销售甲（幂等：固定 id 先删后插，动态补齐非空列）
|
|
cols = dbq("SHOW COLUMNS FROM crm_auth_user")
|
|
row = {c['Field']: None for c in cols}
|
|
row.update({'id': int(SALES_ID), 'username': SALES_NAME, 'dept_id': DEPT_SEED,
|
|
'enabled': 1, 'employment_status': 'active', 'deleted': 0})
|
|
notnull = [c['Field'] for c in cols if c['Null'] == 'NO' and c['Default'] is None and c['Extra'] == '']
|
|
missing = [c for c in notnull if row.get(c) is None]
|
|
if missing:
|
|
raise RuntimeError(f'crm_auth_user 缺不可空列无法 seed 销售甲: {missing}')
|
|
dbx("DELETE FROM crm_auth_user WHERE id=%s", (SALES_ID,))
|
|
dbx(f"INSERT INTO crm_auth_user ({','.join(row)}) VALUES ({','.join(['%s'] * len(row))})",
|
|
tuple(row.values()))
|
|
check('SETUP', 'seed 销售甲(华南销售部/在职/启用)', 'pass', f'id={SALES_ID}')
|
|
# 3) BUDDY 划入华南销售部（teardown 还原职员部）
|
|
dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (DEPT_SEED, BUDDY))
|
|
# 4) 甲乙丙 assign 给 BUDDY（API 路径，owner→BUDDY 才能由 BUDDY 发起交割）
|
|
for k in ('transfer1', 'transfer2', 'transfer3'):
|
|
r = api(S, 'POST', '/api/customer/assign',
|
|
params={'id': C[k], 'userId': BUDDY}, step=f'assign {k}→BUDDY')
|
|
if code_of(r) != 0:
|
|
raise RuntimeError(f'setup assign {k} 失败: {r}')
|
|
check('SETUP', '甲乙丙 assign→BUDDY', 'pass')
|
|
# 5) BUDDY 名下全量（=交接单圈定范围：甲乙丙 + 残留）
|
|
left = dbq("SELECT id, customer_name FROM customer WHERE owner_user_id=%s AND deleted=0 "
|
|
"AND archive_status=1", (BUDDY,))
|
|
_g['buddy_leftover'] = [{'id': str(r['id']), 'name': r['customer_name']} for r in left]
|
|
print(f" BUDDY 名下 {len(left)} 条：{[x['name'] for x in _g['buddy_leftover']]}")
|
|
|
|
|
|
def teardown_heavy():
|
|
"""还原：客户归属/商机 owner→admin；BUDDY 部门；部门 leader。交接单/销售甲留痕（报告注明）。"""
|
|
print('== teardown_heavy ==')
|
|
n = 0
|
|
for k in ('transfer1', 'transfer2', 'transfer3'):
|
|
n += dbx("UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot='罗伟健', "
|
|
"owner_dept_id=744841292348915712, owner_dept_name_snapshot='广东保伦电子股份有限公司', "
|
|
"enter_pool_time=NULL WHERE id=%s", (ADMIN, C[k]))
|
|
if _g.get('buddy_leftover'):
|
|
for x in _g['buddy_leftover']:
|
|
n += dbx("UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot='罗伟健', "
|
|
"owner_dept_id=744841292348915712, owner_dept_name_snapshot='广东保伦电子股份有限公司', "
|
|
"enter_pool_time=NULL WHERE id=%s", (ADMIN, x['id']))
|
|
dbx("UPDATE opportunity o JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
|
|
"SET o.owner_user_id=%s, o.owner_name_snapshot='罗伟健', o.owner_dept_id=744841292348915712 "
|
|
"WHERE oc.customer_id IN (%s,%s,%s) AND oc.delete_key=0",
|
|
(ADMIN, C['transfer1'], C['transfer2'], C['transfer3']))
|
|
dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (DEPT_BUDDY_ORIG, BUDDY))
|
|
dbx("UPDATE sys_dept SET leader_user_id=NULL WHERE id=%s", (DEPT_SEED,))
|
|
print(f' 还原客户 owner {n} 条；商机 owner 已还原 admin；BUDDY 部门/leader 已还原')
|
|
|
|
|
|
# ==================== F-13 客户交割 ====================
|
|
|
|
def f13():
|
|
print('== F-13 客户交割 ==')
|
|
F = 'F-13'
|
|
total_scope = len(_g['buddy_leftover']) # 交接单圈定 = BUDDY 名下全量（甲乙丙 + 残留）
|
|
|
|
# 13.1 preview：发起前名下可交接客户
|
|
r = api(B, 'GET', '/api/customer/transfer/preview', step='preview')
|
|
rows = data_of(r) or []
|
|
ids_prev = {str(x.get('customerId')) for x in rows}
|
|
ok = len(rows) == total_scope and C['transfer1'] in ids_prev and C['transfer3'] in ids_prev
|
|
check(F, f'preview 圈定 {total_scope} 条(甲乙丙+残留)', 'pass' if ok else 'fail',
|
|
f'实际 {len(rows)} 条')
|
|
|
|
# 13.2 reason 缺失 → 67011
|
|
r = api(B, 'POST', '/api/customer/transfer/initiate', form={}, step='initiate 无 reason')
|
|
expect_code(r, 67011, F, 'initiate reason 缺失 → 67011')
|
|
|
|
# 13.3 toDirectorId 越范围（传 BUDDY 自己，≠leader）→ 67011（D22 范围校验）
|
|
r = api(B, 'POST', '/api/customer/transfer/initiate',
|
|
form={'reason': 'region_adjust', 'toDirectorId': BUDDY}, step='initiate 越范围总监')
|
|
expect_code(r, 67011, F, 'toDirectorId 不在可交接范围 → 67011')
|
|
|
|
# 13.4 resign 自动推导唯一接收总监（D22）→ JG 编号 + PENDING
|
|
r = api(B, 'POST', '/api/customer/transfer/initiate', form={'reason': 'resign', 'remark': 'e2e 票05'},
|
|
step='initiate resign 自动推导')
|
|
d = data_of(r) or {}
|
|
tno = str(d.get('transferNo') or '')
|
|
today = datetime.date.today().strftime('%Y%m%d')
|
|
ok = code_of(r) == 0 and tno.startswith('JG' + today) and len(tno) == 14
|
|
_g['transfer_id'] = d.get('id')
|
|
_g['transfer_no'] = tno
|
|
check(F, f'initiate resign → 交接单 {tno}（D22 自动推导罗伟健）',
|
|
'pass' if ok else 'fail', f'status={d.get("status")} totalCount={d.get("totalCount")}')
|
|
check(F, 'totalCount=圈定全量（无 customerIds 字段=名下全量口径）',
|
|
'pass' if int(d.get('totalCount') or 0) == total_scope else 'warn',
|
|
f'totalCount={d.get("totalCount")} vs 预览 {total_scope}')
|
|
tid = _g['transfer_id']
|
|
|
|
# 13.5 重入 initiate → 67010
|
|
r = api(B, 'POST', '/api/customer/transfer/initiate', form={'reason': 'resign'}, step='重入')
|
|
expect_code(r, 67010, F, '存在未完成交接单重入 → 67010')
|
|
|
|
# 13.6 DB：整批原子——名下客户 owner→总监（admin）四件套
|
|
row = dbq("SELECT owner_user_id, owner_user_name_snapshot, owner_dept_id, enter_pool_time "
|
|
"FROM customer WHERE id=%s", (C['transfer1'],))
|
|
ok = row and str(row[0]['owner_user_id']) == ADMIN and row[0]['enter_pool_time'] is None
|
|
check(F, 'initiate 后客户 owner→接收总监(整批原子)', 'pass' if ok else 'fail',
|
|
f"owner={row[0]['owner_user_id'] if row else '?'}")
|
|
|
|
# 13.7 DB：商机 owner 三列联动（真源同步）
|
|
orow = dbq("SELECT o.owner_user_id, o.owner_name_snapshot, o.owner_dept_id FROM opportunity o "
|
|
"JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
|
|
"WHERE oc.customer_id=%s AND oc.delete_key=0", (C['transfer1'],))
|
|
ok = orow and str(orow[0]['owner_user_id']) == ADMIN and orow[0]['owner_name_snapshot'] == '罗伟健'
|
|
check(F, '商机 owner 三列联动(owner/快照/部门)', 'pass' if ok else 'fail',
|
|
f"e2c-联动-交割 owner={orow[0]['owner_user_id'] if orow else '?'}")
|
|
|
|
# 13.8 DB：oplog 系统叙事含交接编号
|
|
lrow = dbq("SELECT detail FROM customer_oplog WHERE customer_id=%s AND action='TRANSFER' "
|
|
"ORDER BY id DESC LIMIT 1", (C['transfer1'],))
|
|
ok = lrow and tno in (lrow[0]['detail'] or '') and '发起交接' in (lrow[0]['detail'] or '')
|
|
check(F, 'oplog 含交接编号+发起交接叙事', 'pass' if ok else 'fail',
|
|
(lrow[0]['detail'] or '')[:60] if lrow else '无记录')
|
|
|
|
# 13.9 detail 明细
|
|
r = api(B, 'GET', f'/api/customer/transfer/{tid}', step='detail')
|
|
d = data_of(r) or {}
|
|
det = d.get('details') or []
|
|
ok = len(det) == total_scope and all(int(x.get('assignStatus') or 0) == 0 for x in det)
|
|
check(F, 'detail 明细齐且全待分配', 'pass' if ok else 'fail', f'minutes={len(det)}')
|
|
detail_ids = [str(x.get('customerId')) for x in det]
|
|
|
|
# 13.10 assignable（D-07：根部门全树查询稳定超时 → 缺陷验证模式，双次重试留证据）
|
|
users, uids = [], set()
|
|
for _ in range(2):
|
|
r = api(B, 'GET', f'/api/customer/transfer/{tid}/assignable', step='assignable')
|
|
users = data_of(r) or []
|
|
uids = {str(x.get('userId')) for x in users}
|
|
if uids:
|
|
break
|
|
time.sleep(2)
|
|
_g['sales_assignable'] = bool(uids)
|
|
if uids:
|
|
ok = SALES_ID in uids and BUDDY not in uids and ADMIN not in uids
|
|
check(F, 'assignable 含销售甲且排除发起人/总监', 'pass' if ok else 'fail',
|
|
f'共 {len(users)} 人')
|
|
else:
|
|
check(F, 'assignable 含销售甲且排除发起人/总监', 'warn',
|
|
'接口 30s×2 稳定超时未返回 → 登记 D-07')
|
|
defect('D-07', 'P1', '交割可分配销售列表接口超时',
|
|
'复现：发起交割后 GET /api/customer/transfer/{id}/assignable → '
|
|
'30s（E2E）与 60s（手动 probe）均稳定超时，两轮各复现 2 次。'
|
|
'前端「选择分配对象」无法加载。疑因 getChildDeptIds 根部门全树递归 + '
|
|
'用户过滤全表扫描；assign 本体不受影响（直接指定 userId 成功）。待复核。')
|
|
|
|
# 13.11a assign 单条（甲→销售甲，逐条独立事务）
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
|
|
step='assign 甲→销售甲')
|
|
br = data_of(r) or {}
|
|
check(F, 'assign 单条 甲→销售甲', 'pass' if int(br.get('successCount') or 0) == 1 else 'fail',
|
|
f'fails={br.get("failures")}')
|
|
|
|
# 13.11b 重复 assign 已分配明细 → 67012（失败行）
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
|
|
step='重复 assign 甲')
|
|
br = data_of(r) or {}
|
|
fails = br.get('failures') or []
|
|
ok = int(br.get('successCount') or 0) == 0 and fails
|
|
check(F, '重复分配已分配明细 → 67012(失败行)', 'pass' if ok else 'fail',
|
|
f'{fails[0] if fails else br}')
|
|
|
|
# 13.11c 分配给当前负责人 → 67012（乙明细仍 PENDING，临时把乙 owner 指向 BUDDY）
|
|
dbx("UPDATE customer SET owner_user_id=%s WHERE id=%s", (BUDDY, C['transfer2']))
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', C['transfer2']), ('assignUserId', BUDDY)],
|
|
step='分配给当前负责人')
|
|
br = data_of(r) or {}
|
|
fails = br.get('failures') or []
|
|
ok = int(br.get('successCount') or 0) == 0 and fails
|
|
check(F, '分配给当前负责人 → 67012(失败行)', 'pass' if ok else 'warn',
|
|
f'{fails[0] if fails else br}')
|
|
dbx("UPDATE customer SET owner_user_id=%s WHERE id=%s", (ADMIN, C['transfer2']))
|
|
|
|
# 13.11d 无效分配对象 → 67012（spec 写 66001，实现统一收口 67012 = 契约发现）
|
|
bad = 999999999999
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', C['transfer2']), ('assignUserId', bad)], step='分配无效用户')
|
|
got = code_of(r)
|
|
br = data_of(r) or {}
|
|
fails = br.get('failures') or []
|
|
ok = (got == 67012) or (got == 0 and fails and int(fails[0].get('code') or fails[0].get('errorCode') or 0) == 67012)
|
|
check(F, '分配对象不存在 → 67012(spec 66001 偏差记录)', 'pass' if ok else 'warn',
|
|
f'code={got} fails={fails[:1]}')
|
|
|
|
# 13.11e 剩余明细批量 assign → 全部完成
|
|
rest = [i for i in detail_ids if i != str(C['transfer1'])]
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', i) for i in rest] + [('assignUserId', SALES_ID)],
|
|
step='assign 剩余→销售甲')
|
|
br = data_of(r) or {}
|
|
check(F, f'批量 assign 剩余成功 {br.get("successCount")}/{len(rest)}',
|
|
'pass' if int(br.get('successCount') or 0) == len(rest) else 'fail',
|
|
f'fails={br.get("failures")}')
|
|
|
|
# 13.12 D23 自动完成
|
|
r = api(B, 'GET', f'/api/customer/transfer/{tid}', step='detail after assign')
|
|
d = data_of(r) or {}
|
|
ok = int(d.get('assignedCount') or 0) == int(d.get('totalCount') or -1) and int(d.get('status') or -1) == 1
|
|
check(F, 'assignedCount==totalCount → 自动置已完成(D23)', 'pass' if ok else 'fail',
|
|
f"assigned={d.get('assignedCount')}/{d.get('totalCount')} status={d.get('status')}")
|
|
|
|
# 13.13 已完成单再 assign → 状态校验（实现在 BatchResult 失败行内收口，非抛 67004）
|
|
r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
|
|
form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
|
|
step='已完成单再分配')
|
|
got = code_of(r)
|
|
br = data_of(r) or {}
|
|
fails = br.get('failures') or []
|
|
ok = (got == 67004) or (got == 0 and fails and '不可再分配' in str(fails[0].get('message', '')))
|
|
check(F, '交接单已完成不可再分配（67004 或失败行收口）', 'pass' if ok else 'fail',
|
|
f'code={got} fails={fails[:1]}')
|
|
|
|
# 13.16 DB：分配后客户 owner=销售甲 + 商机二次联动
|
|
row = dbq("SELECT owner_user_id, owner_user_name_snapshot FROM customer WHERE id=%s", (C['transfer1'],))
|
|
orow = dbq("SELECT o.owner_user_id, o.owner_name_snapshot FROM opportunity o "
|
|
"JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
|
|
"WHERE oc.customer_id=%s AND oc.delete_key=0", (C['transfer1'],))
|
|
ok = row and str(row[0]['owner_user_id']) == SALES_ID and orow and str(orow[0]['owner_user_id']) == SALES_ID
|
|
check(F, 'assign 后客户/商机 owner→销售甲', 'pass' if ok else 'fail',
|
|
f"cust={row[0]['owner_user_id'] if row else '?'} opp={orow[0]['owner_user_id'] if orow else '?'}")
|
|
|
|
# 13.17 page 按 transferNo 过滤
|
|
r = api(B, 'GET', '/api/customer/transfer/page', params={'transferNo': tno, 'pageSize': 10},
|
|
step='page 交接编号过滤')
|
|
recs = (data_of(r) or {}).get('content') or []
|
|
ok = any(str(x.get('transferNo')) == tno for x in recs)
|
|
check(F, 'page 交接编号过滤命中', 'pass' if ok else 'fail', f'命中 {len(recs)} 条')
|
|
|
|
# 13.18 交接编号当日续号（DB 断言当日 max 即本单）
|
|
mx = dbq("SELECT MAX(transfer_no) m FROM customer_transfer WHERE transfer_no LIKE %s",
|
|
('JG' + today + '%',))
|
|
check(F, '交接编号当日续号不重号', 'pass' if mx and str(mx[0]['m']) == tno else 'fail',
|
|
f"max={mx[0]['m'] if mx else '?'} vs 本单 {tno}")
|
|
|
|
|
|
# ==================== F-14 客户导入 ====================
|
|
|
|
def _mk_xlsx(path, cust_rows, contact_rows=None, header_override=None):
|
|
"""openpyxl 造 EasyExcel 兼容 xlsx：sheet0=客户(9 列 index 绑定) sheet1=联系人(5 列)。"""
|
|
from openpyxl import Workbook
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = '客户'
|
|
headers = header_override or ['客户编号', '客户名称', '客户类型', '统一社会信用代码',
|
|
'省份编码', '城市编码', '行业编码', '客户星级(1-5)', '备注']
|
|
ws.append(headers)
|
|
for row in cust_rows:
|
|
ws.append([(row[i] if i < len(row) else None) for i in range(9)])
|
|
ws2 = wb.create_sheet('联系人')
|
|
ws2.append(['客户编号', '联系人姓名', '联系电话', '职务名称', '是否关键联系人(是/否)'])
|
|
for row in (contact_rows or []):
|
|
ws2.append([(row[i] if i < len(row) else None) for i in range(5)])
|
|
wb.save(path)
|
|
|
|
|
|
def _upload(sess, path, import_mode, step):
|
|
with open(path, 'rb') as f:
|
|
r = api_raw(sess, 'POST', '/api/customer/import/upload',
|
|
files={'file': (os.path.basename(path), f,
|
|
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')},
|
|
data={'importMode': importMode_fix(import_mode)})
|
|
if r.status_code != 200:
|
|
_capture('POST', '/api/customer/import/upload', {'importMode': import_mode},
|
|
{'_http': r.status_code, '_raw': r.text[:400]})
|
|
print(f' ✘ {step}: HTTP {r.status_code} {r.text[:140]}')
|
|
return {'_http': r.status_code, 'code': r.status_code, 'message': r.text[:200]}
|
|
body = r.json()
|
|
_capture('POST', '/api/customer/import/upload', {'importMode': import_mode}, body)
|
|
return body
|
|
|
|
|
|
def importMode_fix(m):
|
|
return m
|
|
|
|
|
|
def f14():
|
|
print('== F-14 客户导入 ==')
|
|
F = 'F-14'
|
|
XLSX = '.scratch/customer-e2e/_import-heavy.xlsx'
|
|
new_name = f'{PFX}导入-新增-{TS}'
|
|
|
|
# 14.1 模板下载：RFC 5987 + xlsx 魔数
|
|
r = api_raw(S, 'GET', '/api/customer/import/template')
|
|
cd = r.headers.get('Content-Disposition', '')
|
|
ok = (r.status_code == 200 and r.content[:2] == b'PK'
|
|
and ("filename*=UTF-8''" in cd or 'filename*=' in cd))
|
|
check(F, '模板下载(RFC5987 文件名+xlsx 魔数)', 'pass' if ok else 'fail', f'CD={cd[:80]}')
|
|
|
|
ind = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
|
|
"WHERE g.code='industry' AND i.deleted=0 AND i.parent_id IS NOT NULL "
|
|
"ORDER BY i.sort_no LIMIT 1")
|
|
ind_code = ind[0]['c'] if ind else ''
|
|
# 14.2 构造 7 行客户 + 3 行联系人
|
|
# r1 带行内编号（联系人 sheet 需按编号挂接）；r2/r3/r4 必须带名称（名称必填先于匹配/重复组判定）；
|
|
# r5 备注列非空防 EasyExcel 忽略全空行
|
|
new_no = f'E2C-IMP-{TS}'
|
|
cust = [
|
|
[new_no, new_name, CTYPE, None, '440000', '440100', ind_code, '3', '票05 新增行'], # r1 INSERT
|
|
[NO['transfer2'], PFX + '交割-乙', None, None, None, None, None, '4', None], # r2 按编号 UPDATE 乙
|
|
[NO['transfer3'], PFX + '交割-丙', None, None, None, None, None, None, None], # r3 重复组A
|
|
[NO['transfer3'], PFX + '交割-丙', None, None, None, None, None, None, None], # r4 重复组B（D24）
|
|
[None, None, None, None, None, None, None, None, '缺名称行'], # r5 缺名称
|
|
[None, f'{PFX}导入-星级-{TS}', CTYPE, None, None, None, None, '6', None], # r6 星级越界
|
|
[None, f"{dbq('SELECT customer_name n FROM customer WHERE id=%s', (IDS['customers']['simA']['id'],))[0]['n']}",
|
|
None, None, None, None, None, None, None], # r7 与 simA 同名 → SUSPECT
|
|
]
|
|
contacts = [
|
|
[new_no, 'e2c-联系人甲', '13800007701', '项目经理', '是'], # 挂 r1（文件内新增客户）
|
|
[None, 'e2c-联系人乙', None, None, None], # 缺编号
|
|
['E2C-NOT-EXIST', 'e2c-联系人丙', '13800007702', '组长', None], # 编号不存在
|
|
]
|
|
_mk_xlsx(XLSX, cust, contacts)
|
|
|
|
# 14.3 非法 importMode → 67013
|
|
r = _upload(S, XLSX, 'BAD_MODE', '非法 mode')
|
|
expect_code(r, 67013, F, 'importMode 非法 → 67013')
|
|
|
|
# 14.4 表头错列 → 67013（模板版本锚点逐列比对）
|
|
_mk_xlsx(XLSX, cust[:1], header_override=['客户编号', '客户名称X', '客户类型', '统一社会信用代码',
|
|
'省份编码', '城市编码', '行业编码', '客户星级(1-5)', '备注'])
|
|
r = _upload(S, XLSX, 'APPEND_ONLY', '表头错列')
|
|
expect_code(r, 67013, F, '模板表头不符 → 67013')
|
|
|
|
# 14.5 空数据行 → 67013
|
|
_mk_xlsx(XLSX, [])
|
|
r = _upload(S, XLSX, 'APPEND_ONLY', '空数据')
|
|
expect_code(r, 67013, F, '文件内无数据行 → 67013')
|
|
|
|
# 14.6 合法文件 APPEND_ONLY 上传 → DRAFT 预检计数
|
|
_mk_xlsx(XLSX, cust, contacts)
|
|
r = _upload(S, XLSX, 'APPEND_ONLY', 'APPEND_ONLY 上传')
|
|
d = data_of(r) or {}
|
|
task_id = d.get('taskId')
|
|
ok = (code_of(r) == 0 and str(d.get('totalCount')) == '7'
|
|
and str(d.get('insertCount')) == '1' and str(d.get('updateCount')) == '0'
|
|
and str(d.get('failCount')) == '5' and str(d.get('suspectCount')) == '1')
|
|
# APPEND_ONLY：r2 已存在 → fail「客户已存在」；insert=r1；suspect=r7
|
|
check(F, 'APPEND_ONLY 预检：t=7 insert=1 fail=5(含已存在) suspect=1',
|
|
'pass' if ok else 'fail', f"t={d.get('totalCount')} i={d.get('insertCount')} "
|
|
f"u={d.get('updateCount')} f={d.get('failCount')} s={d.get('suspectCount')}")
|
|
|
|
# 14.7 行级 verdict 语义抽查（预览 rows）
|
|
pmap = {(x.get('sheetName'), str(x.get('rowNum'))): x for x in (d.get('rows') or [])
|
|
if x.get('sheetName') != '联系人'}
|
|
v2 = pmap.get(('客户', '2'), {}).get('verdict')
|
|
v3 = pmap.get(('客户', '3'), {}).get('reason', '')
|
|
v7 = pmap.get(('客户', '7'), {})
|
|
ok = ('FAIL' in str(v2) or 'fail' in str(v2)) and '重复时不导入' in v3 \
|
|
and 'SUSPECT' in str(v7.get('verdict', ''))
|
|
check(F, '行级判定：已存在 FAIL / 文件内重复组 FAIL(D24) / 相似 SUSPECT',
|
|
'pass' if ok else 'fail', f"r2={v2} r3={v3[:24]} r7={v7.get('verdict')}")
|
|
|
|
# 14.8 UPDATE_ONLY：r1 新客户行 → fail「未找到可更新客户」；r2 按编号 UPDATE
|
|
r = _upload(S, XLSX, 'UPDATE_ONLY', 'UPDATE_ONLY')
|
|
d2 = data_of(r) or {}
|
|
ok = code_of(r) == 0 and str(d2.get('updateCount')) == '1' and str(d2.get('insertCount')) == '0'
|
|
check(F, 'UPDATE_ONLY 预检：update=1(按编号) insert=0', 'pass' if ok else 'fail',
|
|
f"u={d2.get('updateCount')} i={d2.get('insertCount')}")
|
|
|
|
# 14.9 UPSERT 独立文件：新编号 + 已有客户同名 → 不按名称匹配（SUSPECT 而非 UPDATE）
|
|
up_no = f'E2C-UP-{TS}'
|
|
_mk_xlsx(XLSX, [[up_no, f"{dbq('SELECT customer_name n FROM customer WHERE id=%s', (IDS['customers']['simA']['id'],))[0]['n']}",
|
|
None, None, None, None, None, None, None]])
|
|
r = _upload(S, XLSX, 'UPSERT', 'UPSERT 同名新编号')
|
|
d3 = data_of(r) or {}
|
|
pmap3 = {(x.get('sheetName'), str(x.get('rowNum'))): x for x in (d3.get('rows') or [])
|
|
if x.get('sheetName') != '联系人'}
|
|
v1u = pmap3.get(('客户', '1'), {})
|
|
ok = ('SUSPECT' in str(v1u.get('verdict', ''))
|
|
and '疑似重复' in str(v1u.get('reason', '')))
|
|
check(F, 'UPSERT 同名(新编号)→SUSPECT 不按名称匹配', 'pass' if ok else 'warn',
|
|
f"r1={v1u.get('verdict')} {str(v1u.get('reason'))[:30]}")
|
|
|
|
# 14.10 confirm 提交 + 状态机 67014
|
|
r = api(S, 'POST', f'/api/customer/import/{task_id}/confirm', step='confirm')
|
|
ok = code_of(r) == 0
|
|
check(F, 'confirm DRAFT→RUNNING 异步执行', 'pass' if ok else 'fail', f'taskId={task_id}')
|
|
r = api(S, 'POST', f'/api/customer/import/{task_id}/confirm', step='重复 confirm')
|
|
expect_code(r, 67014, F, '重复 confirm → 67014')
|
|
|
|
# 14.11 轮询 result → 终态（D-05 阻断下预期 FAILED；修复后应为 DONE 行级部分成功）
|
|
status, result = None, {}
|
|
for _ in range(30):
|
|
time.sleep(1)
|
|
r = api(S, 'GET', f'/api/customer/import/{task_id}', step='poll result')
|
|
result = data_of(r) or {}
|
|
status = result.get('status')
|
|
if str(status) in ('2', '3'):
|
|
break
|
|
if str(status) == '3':
|
|
check(F, '执行终态', 'warn',
|
|
f"status=FAILED fail_reason={result.get('failReason')}（D-05 导入 INSERT 缺省列阻断）")
|
|
defect('D-05', 'P1',
|
|
'导入新增客户行必失败：customer.is_biz_negotiated NOT NULL 无默认值，createCustomer 未赋值',
|
|
'复现：APPEND_ONLY/UPSERT 上传含新增行文件 → confirm → 单行 INSERT 报 '
|
|
"Field 'is_biz_negotiated' doesn't have a default value（日志 23:37:05 task-4）→ "
|
|
'catch 后 insertFail 把完整 MyBatis 异常文本写入 customer_import_fail.fail_reason '
|
|
'varchar(500) → Data too long → 整任务 FAILED、行级明细全丢（fail 表 0 行）。'
|
|
'期望：实体默认值或 INSERT 前补 0；insertFail reason 截断防放大。')
|
|
else:
|
|
ok = str(status) == '2' and str(result.get('insertCount')) == '1' \
|
|
and str(result.get('updateCount')) == '1'
|
|
check(F, '执行完成 DONE：insert=1 update=1', 'pass' if ok else 'fail',
|
|
f"status={status} i={result.get('insertCount')} u={result.get('updateCount')} "
|
|
f"f={result.get('failCount')} s={result.get('suspectCount')}")
|
|
|
|
# 14.12 failures 明细（D-05 阻断下 fail 表 0 行 = 明细丢失实证）
|
|
r = api(S, 'GET', f'/api/customer/import/{task_id}/failures', step='failures')
|
|
fails = data_of(r) or []
|
|
check(F, f'失败明细 {len(fails)} 行', 'warn' if str(status) == '3' and not fails else 'pass',
|
|
'；'.join(str(x.get('reason'))[:20] for x in fails[:3]) or 'D-05 阻断：明细丢失')
|
|
|
|
# 14.13 DB：任务态落库证据（r1 未落库 = D-05 实证；r2 未更新 = 阻断实证）
|
|
row = dbq("SELECT id FROM customer WHERE customer_no=%s AND deleted=0", (new_no,))
|
|
row2 = dbq("SELECT customer_star_level FROM customer WHERE id=%s", (C['transfer2'],))
|
|
check(F, 'r1 新客户未落库（D-05）/r2 未更新（阻断）', 'warn',
|
|
f"r1={'未落库' if not row else '落库'} r2星={row2[0]['customer_star_level'] if row2 else '?'}")
|
|
if row:
|
|
_g['import_new_id'] = str(row[0]['id'])
|
|
|
|
# 14.14 page 仅当前用户可见（PageResult 序列化字段=content，非 records）
|
|
r = api(S, 'GET', '/api/customer/import/page', params={'pageSize': 50}, step='page admin')
|
|
mine = {str(x.get('taskId')) for x in (data_of(r) or {}).get('content') or []}
|
|
r = api(B, 'GET', '/api/customer/import/page', params={'pageSize': 50}, step='page buddy')
|
|
theirs = {str(x.get('taskId')) for x in (data_of(r) or {}).get('content') or []}
|
|
if str(task_id) not in mine:
|
|
check(F, '最近导入仅提交人可见(creator 过滤)', 'warn',
|
|
f'admin page 未含本 task（buddy 有={str(task_id) in theirs}）——需复核 creator_id 过滤')
|
|
defect('D-06', 'P1',
|
|
'导入任务分页查不到本人任务',
|
|
'复现：任意导入任务（DB creator_id 正确=' + str(ADMIN) + '）→ GET '
|
|
'/api/customer/import/page 以提交人身份查询 → content 不含该任务。'
|
|
'前端「最近导入记录」列表将恒为空。待复核 eq(creatorId, getRequiredUserId) 条件。')
|
|
else:
|
|
ok = str(task_id) not in theirs
|
|
check(F, '最近导入仅提交人可见(creator 过滤)', 'pass' if ok else 'fail',
|
|
f'admin 有={str(task_id) in mine} buddy 有={str(task_id) in theirs}')
|
|
|
|
|
|
def f14_cleanup():
|
|
print('== f14_cleanup ==')
|
|
nid = _g.get('import_new_id')
|
|
if nid:
|
|
dbx("DELETE FROM customer_oplog WHERE customer_id=%s", (nid,))
|
|
dbx("DELETE FROM customer_contact WHERE customer_id=%s", (nid,))
|
|
dbx("DELETE FROM customer WHERE id=%s", (nid,))
|
|
print(f' 已清导入新客户 {nid}')
|
|
dbx("UPDATE customer SET customer_name=%s, customer_star_level=3 WHERE id=%s",
|
|
(PFX + '交割-乙', C['transfer2']))
|
|
print(' 乙名称/星级已还原')
|
|
|
|
|
|
# ==================== 服务重启（f15 Job 观测用，U-02 拍板：env 注入 cron 实测） ====================
|
|
|
|
JAR = r'e:\code\crm-backend-matt\crm-app\target\crm-app-1.0.0-SNAPSHOT.jar'
|
|
|
|
|
|
def _pid_on_8080():
|
|
out = subprocess.run('netstat -ano | findstr :8080 | findstr LISTENING',
|
|
shell=True, capture_output=True, text=True).stdout
|
|
for line in out.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 5 and parts[3] == 'LISTENING':
|
|
return int(parts[4])
|
|
return None
|
|
|
|
|
|
def restart_service(extra_env=None, wait=300):
|
|
pid = _pid_on_8080()
|
|
if pid:
|
|
subprocess.run(f'taskkill /F /PID {pid}', shell=True, capture_output=True)
|
|
for _ in range(15): # 等 LISTENING 消失（端口释放）再起，避免绑定失败
|
|
time.sleep(1)
|
|
if not _pid_on_8080():
|
|
break
|
|
env = dict(os.environ)
|
|
env.update({'SPRING_PROFILES_ACTIVE': 'verify', 'CRM_MINIO_AK': 'admin',
|
|
'CRM_MINIO_SK': 'Itc@123456'})
|
|
env.update(extra_env or {})
|
|
print(f' -- 起服 {os.path.basename(JAVA_EXE)} extra_env={extra_env or "默认"} --', flush=True)
|
|
subprocess.Popen([JAVA_EXE, '-jar', JAR], env=env, cwd=r'e:\code\crm-backend-matt',
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS)
|
|
t0 = time.time()
|
|
while time.time() - t0 < wait:
|
|
time.sleep(5)
|
|
try:
|
|
r = requests.post(f'{BASE}/api/customer/workspace/pool/page',
|
|
data={'current': 1, 'size': 1}, timeout=5,
|
|
headers={'Authorization': S.headers['Authorization']})
|
|
if r.status_code == 200:
|
|
print(f' 服务已起（{int(time.time() - t0)}s）', flush=True)
|
|
return True
|
|
if r.status_code == 401:
|
|
S.headers['Authorization'] = f'Bearer {get_token(ADMIN)}'
|
|
except Exception:
|
|
print('.', end='', flush=True)
|
|
raise RuntimeError('服务重启后未就绪')
|
|
|
|
|
|
def f15():
|
|
print('== F-15 提醒规则族 ==')
|
|
F = 'F-15'
|
|
URL = '/api/rule/customer/reminder'
|
|
RULE_ORIG = {'masterEnabled': 1, 'firstTriggerEnabled': 1, 'firstTriggerDays': 30,
|
|
'secondIntervalEnabled': 1, 'secondIntervalDays': 7}
|
|
|
|
# 15.1 GET 单例回显
|
|
r = api(S, 'GET', URL, step='current')
|
|
d = data_of(r) or {}
|
|
ok = code_of(r) == 0 and d.get('masterEnabled') is not None
|
|
check(F, 'GET 单例可读（记录当前值，不预设——重跑时规则可能留测试态）', 'pass' if ok else 'fail',
|
|
f"master={d.get('masterEnabled')} first={d.get('firstTriggerEnabled')}/{d.get('firstTriggerDays')} "
|
|
f"second={d.get('secondIntervalEnabled')}/{d.get('secondIntervalDays')}")
|
|
|
|
# 15.2 PUT 全量覆盖（测试态：first=1/second=1 全开）
|
|
r = api(S, 'PUT', URL, json_body={'masterEnabled': 1, 'firstTriggerEnabled': 1,
|
|
'firstTriggerDays': 1, 'secondIntervalEnabled': 1,
|
|
'secondIntervalDays': 1}, step='PUT 测试态')
|
|
d = data_of(r) or {}
|
|
ok = code_of(r) == 0 and str(d.get('firstTriggerDays')) == '1'
|
|
check(F, 'PUT 全量覆盖保存即生效', 'pass' if ok else 'fail', f"first={d.get('firstTriggerDays')}")
|
|
|
|
# 15.3/15.4 64023：天数非正整数（校验失败不落库）
|
|
for case, days2 in (('firstTriggerDays=0 → 64023', 0), ('secondIntervalDays=-1 → 64023', -1)):
|
|
body = {'masterEnabled': 1, 'firstTriggerEnabled': 1, 'firstTriggerDays': 1,
|
|
'secondIntervalEnabled': 1, 'secondIntervalDays': 1}
|
|
if 'first' in case:
|
|
body['firstTriggerDays'] = days2
|
|
else:
|
|
body['secondIntervalDays'] = days2
|
|
r = api(S, 'PUT', URL, json_body=body, step=case)
|
|
expect_code(r, 64023, F, case)
|
|
d = data_of(api(S, 'GET', URL, step='复查')) or {}
|
|
ok = str(d.get('firstTriggerDays')) == '1' and str(d.get('secondIntervalDays')) == '1'
|
|
check(F, '64023 校验失败未落库（1/1 保持）', 'pass' if ok else 'fail',
|
|
f"first={d.get('firstTriggerDays')} second={d.get('secondIntervalDays')}")
|
|
|
|
# 15.5 开关缺省=0（orDisabled：全量覆盖不静默保留旧值）
|
|
r = api(S, 'PUT', URL, json_body={'firstTriggerDays': 2, 'secondIntervalDays': 2}, step='PUT 缺开关')
|
|
d = data_of(r) or {}
|
|
ok = code_of(r) == 0 and str(d.get('masterEnabled')) == '0' \
|
|
and str(d.get('firstTriggerEnabled')) == '0' and str(d.get('secondIntervalEnabled')) == '0'
|
|
check(F, '开关缺省 → 0（缺省禁用口径）', 'pass' if ok else 'fail',
|
|
f"master={d.get('masterEnabled')} first={d.get('firstTriggerEnabled')} "
|
|
f"second={d.get('secondIntervalEnabled')}")
|
|
|
|
# 15.6 恢复测试态（Job 观测前提）
|
|
api(S, 'PUT', URL, json_body={'masterEnabled': 1, 'firstTriggerEnabled': 1,
|
|
'firstTriggerDays': 1, 'secondIntervalEnabled': 1,
|
|
'secondIntervalDays': 1}, step='恢复测试态')
|
|
|
|
# 15.7 Job 观测（U-02 拍板：重启注入 cron 每 30s 实测；对称商机 U03 仅 SKIP 的升级做法）
|
|
print(' -- 重启服务带 CRM_CUSTOMER_REMINDERCRON=*/30 * * * * ? --')
|
|
restart_service({'CRM_CUSTOMER_REMINDERCRON': '*/30 * * * * ?'})
|
|
time.sleep(70) # 等至少两轮 Job
|
|
old_anchor = dbq("SELECT last_valid_follow_time a FROM customer WHERE id=%s", (C['followOld'],))[0]['a']
|
|
plan_expect = (old_anchor + datetime.timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')
|
|
notices = dbq("SELECT * FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
|
|
n = notices[0] if notices else {}
|
|
ok = (len(notices) == 1 and str(n.get('notice_status')) == '0'
|
|
and 'FOLLOW' in str(n.get('notice_type'))
|
|
and str(n.get('target_user_id')) == ADMIN
|
|
and str(n.get('plan_notify_time')).startswith(plan_expect[:10]))
|
|
check(F, f'Job 落待发：超期样例 1 条 plan={plan_expect[:10]}（锚点+first=1）',
|
|
'pass' if ok else 'fail',
|
|
f"status={n.get('notice_status')} type={n.get('notice_type')} "
|
|
f"target={n.get('target_user_id')} plan={n.get('plan_notify_time')}")
|
|
payload = json.loads(n.get('payload_json') or '{}') if n else {}
|
|
ok = payload.get('remindSeq') == 0 and 'anchor' in payload and payload.get('firstTriggerDays') == 1
|
|
check(F, 'payload 快照（remindSeq/anchor/规则参数）', 'pass' if ok else 'warn',
|
|
str(payload)[:100])
|
|
|
|
# 15.8 未到期不落 + 公海/归档不参与
|
|
f2 = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['followFuture'],))
|
|
pool_ids = ','.join(str(C[k]) for k in ('pool1', 'pool2') if k in C)
|
|
arc = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['archived'],))
|
|
ok = f2[0]['c'] == 0 and arc[0]['c'] == 0
|
|
check(F, '未到期/公海/归档客户均未落待发', 'pass' if ok else 'fail',
|
|
f"待办={f2[0]['c']} 公海+归档={arc[0]['c']}")
|
|
|
|
# 15.9 幂等：第二轮 Job 同 plan 不重写
|
|
time.sleep(70)
|
|
n2 = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
|
|
check(F, '幂等：同计划时刻任意状态占位不重写', 'pass' if n2[0]['c'] == 1 else 'fail',
|
|
f'条数 {n2[0]["c"]}')
|
|
|
|
# 15.10 锚点刷新 → 旧锚点未发提醒置失效
|
|
r = api(S, 'POST', f"/api/customer/{C['followOld']}/follow",
|
|
form={'followWay': FOLLOW_WAY, 'followContent': f'票05 锚点刷新 {TS}'},
|
|
step='follow 超期客户')
|
|
if code_of(r) == 0:
|
|
time.sleep(70)
|
|
rows = dbq("SELECT notice_status FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
|
|
invalidated = all(str(r0['notice_status']) != '0' for r0 in rows)
|
|
check(F, '锚点被跟进刷新 → 旧未发提醒置已失效', 'pass' if invalidated else 'fail',
|
|
f'status={[str(r0["notice_status"]) for r0 in rows]}')
|
|
else:
|
|
check(F, '锚点被跟进刷新 → 旧未发提醒置已失效', 'fail', f'follow 失败 {r}')
|
|
|
|
# 15.11 还原：规则回原值 + 服务还原默认 cron
|
|
api(S, 'PUT', URL, json_body=RULE_ORIG, step='规则还原')
|
|
d = data_of(api(S, 'GET', URL, step='规则复查')) or {}
|
|
ok = str(d.get('firstTriggerDays')) == '30' and str(d.get('secondIntervalDays')) == '7'
|
|
check(F, '提醒规则还原出厂值（30/7）', 'pass' if ok else 'fail', f'{d}')
|
|
print(' -- 还原服务（默认 cron 凌晨 4 点） --')
|
|
restart_service()
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/pool/page',
|
|
form={'current': 1, 'size': 1}, step='smoke')) or {}
|
|
check(F, '服务还原后 smoke', 'pass' if d else 'fail')
|
|
|
|
|
|
def f16():
|
|
print('== F-16 查重设置消费侧（出厂默认：单例未落地，DefaultCustomerDedupSettingProvider 兜底） ==')
|
|
F = 'F-16'
|
|
simA_name = dbq("SELECT customer_name n FROM customer WHERE id=%s",
|
|
(IDS['customers']['simA']['id'],))[0]['n']
|
|
simA_code = dbq("SELECT unified_credit_code c FROM customer WHERE id=%s",
|
|
(IDS['customers']['simA']['id'],))[0]['c']
|
|
|
|
# 16.1 信用代码硬拦（出厂 creditCodeEnabled=true：开关关也拦——单例缺省无法验关闭态）
|
|
r = api(S, 'GET', '/api/customer/check-credit-code', params={'creditCode': simA_code}, step='check-credit')
|
|
hit = data_of(r)
|
|
check(F, 'check-credit-code 命中已有信用代码', 'pass' if hit else 'fail', str(hit)[:80])
|
|
form = {'customerName': f'{PFX}查重-撞码-{TS}', 'customerType': CTYPE,
|
|
'unifiedCreditCode': simA_code, 'provinceCode': '440000',
|
|
'cityCode': '440100', 'districtCode': '440103', 'industryCode': GOV,
|
|
'customerStarLevel': 3, 'relationStarLevel': 3,
|
|
'isBizNegotiated': '0', 'isChild': '0', 'ownerUserId': ADMIN}
|
|
r = api(S, 'POST', '/api/customer', form=form, step='创建撞码')
|
|
expect_code(r, 67003, F, '创建撞已有信用代码 → 67003 硬拦')
|
|
|
|
# 16.2 名称相似（出厂 nameSimilarEnabled=true + threshold=80）
|
|
d = data_of(api(S, 'GET', '/api/customer/check-name', params={'name': simA_name}, step='check-name'))
|
|
check(F, 'check-name 相似命中（出厂阈值 80 生效）', 'pass' if d else 'fail', str(d)[:80])
|
|
form = {'customerName': simA_name, 'customerType': CTYPE, 'provinceCode': '440000',
|
|
'cityCode': '440100', 'districtCode': '440103', 'industryCode': GOV,
|
|
'customerStarLevel': 3, 'relationStarLevel': 3,
|
|
'isBizNegotiated': '0', 'isChild': '0', 'ownerUserId': ADMIN}
|
|
r = api(S, 'POST', '/api/customer', form=form, step='创建相似名')
|
|
d = data_of(r) or {}
|
|
ok = code_of(r) == 0 and d.get('needConfirm') is True
|
|
check(F, '创建名称相似 → needConfirm 弹窗未落库', 'pass' if ok else 'fail',
|
|
f'needConfirm={d.get("needConfirm")}')
|
|
|
|
# 16.3 电话软提示（出厂 phoneSoftCheckEnabled=true；D26 关闭路径依赖单例=票 09，观察点）
|
|
d = data_of(api(S, 'GET', '/api/customer/contact/check-phone',
|
|
params={'phone': '13812340001'}, step='check-phone'))
|
|
check(F, 'check-phone 跨客户软提示命中', 'pass' if d else 'fail', str(d)[:80])
|
|
|
|
# 16.4 导入消费点（f14 已实测：UPSERT 同名→SUSPECT 不写入）
|
|
check(F, '导入侧消费：名称相似 → SUSPECT 不写入（f14 14.9 实测引用）', 'pass', '见 F-14')
|
|
|
|
# 16.5 结论：单例未落地
|
|
check(F, 'customer_dedup_rule 单例未落地（票 09 范围）→ 出厂默认全开生效',
|
|
'pass', 'DefaultCustomerDedupSettingProvider: 名称相似开/80/信用码开/电话软提示开')
|
|
|
|
|
|
# ==================== main ====================
|
|
|
|
def main():
|
|
print(f'==== e2e-heavy 开始 TS={TS} ====')
|
|
flows = []
|
|
try:
|
|
setup_heavy()
|
|
flows.append(('F-13', f13))
|
|
flows.append(('F-14', f14))
|
|
flows.append(('F-15', f15))
|
|
flows.append(('F-16', f16))
|
|
for name, fn in flows:
|
|
print(f'\n—— {name} ——')
|
|
try:
|
|
fn()
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
check(name, f'流程异常中断: {e}', 'fail')
|
|
f14_cleanup()
|
|
teardown_heavy()
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
check('MAIN', f'骨架异常: {e}', 'fail')
|
|
try:
|
|
teardown_heavy()
|
|
except Exception as e2:
|
|
print(f' teardown 也失败: {e2}')
|
|
|
|
# 统计与落盘
|
|
n_pass = sum(1 for c in checks if c['verdict'] == '✅')
|
|
n_warn = sum(1 for c in checks if c['verdict'] == '⚠')
|
|
n_fail = sum(1 for c in checks if c['verdict'] == '❌')
|
|
print(f'\n==== 结果：✅{n_pass} ⚠{n_warn} ❌{n_fail} ====')
|
|
for d in defects:
|
|
print(f" 缺陷 {d['id']}({d['severity']}): {d['title']}")
|
|
out = '.scratch/customer-e2e/'
|
|
json.dump(checks, open(out + 'e2e-heavy-checks.json', 'w', encoding='utf-8'),
|
|
ensure_ascii=False, indent=1)
|
|
json.dump(specimens, open(out + 'specimens-heavy.json', 'w', encoding='utf-8'),
|
|
ensure_ascii=False, indent=1, default=str)
|
|
json.dump(defects, open(out + 'e2e-heavy-defects.json', 'w', encoding='utf-8'),
|
|
ensure_ascii=False, indent=1)
|
|
print(f' checks/specimens/defects 已落盘（specimens {len(specimens)} 端点）')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|