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.
94 lines
4.1 KiB
94 lines
4.1 KiB
15 hours ago
|
# -*- coding: utf-8 -*-
|
||
|
|
"""smoke-userlist-collab.py — D-07 影响面冒烟（UserQueryServiceImpl）+ collab owner 四字段核对
|
||
|
|
|
||
|
|
1. POST /api/system/users/page（表单分页）——计时 + 总数 + 首行部门名解析（getChildDeptIds 消费面）
|
||
|
|
2. GET /api/system/users/stats?deptId=广东保伦本部——计时（子树统计，L168 同源消费）
|
||
|
|
3. DB：collab 协同样例 owner 四字段（heavy F-13 圈定副作用回归点）
|
||
|
|
4. DB：e2c-销售甲 部门还原核对（F-13.10 临时挂入的还原确认）
|
||
|
|
落盘：.scratch/customer-defectfix/smoke-userlist-collab.json
|
||
|
|
"""
|
||
|
|
import io, sys, json, time
|
||
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
|
import requests
|
||
|
|
import pymysql
|
||
|
|
|
||
|
|
BASE = 'http://localhost:8080'
|
||
|
|
ADMIN = '739564171091247104'
|
||
|
|
OUT = '.scratch/customer-defectfix/smoke-userlist-collab.json'
|
||
|
|
DEPT_ROOT = '744841292348915712' # 广东保伦本部（D-07 探测同款参数）
|
||
|
|
COLLAB_ID = '750844482391375872' # collab 协同样例（票 04 还原实录对象）
|
||
|
|
SALES_ID = '760000000000000001'
|
||
|
|
SALES_DEPT_SEED = '744700334353416192' # 华南销售部(seed)
|
||
|
|
BUDDY = '744842318024015872'
|
||
|
|
BUDDY_DEPT = '744841292483133440'
|
||
|
|
|
||
|
|
|
||
|
|
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()
|
||
|
|
|
||
|
|
|
||
|
|
r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': ADMIN}, timeout=30)
|
||
|
|
tok = r.json().get('data')
|
||
|
|
tok = tok if isinstance(tok, str) else (tok or {}).get('token')
|
||
|
|
S = requests.Session()
|
||
|
|
S.headers['Authorization'] = f'Bearer {tok}'
|
||
|
|
report = {}
|
||
|
|
|
||
|
|
|
||
|
|
def step(name, ok, detail=''):
|
||
|
|
report.setdefault('steps', []).append({'step': name, 'ok': ok, 'detail': detail})
|
||
|
|
print((' pass' if ok else ' FAIL'), name, '—', detail)
|
||
|
|
|
||
|
|
|
||
|
|
# ---- 1. users/page 冒烟 ----
|
||
|
|
t0 = time.time()
|
||
|
|
r = S.post(f'{BASE}/api/system/users/page',
|
||
|
|
data={'current': 1, 'size': 10}, timeout=30)
|
||
|
|
el = time.time() - t0
|
||
|
|
body = r.json() if r.status_code == 200 else {}
|
||
|
|
d = body.get('data') or {}
|
||
|
|
content = d.get('content') or []
|
||
|
|
first = content[0] if content else {}
|
||
|
|
ok = (r.status_code == 200 and body.get('code') == 0
|
||
|
|
and int(d.get('total') or 0) > 0 and el < 5
|
||
|
|
and bool(first.get('primaryDeptName')))
|
||
|
|
step('users/page 冒烟（<5s 且部门名已解析）', ok,
|
||
|
|
f"HTTP {r.status_code} elapsed={el:.2f}s total={d.get('total')} "
|
||
|
|
f"first={first.get('username')}/{first.get('primaryDeptName')}")
|
||
|
|
|
||
|
|
# ---- 2. users/stats 子树冒烟 ----
|
||
|
|
t0 = time.time()
|
||
|
|
r = S.get(f'{BASE}/api/system/users/stats', params={'deptId': DEPT_ROOT}, timeout=30)
|
||
|
|
el2 = time.time() - t0
|
||
|
|
body2 = r.json() if r.status_code == 200 else {}
|
||
|
|
ok = r.status_code == 200 and body2.get('code') == 0 and el2 < 5
|
||
|
|
step('users/stats 子树冒烟（<5s）', ok,
|
||
|
|
f"HTTP {r.status_code} elapsed={el2:.2f}s data={json.dumps(body2.get('data'), ensure_ascii=False)[:160]}")
|
||
|
|
|
||
|
|
# ---- 3. collab owner 四字段 ----
|
||
|
|
rows = dbq("SELECT owner_user_id o, owner_user_name_snapshot on_, owner_dept_id od, "
|
||
|
|
"owner_dept_name_snapshot odn FROM customer WHERE id=%s", (COLLAB_ID,))
|
||
|
|
row = rows[0] if rows else {}
|
||
|
|
ok = (row and str(row['o']) == BUDDY and str(row['od']) == BUDDY_DEPT
|
||
|
|
and row['on_'] == '曾偲青')
|
||
|
|
step('collab 同样例 owner 四字段（曾偲青/职员部）', ok,
|
||
|
|
json.dumps({k: str(v) for k, v in row.items()}, ensure_ascii=False))
|
||
|
|
|
||
|
|
# ---- 4. 销售甲部门还原 ----
|
||
|
|
rows = dbq("SELECT dept_id d FROM crm_auth_user WHERE id=%s", (SALES_ID,))
|
||
|
|
ok = rows and str(rows[0]['d']) == SALES_DEPT_SEED
|
||
|
|
step('e2c-销售甲 部门还原（华南销售部 seed）', ok,
|
||
|
|
f"dept={rows[0]['d'] if rows else '无'}")
|
||
|
|
|
||
|
|
report['ALL_OK'] = all(s['ok'] for s in report.get('steps', []))
|
||
|
|
json.dump(report, open(OUT, 'w', encoding='utf-8'), ensure_ascii=False, indent=2)
|
||
|
|
print('\n== 冒烟+核对', 'ALL_OK ✅' if report['ALL_OK'] else '存在 FAIL ❌', '，证据落盘', OUT, '==')
|