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.
866 lines
53 KiB
866 lines
53 KiB
# -*- coding: utf-8 -*-
|
|
"""e2e-core.py — 客户核心域 API 级实测（票 04，F-01..F-12）
|
|
|
|
产出：report-core.md（三档结论 + 缺陷登记）+ specimens-core.json（票 06 Bruno 注入素材）
|
|
输入：seed-ids.json（票 03 交付，禁止硬编码 id）
|
|
纪律：flat 实路径；Result<Void> 判 code；孤立 401 重试一次；已知缺口不算新缺陷；
|
|
判定依据 原型 > spec > 问用户；坑㉕ 库检每步新连接 autocommit=True。
|
|
"""
|
|
import sys, io, json, time, re
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
|
import requests
|
|
import pymysql
|
|
|
|
BASE = 'http://localhost:8080'
|
|
ADMIN = '739564171091247104'
|
|
BUDDY = '744842318024015872'
|
|
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()}
|
|
|
|
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）→ 返回 None 会变成全 401，
|
|
# 必须重试 + fail fast（票 04 首跑教训：S token=None 十二流程全灭）。
|
|
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)
|
|
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)}' # 重取 token（旧的可能已被单设备顶掉）
|
|
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 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):
|
|
"""断言业务错误码命中（want int）。"""
|
|
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):
|
|
# 字典组首个活跃码值（动态契约，防字典码硬编码漂移；失败回退 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'
|
|
GOV_CHILD = ''
|
|
try:
|
|
_p = dbq("SELECT i.id 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 NULL ORDER BY i.sort_no LIMIT 1")
|
|
if _p:
|
|
GOV_CHILD = dict_one('industry', _p[0]['id']) or ''
|
|
except Exception:
|
|
GOV_CHILD = ''
|
|
|
|
|
|
F = lambda: None # noqa
|
|
|
|
|
|
# ============ F-01 三 workspace 列表 ============
|
|
def f01():
|
|
print('\n== F-01 三 workspace 列表 ==')
|
|
# overview（DataScope 全量，含 archived? 默认 archiveStatus=1）
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/page', form={'current': 1, 'size': 50}))
|
|
if isinstance(d, dict):
|
|
rows = d.get('content') or []
|
|
e2c_rows = [r for r in rows if str(r.get('customerName', '')).startswith('e2c-')]
|
|
check('F-01', 'overview 分页返回 content/total', 'pass', f"total={d.get('total')} e2c行={len(e2c_rows)}")
|
|
col_ok = all(k in (rows[0] if rows else {}) for k in
|
|
['customerNo', 'customerName', 'customerStage', 'customerStarLevel',
|
|
'relationStarLevel', 'ownerUserId', 'opportunityCount', 'archiveStatus'])
|
|
check('F-01', '行字段齐（编号/名称/阶段/双星级/负责人/关联业务计数）',
|
|
'pass' if col_ok else 'fail', f'缺列={[k for k in ["customerNo","customerName","customerStage","ownerUserId","opportunityCount"] if rows and k not in rows[0]]}')
|
|
stage2 = next((r for r in rows if r.get('customerName') == 'e2c-阶段-重潜-乙'), None)
|
|
check('F-01', 'admin（data_scope=4）可见 stage2 样例', 'pass' if stage2 else 'fail', f"id={stage2 and stage2.get('id')}")
|
|
else:
|
|
check('F-01', 'overview 分页', 'fail', f'响应异常 {d}')
|
|
# mine + ASSIGNED
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/mine/page',
|
|
form={'current': 1, 'size': 50, 'viewType': 'ASSIGNED'}))
|
|
mine_names = {r.get('customerName') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-01', 'mine ASSIGNED 含交割-甲/乙/丙（admin 负责人）',
|
|
'pass' if {'e2c-交割-甲', 'e2c-交割-乙', 'e2c-交割-丙'} <= mine_names else 'fail',
|
|
f'e2c命中={sorted(n for n in mine_names if str(n).startswith("e2c-"))}')
|
|
check('F-01', 'mine ASSIGNED 排除协同样例（owner≠当前用户）',
|
|
'pass' if 'e2c-协同-样例' not in mine_names else 'warn', '协同样例 owner=曾偲青')
|
|
# COLLABORATING
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/mine/page',
|
|
form={'current': 1, 'size': 50, 'viewType': 'COLLABORATING'}))
|
|
collab_names = {r.get('customerName') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-01', 'COLLABORATING 视图命中协同样例（admin 是协同人）',
|
|
'pass' if 'e2c-协同-样例' in collab_names else 'fail', f'e2c命中={sorted(n for n in collab_names if str(n).startswith("e2c-"))}')
|
|
# FOLLOW_UP_DUE（seed 直插过期 follow）
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/mine/page',
|
|
form={'current': 1, 'size': 50, 'viewType': 'FOLLOW_UP_DUE'}))
|
|
due_names = {r.get('customerName') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-01', 'FOLLOW_UP_DUE 视图命中提醒-超期（next_follow_time 已过）',
|
|
'pass' if 'e2c-提醒-超期' in due_names else 'fail', f'e2c命中={sorted(n for n in due_names if str(n).startswith("e2c-"))}')
|
|
# pool：enterPoolTime 有值
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/pool/page', form={'current': 1, 'size': 50}))
|
|
pool_rows = [r for r in (d.get('content') or []) if str(r.get('customerName', '')).startswith('e2c-')] if isinstance(d, dict) else []
|
|
ep_ok = all(r.get('enterPoolTime') for r in pool_rows)
|
|
check('F-01', '公海列表 e2c 2 条且 enterPoolTime 有值（可排序）',
|
|
'pass' if len(pool_rows) == 2 and ep_ok else 'fail',
|
|
f'e2c公海={len(pool_rows)} enterPoolTime全有值={ep_ok}')
|
|
# 检索组合
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/page',
|
|
form={'current': 1, 'size': 50, 'keyword': '恒信达'}))
|
|
hit = [r.get('customerName') for r in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-01', 'keyword 检索「恒信达」命中相似对',
|
|
'pass' if isinstance(d, dict) and {'e2c-恒信达科技有限公司', 'e2c-恒信达科技有限责任公司'} <= set(hit or []) else 'warn',
|
|
f'命中={hit}')
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/page',
|
|
form={'current': 1, 'size': 50, 'customerStage': 2}))
|
|
hit = [r.get('customerName') for r in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-01', 'customerStage=2 精确过滤', 'pass' if 'e2c-阶段-重潜-乙' in hit and len(hit) == 1 else 'warn', f'命中={hit}')
|
|
# archived 默认排除（D19）
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/page', form={'current': 1, 'size': 50}))
|
|
names = {r.get('customerName') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-01', 'archiveStatus 缺省=1 默认排除已归档（D19）',
|
|
'pass' if 'e2c-已归档' not in names else 'fail', '')
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/page',
|
|
form={'current': 1, 'size': 50, 'archiveStatus': 2}))
|
|
names2 = {r.get('customerName') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-01', 'archiveStatus=2 可见已归档', 'pass' if 'e2c-已归档' in names2 else 'fail', '')
|
|
|
|
|
|
# ============ F-02 看板三分组 ============
|
|
def f02():
|
|
print('\n== F-02 看板三分组 ==')
|
|
for gc, label, want in [('stage', '阶段', {1, 2, 3}),
|
|
('star', '客户星级', None), ('relation', '关系星级', None)]:
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/board/summary',
|
|
form={'current': 1, 'size': 1, 'groupColumn': gc}))
|
|
if isinstance(d, list):
|
|
gv = {r.get('groupValue') for r in d}
|
|
total = sum(int(r.get('cnt') or r.get('count') or 0) for r in d) # 实测字段名 cnt（字符串数字）
|
|
check('F-02', f'board/summary groupColumn={gc}（{label}）返回有数据分组',
|
|
'pass', f'groups={sorted(gv)} sum={total}')
|
|
if want:
|
|
check('F-02', f'{label} 三列齐（潜在/重潜/已成交）', 'pass' if want <= gv else 'fail', '')
|
|
else:
|
|
check('F-02', f'board/summary groupColumn={gc}', 'fail', f'响应={d}')
|
|
# summary 与 page 自洽
|
|
s = data_of(api(S, 'POST', '/api/customer/workspace/overview/board/summary',
|
|
form={'current': 1, 'size': 1, 'groupColumn': 'stage'}))
|
|
p = data_of(api(S, 'POST', '/api/customer/workspace/overview/page', form={'current': 1, 'size': 1}))
|
|
s_total = sum(int(r.get('cnt') or r.get('count') or 0) for r in s) if isinstance(s, list) else -1
|
|
p_total = p.get('total') if isinstance(p, dict) else -2
|
|
check('F-02', 'summary 总数与 page total 自洽（同 WHERE 口径）',
|
|
'pass' if s_total == int(p_total or 0) else 'warn', f'summary={s_total} page={p_total}')
|
|
# cards
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/overview/board/cards',
|
|
form={'current': 1, 'size': 20, 'groupColumn': 'stage', 'groupValue': 2}))
|
|
names = [r.get('customerName') for r in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-02', 'board/cards stage=2 取卡命中重潜样例',
|
|
'pass' if 'e2c-阶段-重潜-乙' in names else 'fail', f'卡片={names}')
|
|
# 越界 groupColumn 静默忽略
|
|
d = api(S, 'POST', '/api/customer/workspace/overview/board/summary',
|
|
form={'current': 1, 'size': 1, 'groupColumn': 'evil_col'})
|
|
check('F-02', 'groupColumn 越界静默忽略（不炸）', 'pass' if code_of(d) in (0, '0', 200, '200') else 'fail', '')
|
|
|
|
|
|
# ============ F-03 saved-view CRUD 与应用 ============
|
|
def f03():
|
|
print('\n== F-03 saved-view 自定义检索 ==')
|
|
views = {}
|
|
for scope in ['customer.mine', 'customer.overview', 'customer.pool']:
|
|
d = data_of(api(S, 'GET', '/api/preference/view/list', params={'scopeKey': scope}))
|
|
views[scope] = d if isinstance(d, list) else []
|
|
check('F-03', f'list scopeKey={scope}（互不影响）', 'pass' if isinstance(d, list) else 'fail',
|
|
f'现有 {len(views[scope])} 个')
|
|
# 三 scope 各建一视图
|
|
made = {}
|
|
for scope, name, cond in [
|
|
('customer.mine', 'e2c-视图-未跟进', {'field': 'customerName', 'operator': 'contains', 'value': '超期'}),
|
|
('customer.overview', 'e2c-视图-五星', {'field': 'customerStarLevel', 'operator': 'in', 'value': '5'}),
|
|
('customer.pool', 'e2c-视图-公海甲', {'field': 'customerName', 'operator': 'contains', 'value': '公海-甲'})]:
|
|
body = {'viewId': None, 'name': name, 'conditions': [cond],
|
|
'sortField': None, 'sortDirection': None, 'isDefault': False, 'seqNo': 1}
|
|
r = api(S, 'POST', '/api/preference/view/save', json_body=body, params={'scopeKey': scope})
|
|
vid = data_of(r)
|
|
check('F-03', f'save {scope} → {name}', 'pass' if vid else 'fail', f'viewId={vid}')
|
|
made[scope] = vid
|
|
# 应用：mine 视图（contains 超期）→ 只命中提醒-超期
|
|
r = api(S, 'POST', '/api/customer/workspace/mine/page',
|
|
form={'current': 1, 'size': 50, 'savedViewId': made['customer.mine']})
|
|
names = [x.get('customerName') for x in ((data_of(r) or {}).get('content') or [])] if code_of(r) in (0, '0', 200, '200') else []
|
|
check('F-03', 'savedViewId 应用生效（过滤叠加，DataScope 不变）',
|
|
'pass' if names == ['e2c-提醒-超期'] else 'warn', f'命中={names}')
|
|
# set-default + delete
|
|
r = api(S, 'POST', '/api/preference/view/set-default',
|
|
params={'scopeKey': 'customer.mine', 'viewId': made['customer.mine']})
|
|
check('F-03', 'set-default 图钉（单值互斥）', 'pass' if code_of(r) in (0, '0', 200, '200') else 'fail', '')
|
|
r = api(S, 'POST', '/api/preference/view/delete',
|
|
params={'scopeKey': 'customer.pool', 'viewId': made['customer.pool']})
|
|
check('F-03', 'delete 视图', 'pass' if code_of(r) in (0, '0', 200, '200') else 'fail', '')
|
|
# 删除后应用 → 不炸（视图不存在）
|
|
r = api(S, 'POST', '/api/customer/workspace/pool/page',
|
|
form={'current': 1, 'size': 10, 'savedViewId': made['customer.pool']})
|
|
check('F-03', '删除后应用已删视图：不炸（静默/报错均可接受）',
|
|
'pass' if isinstance(r, dict) else 'fail', f'code={code_of(r)}')
|
|
# 清理视图（不留脏数据；mine/overview 也删）
|
|
for scope in ['customer.mine', 'customer.overview']:
|
|
if made.get(scope):
|
|
api(S, 'POST', '/api/preference/view/delete', params={'scopeKey': scope, 'viewId': made[scope]})
|
|
|
|
|
|
# ============ F-04 新增 + 三层查重 ============
|
|
def f04():
|
|
print('\n== F-04 新增客户 + 三层查重 ==')
|
|
tsn = str(int(time.time()))[-6:]
|
|
# L1 停输防抖：check-name
|
|
d = data_of(api(S, 'GET', '/api/customer/check-name', params={'name': 'e2c-恒信达科技有限公司'}))
|
|
check('F-04', 'check-name 精确名命中', 'pass' if d else 'fail', f'返回 {str(d)[:100]}')
|
|
d = data_of(api(S, 'GET', '/api/customer/check-name', params={'name': '宇宙无敌完全不存在公司xyz'}))
|
|
check('F-04', 'check-name 无命中返回空', 'pass' if not d else 'warn', f'返回 {str(d)[:80]}')
|
|
# L1 信用代码
|
|
d = data_of(api(S, 'GET', '/api/customer/check-credit-code', params={'creditCode': '91440101E2CTEST001'}))
|
|
check('F-04', 'check-credit-code 命中全字段样例', 'pass' if d else 'fail', f'返回 {str(d)[:100]}')
|
|
# company-lookup stub
|
|
d = api(S, 'GET', '/api/customer/company-lookup', params={'keyword': '腾讯'})
|
|
check('F-04', 'company-lookup stub 不炸（D5 划出仅 seam）', 'pass' if isinstance(d, dict) else 'warn', f'code={code_of(d)}')
|
|
# L2 名称相似弹窗：不带 confirmSimilar（用 simA 完全同名稳定触发相似命中）
|
|
nm = 'e2c-恒信达科技有限公司'
|
|
form = dict(customerName=nm, 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)
|
|
d = data_of(r)
|
|
nc = d.get('needConfirm') if isinstance(d, dict) else None
|
|
hits = d.get('similarHits') if isinstance(d, dict) else None
|
|
check('F-04', '名称相似 → needConfirm=true + similarHits（未落库）',
|
|
'pass' if nc else ('fail' if r is not None else 'fail'),
|
|
f'needConfirm={nc} hits={str(hits)[:120]}')
|
|
# confirmSimilar=true 重发 → 落库
|
|
form2 = dict(form, confirmSimilar='true')
|
|
r = api(S, 'POST', '/api/customer', form=form2)
|
|
d = data_of(r)
|
|
new_id = str(d.get('id')) if isinstance(d, dict) and d.get('needConfirm') is None else None
|
|
new_no = d.get('customerNo') if isinstance(d, dict) else None
|
|
check('F-04', 'confirmSimilar=true「仍要创建」落库回 id+customerNo',
|
|
'pass' if new_id and new_no else 'fail', f'id={new_id} no={new_no}')
|
|
if new_id:
|
|
C['_tmp_confirmed'] = new_id
|
|
NO['_tmp_confirmed'] = new_no or ''
|
|
# masked 语义：similarHits 无权时显示「存在匹配记录」——admin 全可见带全名，语义记录
|
|
check('F-04', 'similarHits 命中行含编号/名称/负责人/阶段（admin 全可见未脱敏）',
|
|
'pass' if hits and 'customerId' in str(hits[0]) else 'warn', f'首个hit={str(hits[0])[:140] if hits else "无"}')
|
|
# L3 信用代码硬拦（即使 confirmSimilar=true）
|
|
form3 = dict(form2, unifiedCreditCode='91440101E2CTEST001')
|
|
r = api(S, 'POST', '/api/customer', form=form3)
|
|
expect_code(r, 67003, 'F-04', '信用代码已存在 → 67003 硬拦（无仍要创建）')
|
|
# 编号格式
|
|
if new_no:
|
|
check('F-04', '客户编号格式 KH+yyyyMMdd+4位', 'pass' if re.match(r'^KH\d{8}\d{4}$', new_no) else 'warn', f'no={new_no}')
|
|
|
|
|
|
# ============ F-05 编辑客户 ============
|
|
def f05():
|
|
print('\n== F-05 编辑客户 ==')
|
|
fid = C['full']
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}'))
|
|
need = ['customerName', 'customerType', 'industryCode', 'industryChildCode',
|
|
'provinceCode', 'cityCode', 'districtCode', 'unifiedCreditCode',
|
|
'legalRepresentative', 'registeredCapital', 'businessAddress',
|
|
'joinedPresidentClass', 'presidentClassPerson', 'version']
|
|
missing = [k for k in need if d is not None and k not in d]
|
|
ver = d.get('version') if isinstance(d, dict) else None
|
|
check('F-05', 'GET /{id} 全字段回显（含字典名/version）',
|
|
'pass' if d is not None and not missing else 'fail', f'缺={missing} version={ver}')
|
|
# 不带 version → 67005（必填先过，才轮到 CAS 校验；缺必填会先报 67001）
|
|
r = api(S, 'PUT', f'/api/customer/{fid}',
|
|
form={'customerName': 'e2c-全字段-科技v2', 'customerType': CTYPE,
|
|
'provinceCode': '440000', 'cityCode': '440100', 'districtCode': '440103',
|
|
'industryCode': GOV, 'customerStarLevel': 3, 'relationStarLevel': 3,
|
|
'isBizNegotiated': 1, 'isChild': 0})
|
|
got = code_of(r)
|
|
if got in (67005, '67005'):
|
|
check('F-05', 'CAS version 缺失 → 67005（spec D-06）', 'pass', f'code={got}')
|
|
elif got in (67001, '67001') and '乐观锁' in str(r.get('message') or ''):
|
|
check('F-05', 'CAS version 缺失 → 67005（spec D-06）', 'warn',
|
|
f'实际 code=67001 msg=缺少乐观锁版本号——语义可达但错误码与 spec 不符')
|
|
defect('D-03', 'P2', '编辑缺 version 报 67001，spec 约定 67005',
|
|
'PUT /api/customer/{id} 缺 version 返回 67001+「缺少乐观锁版本号」；spec D-06 定义 67005。前端按错误码分流提示会踩坑，需对齐。')
|
|
else:
|
|
check('F-05', 'CAS version 缺失 → 67005（spec D-06）', 'fail', f'code={got} msg={str(r.get("message"))[:80]}')
|
|
# 正常编辑
|
|
r = api(S, 'PUT', f'/api/customer/{fid}',
|
|
form={'customerName': 'e2c-全字段-科技', 'customerType': CTYPE,
|
|
'provinceCode': '440000', 'cityCode': '440100', 'districtCode': '440103',
|
|
'industryCode': GOV, 'industryChildCode': GOV_CHILD, 'customerStarLevel': 3,
|
|
'relationStarLevel': 3, 'isBizNegotiated': 1, 'isChild': 0,
|
|
'unifiedCreditCode': '91440101E2CTEST001', 'remark': 'e2c 编辑实测：备注已更新',
|
|
'version': ver})
|
|
ok = code_of(r) in (0, '0', 200, '200')
|
|
check('F-05', 'PUT 带 version 编辑成功', 'pass' if ok else 'fail',
|
|
f"code={code_of(r)} msg={str(r.get('message'))[:100] if isinstance(r, dict) else ''}")
|
|
# oplog 字段级旧→新叙事
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/oplog/page', params={'current': 1, 'size': 20}))
|
|
logs = d.get('content') or [] if isinstance(d, dict) else []
|
|
upd = next((l for l in logs if l.get('action') == 'UPDATE'), None)
|
|
check('F-05', 'oplog UPDATE 字段级「旧→新」叙事（D13）',
|
|
'pass' if upd and ('备注' in str(upd.get('detail')) or '修改' in str(upd.get('detail'))) else 'warn',
|
|
f'detail={str(upd and upd.get("detail"))[:140]}')
|
|
# 改信用代码撞他人 → 67003
|
|
r = api(S, 'PUT', f'/api/customer/{fid}',
|
|
form={'customerName': 'e2c-全字段-科技', 'customerType': CTYPE,
|
|
'provinceCode': '440000', 'cityCode': '440100', 'districtCode': '440103',
|
|
'industryCode': GOV, 'customerStarLevel': 3, 'relationStarLevel': 3,
|
|
'isBizNegotiated': 1, 'isChild': 0, 'unifiedCreditCode': '91440101E2CHXDA001',
|
|
'version': ver + 1})
|
|
expect_code(r, 67003, 'F-05', '改信用代码撞 simA → 67003')
|
|
|
|
|
|
# ============ F-06 联系人 ============
|
|
def f06():
|
|
print('\n== F-06 联系人 CRUD + 搜索 + 电话软提示 ==')
|
|
fid = C['full']
|
|
# 非法手机号
|
|
r = api(S, 'POST', '/api/customer/contact',
|
|
form={'customerId': fid, 'name': '测试非法', 'jobTitleName': '临时', 'phone': '123'})
|
|
check('F-06', '电话非手机号格式被拒', 'pass' if code_of(r) not in (0, '0', 200, '200') else 'fail',
|
|
f'code={code_of(r)} msg={str(r.get("message"))[:80] if isinstance(r, dict) else ""}')
|
|
# 不可见/不存在客户 → 67002
|
|
r = api(S, 'POST', '/api/customer/contact',
|
|
form={'customerId': 1, 'name': '幽灵', 'jobTitleName': 'x'})
|
|
expect_code(r, 67002, 'F-06', '所属客户不存在 → 67002')
|
|
# page/search
|
|
d = data_of(api(S, 'GET', '/api/customer/contact/page', params={'current': 1, 'size': 20, 'keyword': '张关键'}))
|
|
check('F-06', '联系人分页 keyword 命中', 'pass' if d and int(d.get('total') or 0) >= 1 else 'fail', f'total={d and d.get("total")}')
|
|
d = data_of(api(S, 'GET', '/api/customer/contact/search', params={'keyword': '13812340001', 'limit': 20}))
|
|
names = [x.get('name') for x in d] if isinstance(d, list) else []
|
|
check('F-06', '全局搜索按电话命中 2 条（商机侧数据源）',
|
|
'pass' if set(names) == {'张关键', '王重复'} else 'warn', f'命中={names}')
|
|
dn = data_of(api(S, 'GET', '/api/customer/contact/search', params={'keyword': '李普通'}))
|
|
has_company = dn and all(('customerName' in x) for x in dn) if isinstance(dn, list) else False
|
|
check('F-06', '搜索带出所属公司名（customerName）', 'pass' if has_company else 'fail', f'样本={str(dn and dn[0])[:120]}')
|
|
# check-phone 跨客户软提示
|
|
d = data_of(api(S, 'GET', '/api/customer/contact/check-phone', params={'phone': '13812340001'}))
|
|
check('F-06', 'check-phone 跨客户命中（D26 只提示不阻断）',
|
|
'pass' if isinstance(d, list) and len(d) >= 2 else 'warn', f'命中 {len(d) if isinstance(d, list) else 0} 条')
|
|
# W-01 观察点：source 字段（contact_source 字典组缺失）
|
|
d = data_of(api(S, 'GET', '/api/customer/contact/page', params={'current': 1, 'size': 20, 'keyword': '张关键'}))
|
|
row = (d.get('content') or [{}])[0] if isinstance(d, dict) else {}
|
|
check('F-06', 'W-01 source 字段回显（contact_source 字典组缺失，创建不校验）',
|
|
'pass' if row.get('source') == 'contact_source_01' else 'warn', f"source={row.get('source')}")
|
|
# PUT 编辑 + DELETE
|
|
cid = row.get('id') or list(data_of(api(S, 'GET', '/api/customer/contact/search', params={'keyword': '张关键'})) or [{}])[0].get('id')
|
|
r = api(S, 'PUT', f'/api/customer/contact/{cid}',
|
|
form={'customerId': fid, 'name': '张关键', 'jobTitleName': '采购总监（已更新）',
|
|
'phone': '13812340001', 'isKeyContact': 1, 'isInternal': 1})
|
|
check('F-06', 'PUT 编辑联系人', 'pass' if code_of(r) in (0, '0', 200, '200') else 'fail', '')
|
|
# 新建+删除（不留脏数据）
|
|
r = api(S, 'POST', '/api/customer/contact',
|
|
form={'customerId': fid, 'name': '待删联系人', 'jobTitleName': '临时'})
|
|
tmp_cid = data_of(r)
|
|
r = api(S, 'DELETE', f'/api/customer/contact/{tmp_cid}')
|
|
check('F-06', 'DELETE 软删后列表不含', 'pass' if code_of(r) in (0, '0', 200, '200') else 'fail', f'tmp id={tmp_cid}')
|
|
|
|
|
|
# ============ F-07 详情公共头部 + 8 页签 ============
|
|
def f07():
|
|
print('\n== F-07 详情公共头部 + 8 页签 ==')
|
|
fid = C['full']
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/detail-head'))
|
|
if not isinstance(d, dict):
|
|
check('F-07', 'detail-head', 'fail', f'响应={d}')
|
|
return
|
|
head_need = ['customerNo', 'customerName', 'customerStage', 'customerStageName',
|
|
'customerStarLevel', 'relationStarLevel', 'strategicAgreementLevel',
|
|
'vipCustomerLevel', 'archiveStatus', 'ownerUserId', 'ownerUserNameSnapshot',
|
|
'lastFollowSummary', 'lastFollowTime', 'nextFollowTime',
|
|
'opportunityCount', 'projectCount']
|
|
missing = [k for k in head_need if k not in d]
|
|
check('F-07', '头部关键项齐（阶段只读条/星级/协议VIP快照/负责人/跟进摘要/待跟进提醒/关联计数）',
|
|
'pass' if not missing else 'fail', f'缺={missing}')
|
|
# 汇总卡 5 字段恒 null（不符④拍板 A：后端占位，前端显 --）
|
|
cards = ['wonProjectAmount', 'planEstimateAmount', 'ongoingOpportunityAmount',
|
|
'contractAmount', 'paidAmount']
|
|
not_null = [k for k in cards if d.get(k) is not None]
|
|
check('F-07', '汇总卡 5 字段恒 null（占位契约）', 'pass' if not not_null else 'warn', f'非null={not_null}')
|
|
check('F-07', 'projectCount 恒 null（A5 seam 未接入显 --）',
|
|
'pass' if d.get('projectCount') is None else 'warn', f"projectCount={d.get('projectCount')}")
|
|
# D-01 复验：交割-甲 关联商机计数=1
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["transfer1"]}/detail-head'))
|
|
oc = d.get('opportunityCount') if isinstance(d, dict) else None
|
|
check('F-07', 'D-01 复验：交割-甲 opportunityCount==1（修复后真值）',
|
|
'pass' if str(oc) == '1' else 'fail', f'opportunityCount={oc}') # Long→String 序列化
|
|
# 关联商机页签（port 反查）
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["transfer1"]}/opportunities',
|
|
params={'current': 1, 'size': 10}))
|
|
rows = (d.get('content') or []) if isinstance(d, dict) else []
|
|
names = [str(r.get('oppName') or r.get('opportunityName') or r.get('name') or '') for r in rows]
|
|
check('F-07', '关联商机页签 port 反查命中 e2c-联动-交割',
|
|
'pass' if any('联动-交割' in n for n in names) else 'fail', f'商机={names}')
|
|
# oplog action 值域（11 种白名单）
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/oplog/page', params={'current': 1, 'size': 50}))
|
|
acts = {r.get('action') for r in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
whitelist = {'CREATE', 'UPDATE', 'FOLLOW', 'ARCHIVE', 'RESTORE', 'TRANSFER', 'CLAIM',
|
|
'POOL', 'STAGE_CHANGE', 'MEMBER_ADD', 'MEMBER_REMOVE'}
|
|
check('F-07', 'oplog action 值域 ⊆ 11 种白名单',
|
|
'pass' if acts <= whitelist else 'warn', f'实际={sorted(a for a in acts if a)}')
|
|
# 已归档变体
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["archived"]}/detail-head'))
|
|
check('F-07', '已归档客户详情可开（档案状态=已归档）',
|
|
'pass' if isinstance(d, dict) and d.get('archiveStatus') == 2 else 'fail',
|
|
f"archiveStatus={d.get('archiveStatus') if isinstance(d, dict) else d}")
|
|
# 67002
|
|
r = api(S, 'GET', '/api/customer/999999/detail-head')
|
|
expect_code(r, 67002, 'F-07', '详情客户不存在 → 67002')
|
|
|
|
|
|
# ============ F-08 跟进写入 + 跟进页签 ============
|
|
def f08():
|
|
print('\n== F-08 跟进写入 + 跟进页签 ==')
|
|
fid = C['full']
|
|
way = dict_one('follow_way') or 'follow_way_01'
|
|
old = dbq('SELECT last_valid_follow_time t FROM customer WHERE id=%s', (fid,))
|
|
old_t = old[0]['t'] if old else None
|
|
# 空 content → 67009
|
|
r = api(S, 'POST', f'/api/customer/{fid}/follow', form={'followWay': way, 'followContent': ''})
|
|
expect_code(r, 67009, 'F-08', '跟进内容空 → 67009')
|
|
# nextFollowTime 过去 → 67009（ISO T 分隔，空格 400——seed 实测）
|
|
r = api(S, 'POST', f'/api/customer/{fid}/follow',
|
|
form={'followWay': way, 'followContent': 'e2c-F08 过去时间样例',
|
|
'nextFollowTime': '2026-09-01T10:00:00'})
|
|
got = code_of(r)
|
|
if got in (67009, '67009'):
|
|
check('F-08', 'nextFollowTime 早于当前 → 67009（spec）', 'pass', f'code={got}')
|
|
elif got in (0, '0', 200, '200'):
|
|
check('F-08', 'nextFollowTime 早于当前 → 67009（spec）', 'warn',
|
|
f'实际 code=0 落库成功——过期下次跟进未拦截，会造成永久超期提醒')
|
|
defect('D-02', 'P2', '跟进 nextFollowTime 早于当前未拦截（spec 应 67009）',
|
|
'POST /{id}/follow 传过去时间 nextFollowTime=2026-09-01T10:00:00 返回成功落库；spec F-08 约定 67009。会产生永远无法消除的超期提醒（FOLLOW_UP_DUE 视图命中）。')
|
|
else:
|
|
check('F-08', 'nextFollowTime 早于当前 → 67009（spec）', 'fail', f'code={got}')
|
|
# 正常写入
|
|
content = 'e2c-F08 跟进实测：电话沟通年度合作意向'
|
|
nxt = time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime(time.time() + 7 * 86400))
|
|
r = api(S, 'POST', f'/api/customer/{fid}/follow',
|
|
form={'followWay': way, 'followContent': content, 'nextFollowTime': nxt})
|
|
new_fid = data_of(r)
|
|
check('F-08', '正常写入返回跟进 id', 'pass' if new_fid else 'fail', f'id={new_fid} way={way}')
|
|
rows = dbq('SELECT id FROM customer_follow WHERE customer_id=%s AND deleted=0 ORDER BY id DESC LIMIT 1', (fid,))
|
|
check('F-08', 'customer_follow 行落库（append-only）',
|
|
'pass' if rows and str(rows[0]['id']) == str(new_fid) else 'fail', '')
|
|
anchor = dbq('SELECT last_valid_follow_time t FROM customer WHERE id=%s', (fid,))
|
|
new_t = anchor[0]['t'] if anchor else None
|
|
check('F-08', 'D25 锚点 last_valid_follow_time 刷新',
|
|
'pass' if new_t is not None and (old_t is None or str(new_t) > str(old_t)) else 'fail',
|
|
f'{old_t} → {new_t}')
|
|
# follow/page + followWay 筛选
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/follow/page', params={'current': 1, 'size': 20}))
|
|
tops = [str(x.get('followContent') or '') for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-08', '跟进页签时间倒序含新记录', 'pass' if content in tops else 'fail', f'首条={tops[:1]}')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/follow/page',
|
|
params={'current': 1, 'size': 20, 'followWay': way}))
|
|
fw_ok = all(x.get('followWay') == way for x in (d.get('content') or [])) if isinstance(d, dict) else False
|
|
check('F-08', 'followWay 筛选生效', 'pass' if fw_ok and fw_ok is not None else 'warn', f'way={way}')
|
|
# 30s 防重
|
|
r = api(S, 'POST', f'/api/customer/{fid}/follow',
|
|
form={'followWay': way, 'followContent': content, 'nextFollowTime': nxt})
|
|
expect_code(r, 67009, 'F-08', '30s 内同内容重复提交 → 67009')
|
|
# detail-head 摘要联动 + followFuture 待跟进提醒
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/detail-head'))
|
|
check('F-08', '头部最近跟进摘要联动最新一条',
|
|
'pass' if isinstance(d, dict) and d.get('lastFollowSummary') == content else 'fail',
|
|
f"summary={str(d.get('lastFollowSummary') if isinstance(d, dict) else d)[:60]}")
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["followFuture"]}/detail-head'))
|
|
check('F-08', '待跟进提醒 nextFollowTime 非空（seed 明天回访）',
|
|
'pass' if isinstance(d, dict) and d.get('nextFollowTime') else 'fail',
|
|
f"next={d.get('nextFollowTime') if isinstance(d, dict) else d}")
|
|
# append-only：跟进无改删口
|
|
r = api(S, 'PUT', f'/api/customer/{fid}/follow/{new_fid}', form={'followContent': 'x'})
|
|
check('F-08', 'append-only：跟进无编辑口（PUT 不成功即可）',
|
|
'pass' if code_of(r) not in (0, '0', 200, '200') else 'warn', f'code={code_of(r)}')
|
|
|
|
|
|
# ============ F-09 关注 / 重点标记（一行两态） ============
|
|
def f09():
|
|
print('\n== F-09 关注/重点标记（一行两态）==')
|
|
uid = ADMIN
|
|
|
|
def focus_rows(cid):
|
|
return dbq('SELECT id, starred FROM customer_focus WHERE customer_id=%s AND user_id=%s',
|
|
(cid, uid))
|
|
api(S, 'POST', f'/api/customer/{C["stage1"]}/focus')
|
|
api(S, 'POST', f'/api/customer/{C["stage1"]}/focus')
|
|
rows = focus_rows(C['stage1'])
|
|
check('F-09', 'focus 幂等：两次关注仍单行', 'pass' if len(rows) == 1 else 'fail', f'行数={len(rows)}')
|
|
api(S, 'POST', f'/api/customer/{C["member"]}/star')
|
|
rows = focus_rows(C['member'])
|
|
check('F-09', 'star 未关注自动建行 starred=1',
|
|
'pass' if len(rows) == 1 and rows[0]['starred'] == 1 else 'fail', f'{rows}')
|
|
api(S, 'POST', f'/api/customer/{C["member"]}/unstar')
|
|
rows = focus_rows(C['member'])
|
|
check('F-09', 'unstar 保留行仅 starred=0',
|
|
'pass' if len(rows) == 1 and rows[0]['starred'] == 0 else 'fail', f'{rows}')
|
|
r = api(S, 'POST', '/api/customer/focus-batch',
|
|
params={'ids': ','.join([C['star5'], C['pool1'], C['pool2']])})
|
|
n = sum(len(focus_rows(C[k])) for k in ['star5', 'pool1', 'pool2'])
|
|
check('F-09', 'focus-batch 3 客户各建 1 行',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') and n == 3 else 'fail', f'行数合计={n}')
|
|
d = data_of(api(S, 'POST', '/api/customer/workspace/pool/page',
|
|
form={'current': 1, 'size': 50, 'viewType': 'FOCUSED'}))
|
|
names = [x.get('customerName') for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-09', 'pool FOCUSED 视图命中关注的公海客户',
|
|
'pass' if {'e2c-公海-甲', 'e2c-公海-乙'} <= set(names or []) else 'fail', f'命中={names}')
|
|
api(S, 'POST', f'/api/customer/{C["member"]}/unfocus')
|
|
rows = focus_rows(C['member'])
|
|
check('F-09', 'unfocus 物理删行（连带清 starred）', 'pass' if len(rows) == 0 else 'fail', f'剩余={len(rows)}')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["stage1"]}/oplog/page', params={'current': 1, 'size': 50}))
|
|
acts = {x.get('action') for x in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-09', '关注动作不写 oplog',
|
|
'pass' if not (acts & {'FOCUS', 'UNFOCUS', 'STAR', 'UNSTAR'}) else 'warn',
|
|
f'actions={sorted(a for a in acts if a)}')
|
|
for k in ['stage1', 'star5', 'pool1', 'pool2']:
|
|
api(S, 'POST', f'/api/customer/{C[k]}/unfocus')
|
|
|
|
|
|
# ============ F-10 团队成员 ============
|
|
def f10():
|
|
print('\n== F-10 团队成员 ==')
|
|
fid = C['full']
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/members'))
|
|
if not isinstance(d, list) or not d:
|
|
check('F-10', 'members 列表', 'fail', f'响应={str(d)[:120]}')
|
|
return
|
|
first = d[0]
|
|
rk = next((k for k in first.keys() if 'role' in k.lower()), None)
|
|
rv = str(first.get(rk, '')) if rk else ''
|
|
check('F-10', '负责人在首位（ROLE_OWNER 快照）',
|
|
'pass' if 'OWNER' in rv.upper() or '负责人' in rv else 'warn', f'首行={str(first)[:120]}')
|
|
# 两阶段校验：整批拒绝
|
|
r = api(S, 'POST', f'/api/customer/{fid}/members', form={'memberUserIds': ''})
|
|
check('F-10', '名单空被拒（67016/绑定拒）',
|
|
'pass' if code_of(r) not in (0, '0', 200, '200') else 'fail', f'code={code_of(r)}')
|
|
r = api(S, 'POST', f'/api/customer/{fid}/members', form={'memberUserIds': ADMIN})
|
|
expect_code(r, 67016, 'F-10', '含负责人 → 67016')
|
|
# 正常添加 BUDDY
|
|
r = api(S, 'POST', f'/api/customer/{fid}/members', form={'memberUserIds': BUDDY})
|
|
ok = code_of(r) in (0, '0', 200, '200')
|
|
check('F-10', '添加协同人 BUDDY', 'pass' if ok else 'fail', f'code={code_of(r)}')
|
|
rows = dbq('SELECT id, delete_key FROM customer_team_member WHERE customer_id=%s AND user_id=%s ORDER BY id',
|
|
(fid, BUDDY))
|
|
check('F-10', 'team_member 行在（delete_key=0）',
|
|
'pass' if rows and rows[-1]['delete_key'] == 0 else 'fail', f'{rows}')
|
|
r = api(S, 'POST', f'/api/customer/{fid}/members', form={'memberUserIds': BUDDY})
|
|
expect_code(r, 67016, 'F-10', '重复加入 → 67016（整批拒绝）')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/oplog/page', params={'current': 1, 'size': 20}))
|
|
acts = [x.get('action') for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-10', 'oplog MEMBER_ADD', 'pass' if 'MEMBER_ADD' in acts else 'warn', f'actions={acts[:6]}')
|
|
r = api(S, 'DELETE', f'/api/customer/{fid}/members/{BUDDY}')
|
|
rows = dbq('SELECT delete_key FROM customer_team_member WHERE customer_id=%s AND user_id=%s ORDER BY id',
|
|
(fid, BUDDY))
|
|
check('F-10', '移除软删（delete_key>0）',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') and rows and rows[-1]['delete_key'] > 0 else 'fail', f'{rows}')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{fid}/oplog/page', params={'current': 1, 'size': 20}))
|
|
acts = [x.get('action') for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-10', 'oplog MEMBER_REMOVE', 'pass' if 'MEMBER_REMOVE' in acts else 'warn', '')
|
|
r = api(S, 'POST', f'/api/customer/{fid}/members', form={'memberUserIds': BUDDY})
|
|
rows = dbq('SELECT delete_key FROM customer_team_member WHERE customer_id=%s AND user_id=%s ORDER BY id',
|
|
(fid, BUDDY))
|
|
check('F-10', '移除后可重加（delete_key 复用键）',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') and rows and rows[-1]['delete_key'] == 0 else 'fail', '')
|
|
api(S, 'DELETE', f'/api/customer/{fid}/members/{BUDDY}') # 还原 seed 态
|
|
|
|
|
|
# ============ F-11 归属动作（领取/分配/抛公海/归档/恢复） ============
|
|
def f11():
|
|
print('\n== F-11 归属动作 ==')
|
|
# 1) admin 领取 pool1
|
|
r = api(S, 'POST', '/api/customer/claim', params={'id': C['pool1']})
|
|
ok = code_of(r) in (0, '0', 200, '200')
|
|
row = dbq('SELECT owner_user_id o, owner_user_name_snapshot n, owner_dept_id d, '
|
|
'owner_dept_name_snapshot dn, last_valid_follow_time t FROM customer WHERE id=%s',
|
|
(C['pool1'],))
|
|
pool1_before = row[0] if row else None
|
|
check('F-11', '领取公海客户 → 当前用户（owner+快照）', 'pass' if ok else 'fail', f'code={code_of(r)}')
|
|
check('F-11', '领取后 D25 锚点赋当下（last_valid_follow_time 非空）',
|
|
'pass' if row and row[0]['t'] is not None else 'fail', f"t={row[0]['t'] if row else None}")
|
|
# 2) 重复领取（非公海）失败
|
|
r = api(S, 'POST', '/api/customer/claim', params={'id': C['pool1']})
|
|
check('F-11', '重复领取失败（非公海态）',
|
|
'pass' if code_of(r) not in (0, '0', 200, '200') else 'fail',
|
|
f'code={code_of(r)} msg={str(r.get("message"))[:80] if isinstance(r, dict) else ""}')
|
|
# 3) BUDDY（无角色）领取观察点：目标已被领必败，错误语义记录
|
|
r = api(B, 'POST', '/api/customer/claim', params={'id': C['pool1']})
|
|
got = code_of(r)
|
|
check('F-11', '观察点：BUDDY（无角色）领取被拒', 'pass' if got not in (0, '0', 200, '200') else 'warn',
|
|
f'code={got} msg={str(r.get("message"))[:80] if isinstance(r, dict) else ""}')
|
|
if got in (0, '0', 200, '200'):
|
|
defect('D-02', 'P1', '无角色用户可领取已被领取的公海客户（越权观察点）',
|
|
f'BUDDY（无任何角色）claim 非公海态客户返回成功——DataScope/公海态校验未拦；已 DB 还原')
|
|
dbx('UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot=%s, owner_dept_id=%s, '
|
|
'owner_dept_name_snapshot=%s WHERE id=%s',
|
|
(pool1_before['o'], pool1_before['n'], pool1_before['d'], pool1_before['dn'], C['pool1']))
|
|
# 4) assign 本人 → 67012
|
|
r = api(S, 'POST', '/api/customer/assign', params={'id': C['full'], 'userId': ADMIN})
|
|
expect_code(r, 67012, 'F-11', '分配给本人 → 67012')
|
|
# 5) assign stage1 → BUDDY 成功（启用在职校验通过）
|
|
s1_before = dbq('SELECT owner_user_id o, owner_user_name_snapshot n, owner_dept_id d, '
|
|
'owner_dept_name_snapshot dn FROM customer WHERE id=%s', (C['stage1'],))[0]
|
|
r = api(S, 'POST', '/api/customer/assign', params={'id': C['stage1'], 'userId': BUDDY})
|
|
ok = code_of(r) in (0, '0', 200, '200')
|
|
row = dbq('SELECT owner_user_id o, owner_user_name_snapshot n FROM customer WHERE id=%s', (C['stage1'],))
|
|
check('F-11', '分配给 BUDDY 成功（owner 换绑+快照）',
|
|
'pass' if ok and row and str(row[0]['o']) == BUDDY else 'fail',
|
|
f'code={code_of(r)} owner={row[0]["o"] if row else None}')
|
|
dbx('UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot=%s, owner_dept_id=%s, '
|
|
'owner_dept_name_snapshot=%s WHERE id=%s',
|
|
(s1_before['o'], s1_before['n'], s1_before['d'], s1_before['dn'], C['stage1']))
|
|
check('F-11', '实验后 DB 还原 stage1 归属（assign 回本人被 67012 拦）', 'pass', '')
|
|
# 6) 抛公海：进行中商机 → 67007
|
|
r = api(S, 'POST', '/api/customer/release-pool', params={'id': C['transfer2']})
|
|
expect_code(r, 67007, 'F-11', '有进行中商机抛公海 → 67007')
|
|
# 7) 抛公海 stage1：清 owner、部门锚点保留
|
|
r = api(S, 'POST', '/api/customer/release-pool', params={'id': C['stage1']})
|
|
ok = code_of(r) in (0, '0', 200, '200')
|
|
row = dbq('SELECT owner_user_id o, owner_dept_id d FROM customer WHERE id=%s', (C['stage1'],))
|
|
check('F-11', '抛公海清 owner、部门锚点保留',
|
|
'pass' if ok and row and row[0]['o'] is None and row[0]['d'] is not None else 'fail',
|
|
f'owner={row[0]["o"] if row else None} dept={row[0]["d"] if row else None}')
|
|
# 8) claim 还原
|
|
r = api(S, 'POST', '/api/customer/claim', params={'id': C['stage1']})
|
|
check('F-11', 'claim 还原 stage1（owner=admin）',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') else 'fail', f'code={code_of(r)}')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["stage1"]}/oplog/page', params={'current': 1, 'size': 50}))
|
|
acts = {x.get('action') for x in (d.get('content') or [])} if isinstance(d, dict) else set()
|
|
check('F-11', '归属动作 oplog（TRANSFER/POOL/CLAIM 齐——assign 写 TRANSFER）',
|
|
'pass' if {'TRANSFER', 'POOL', 'CLAIM'} <= acts else 'warn', f'actions={sorted(a for a in acts if a)}')
|
|
# 9) 归档：进行中商机 → 67006
|
|
r = api(S, 'POST', '/api/customer/archive', params={'id': C['transfer2']})
|
|
expect_code(r, 67006, 'F-11', '有进行中商机归档 → 67006')
|
|
# 10) archive-batch 整批预检：任一失败整批不执行
|
|
r = api(S, 'POST', '/api/customer/archive-batch',
|
|
params={'ids': f'{C["star5"]},{C["transfer2"]}'})
|
|
st = dbq('SELECT archive_status a FROM customer WHERE id=%s', (C['star5'],))[0]['a']
|
|
check('F-11', 'archive-batch 整批预检：含阻断户整批不执行（star5 仍有效）',
|
|
'pass' if st == 1 else 'fail', f'star5 archive_status={st} resp={str(r)[:100]}')
|
|
# 11) 单条归档 + 详情变体 + 恢复
|
|
r = api(S, 'POST', '/api/customer/archive', params={'id': C['star5']})
|
|
st = dbq('SELECT archive_status a FROM customer WHERE id=%s', (C['star5'],))[0]['a']
|
|
check('F-11', '归档 star5（archive_status=2）', 'pass' if code_of(r) in (0, '0', 200, '200') and st == 2 else 'fail', f'st={st}')
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["star5"]}/detail-head'))
|
|
check('F-11', '已归档 detail-head archiveStatus=2',
|
|
'pass' if isinstance(d, dict) and d.get('archiveStatus') == 2 else 'fail', '')
|
|
r = api(S, 'POST', '/api/customer/restore', params={'id': C['star5']})
|
|
st = dbq('SELECT archive_status a FROM customer WHERE id=%s', (C['star5'],))[0]['a']
|
|
check('F-11', '恢复 star5（archive_status=1）',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') and st == 1 else 'fail', f'st={st}')
|
|
# 还原 seed 态：pool1 已被 admin 领走 → 抛回公海（供票 05 公海规则用例；顺带验证抛公海写 enter_pool_time）
|
|
r = api(S, 'POST', '/api/customer/release-pool', params={'id': C['pool1']})
|
|
row = dbq('SELECT owner_user_id o, enter_pool_time t FROM customer WHERE id=%s', (C['pool1'],))
|
|
check('F-11', '还原：pool1 抛回公海（owner 空且 enter_pool_time 有值）',
|
|
'pass' if code_of(r) in (0, '0', 200, '200') and row and row[0]['o'] is None and row[0]['t'] else 'fail',
|
|
f"owner={row[0]['o'] if row else None} enter={row[0]['t'] if row else None}")
|
|
|
|
|
|
# ============ F-12 商机侧硬依赖三端点（跨模块回归） ============
|
|
def f12():
|
|
print('\n== F-12 商机侧硬依赖（quick-create/search/contacts）==')
|
|
# 清理历史快创客户（名称相似算法对「仅数字段不同」判相似 → 会弹 needConfirm 卡死主用例）
|
|
dbx("DELETE FROM customer_oplog WHERE customer_id IN (SELECT id FROM (SELECT id FROM customer WHERE customer_name LIKE 'e2c-快创%') x)")
|
|
dbx("DELETE FROM customer WHERE customer_name LIKE 'e2c-快创%'")
|
|
tsn = str(int(time.time()))[-6:]
|
|
form = {'customerName': f'e2c-快创-{tsn}', 'customerType': CTYPE,
|
|
'provinceCode': '440000', 'cityCode': '440100', 'industryCode': GOV,
|
|
'customerStarLevel': 2, 'relationStarLevel': 2}
|
|
r = api(S, 'POST', '/api/customer/quick-create', form=form)
|
|
d = data_of(r)
|
|
qid = d.get('id') if isinstance(d, dict) and d.get('needConfirm') is None else None
|
|
check('F-12', 'quick-create 最小集落库回 id+customerNo',
|
|
'pass' if qid and isinstance(d, dict) and d.get('customerNo') else 'fail', f'{str(d)[:120]}')
|
|
if qid:
|
|
row = dbq('SELECT owner_user_id o, is_child c, is_biz_negotiated b FROM customer WHERE id=%s', (qid,))
|
|
if row:
|
|
o, c, b = row[0]['o'], row[0]['c'], row[0]['b']
|
|
check('F-12', '默认值：owner=当前用户 / is_child=否 / isBizNegotiated=否',
|
|
'pass' if str(o) == ADMIN and int(c or 0) == 0 and int(b or 0) == 0 else 'warn',
|
|
f'owner={o} is_child={c} is_biz={b}')
|
|
else:
|
|
check('F-12', '默认值 DB 验', 'warn', '行未查到（列名差异）')
|
|
# 已知缺口复现：quick-create 无 confirmSimilar 参数
|
|
r = api(S, 'POST', '/api/customer/quick-create',
|
|
form=dict(form, customerName='e2c-恒信达科技有限公司'))
|
|
d = data_of(r)
|
|
nc = d.get('needConfirm') if isinstance(d, dict) else None
|
|
r = api(S, 'POST', '/api/customer/quick-create',
|
|
form=dict(form, customerName='e2c-恒信达科技有限公司', confirmSimilar='true'))
|
|
d = data_of(r)
|
|
nc2 = d.get('needConfirm') if isinstance(d, dict) else None
|
|
check('F-12', '已知缺口复现：无 confirmSimilar，命中相似无法仍要创建（预登记不算新缺陷）',
|
|
'pass' if nc and nc2 else 'warn', f'首次needConfirm={bool(nc)} 带参重发needConfirm={bool(nc2)}')
|
|
# search 三维模糊 + 类型精确 + 排除
|
|
d = data_of(api(S, 'POST', '/api/customer/search', form={'current': 1, 'size': 20, 'keyword': '恒信达'}))
|
|
names = [x.get('customerName') for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-12', 'search 名称模糊命中 simA/simB',
|
|
'pass' if {'e2c-恒信达科技有限公司', 'e2c-恒信达科技有限责任公司'} <= set(names or []) else 'warn',
|
|
f'命中={names}')
|
|
d = data_of(api(S, 'POST', '/api/customer/search',
|
|
form={'current': 1, 'size': 20, 'keyword': '恒信达', 'customerType': CTYPE}))
|
|
t_ok = isinstance(d, dict) and all(x.get('customerType') == CTYPE for x in (d.get('content') or []))
|
|
check('F-12', 'search 类型精确过滤', 'pass' if t_ok else 'warn',
|
|
f"total={d.get('total') if isinstance(d, dict) else d} ctype={CTYPE}")
|
|
d = data_of(api(S, 'POST', '/api/customer/search',
|
|
form={'current': 1, 'size': 20, 'keyword': '恒信达', 'excludeCustomerIds': C['simA']}))
|
|
ids_h = [str(x.get('id')) for x in (d.get('content') or [])] if isinstance(d, dict) else []
|
|
check('F-12', 'search excludeCustomerIds 排除 simA（按 id 判，同名 tmp 不干扰）',
|
|
'pass' if str(C['simA']) not in ids_h else 'fail', f'命中id数={len(ids_h)}')
|
|
# contacts 按客户带出
|
|
d = data_of(api(S, 'GET', f'/api/customer/{C["full"]}/contacts'))
|
|
names3 = [x.get('name') for x in d] if isinstance(d, list) else []
|
|
check('F-12', 'contacts 按客户带出（张关键/李普通/王重复）',
|
|
'pass' if {'张关键', '李普通', '王重复'} <= set(names3 or []) else 'fail', f'名单={names3}')
|
|
|
|
|
|
def main():
|
|
t0 = time.time()
|
|
print(f'== 客户核心域 E2E（票 04）== e2c 客户 {len(C)} 个')
|
|
for fn in [f01, f02, f03, f04, f05, f06, f07, f08, f09, f10, f11, f12]:
|
|
try:
|
|
fn()
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
check(fn.__name__.upper(), '流程级异常', 'fail', str(e)[:160])
|
|
el = time.time() - t0
|
|
npass = sum(1 for c in checks if c['verdict'] == '✅')
|
|
nwarn = sum(1 for c in checks if c['verdict'] == '⚠')
|
|
nfail = sum(1 for c in checks if c['verdict'] == '❌')
|
|
print(f'\n== 汇总 == ✅{npass} ⚠{nwarn} ❌{nfail} 耗时 {el:.0f}s specimens={len(specimens)}')
|
|
for d in defects:
|
|
print(f" ⚑ {d['id']}({d['severity']}) {d['title']}")
|
|
json.dump({'checks': checks, 'defects': defects, 'elapsedSec': round(el, 1)},
|
|
open('.scratch/customer-e2e/e2e-core-checks.json', 'w', encoding='utf-8'),
|
|
ensure_ascii=False, indent=1)
|
|
json.dump(specimens, open('.scratch/customer-e2e/specimens-core.json', 'w', encoding='utf-8'),
|
|
ensure_ascii=False, indent=1)
|
|
print(' ✔ 落盘 e2e-core-checks.json / specimens-core.json')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|