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.

61 lines
2.9 KiB

19 hours ago
# -*- coding: utf-8 -*-
"""r2 票 01 静态验收③:扫描 e2e-*-r2.py 全部 api()/api_raw() 调用,对照 dump_api_out.txt 73 端点基准。
产出调用×端点命中表stdout + calls-r2.json"""
import json
import re
import io
import sys
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
SCRIPTS = [r'e:\code\crm-backend-matt\.scratch\customer-e2e\e2e-core-r2.py',
r'e:\code\crm-backend-matt\.scratch\customer-e2e\e2e-heavy-r2.py']
DUMP = r'e:\code\crm-backend-matt\.scratch\customer-rework\dump_api_out.txt'
# 基准:dump_api_out.txt 全部端点(含跨模块 crm-rule/crm-preference)
endpoints = set()
for line in open(DUMP, encoding='utf-8'):
m = re.match(r'^(GET|POST) (\S+)', line.strip())
if m:
endpoints.add((m.group(1), m.group(2)))
print(f'基准端点(dump_api_out.txt 全模块):{len(endpoints)}')
# 扫描调用:api(sess, 'POST', '/api/...', ...) / api_raw(sess, 'GET', '/api/...', ...)
calls = set()
per_script = {}
for sp in SCRIPTS:
src = open(sp, encoding='utf-8').read()
found = set()
for m in re.finditer(r"api(?:_raw)?\(\s*\w+,\s*'(GET|POST)',\s*(?:f?)'(/api/[^']+)'", src):
found.add((m.group(1), m.group(2)))
per_script[sp.split('\\')[-1]] = sorted(found)
calls |= found
print(f'\nr2 脚本调用端点:{len(calls)}')
unknown = sorted(c for c in calls if c not in endpoints)
# 预期例外:follow/edit 是「append-only 无编辑口」断言故意打的未注册端点
KNOWN_EXCEPTIONS = {('POST', '/api/customer/follow/edit')}
real_unknown = [c for c in unknown if c not in KNOWN_EXCEPTIONS]
print('\n== 命中情况 ==')
for m, p in sorted(calls):
tag = 'OK' if (m, p) in endpoints else ('预期例外(append-only断言)' if (m, p) in KNOWN_EXCEPTIONS else '!!未命中')
print(f' [{tag}] {m} {p}')
# 反向:73 端点中客户域端点未被 r2 调用覆盖的(客户域=/api/customer /api/rule/customer /api/preference + ping)
cust_dom = {e for e in endpoints if e[1].startswith(('/api/customer', '/api/rule/customer', '/api/preference'))}
not_called = sorted(cust_dom - calls)
print(f'\n客户域基准 {len(cust_dom)} 端点,r2 调用 {len(calls & cust_dom)},未覆盖 {len(not_called)}')
for m, p in not_called:
print(f' - {m} {p}')
json.dump({'endpoints_base': sorted(f'{m} {p}' for m, p in endpoints),
'r2_calls': sorted(f'{m} {p}' for m, p in calls),
'unknown': [f'{m} {p}' for m, p in real_unknown],
'not_called_customer_domain': [f'{m} {p}' for m, p in not_called],
'per_script': {k: [f'{m} {p}' for m, p in v] for k, v in per_script.items()}},
open(r'e:\code\crm-backend-matt\.scratch\customer-e2e-r2\calls-r2.json', 'w', encoding='utf-8'),
ensure_ascii=False, indent=1)
print('\n✔ 落盘 calls-r2.json')
print('静态验收结论:' + ('PASS(零未命中)' if not real_unknown else f'FAIL({len(real_unknown)} 未命中)'))