345 lines
15 KiB
Python
345 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""E2E 契约层负向探测脚本（S-E2E 任务 00c8cc0d）。
|
||
|
||
对接口管理规范 V4.5 的 15 个在用端点发起真实请求，两组负向用例：
|
||
A 组：缺鉴权三要素头（SecretKey/SystemCode/Token 全缺）；
|
||
B 组：带真实 SecretKey/SystemCode（运行时读 docs/SystemCode_secertkey.txt，不落明文）、缺 Token。
|
||
|
||
判定口径（任务书 + core/network/ApiCaller.kt 实证契约）：
|
||
存在性 ：HTTP 状态非 404 且非连接错误（000）；
|
||
双层结构 ：HTTP 500 且 body 为 JSON、含字符串字段 ExceptionMessage，
|
||
其值可再解析为含 code 字段的信封 JSON（ASP.NET 异常包装内嵌业务信封）；
|
||
信封解包模拟：内嵌信封 code 必为字符串且取值 ∈ {"200","500","429","403"}（对齐 ApiEnvelope）。
|
||
|
||
节流：请求间隔 ≥1.5s；不主动触发 403（连续报错锁定）；429 若出现如实记录。
|
||
|
||
用法：python3 docs/e2e-evidence/contract_probe.py
|
||
产出：docs/e2e-evidence/responses/{A|B}_{章节}.txt 原始响应
|
||
docs/e2e-evidence/contract_results.json 结构化结果
|
||
docs/e2e-evidence/contract_evidence.md 证据汇总表
|
||
"""
|
||
import json
|
||
import time
|
||
import uuid
|
||
import urllib.request
|
||
import urllib.error
|
||
import ssl
|
||
from pathlib import Path
|
||
|
||
BASE_URL = "https://jcd.stec.p-q.co"
|
||
WORKSPACE = Path(__file__).resolve().parents[2]
|
||
EVIDENCE_DIR = WORKSPACE / "docs" / "e2e-evidence"
|
||
RESPONSE_DIR = EVIDENCE_DIR / "responses"
|
||
KEY_FILE = WORKSPACE / "docs" / "SystemCode_secertkey.txt"
|
||
|
||
REQUEST_INTERVAL_SEC = 1.5
|
||
TIMEOUT_SEC = 25
|
||
VALID_ENVELOPE_CODES = {"200", "500", "429", "403"}
|
||
|
||
# 登录/验证码类端点按接口文档 1.2.3 天然不携带 Token。
|
||
# 探测用参数均为无效占位（10000000000 非法号段 / e2e-probe-* 假 ID / e2e 弱凭据），
|
||
# 不触达任何真实业务数据，也不触发真实短信下发。
|
||
ENDPOINTS = [
|
||
{
|
||
"sec": "2.2.1", "name": "账号密码登录", "method": "POST",
|
||
"path": "/OutWebApi/api/LoginByAccount",
|
||
"body_json": {"LoginName": "e2e_probe_placeholder", "Password": "E2eProbe#2026"},
|
||
},
|
||
{
|
||
"sec": "2.2.2", "name": "发送验证码", "method": "GET",
|
||
"path": "/OutWebApi/api/LoginSecurityCode",
|
||
"query": {"MobilePhone": "10000000000"},
|
||
},
|
||
{
|
||
"sec": "2.2.3", "name": "手机验证码登录", "method": "POST",
|
||
"path": "/OutWebApi/api/LoginByMobile",
|
||
"body_json": {"Mobile": "10000000000", "VerificationCode": "000000"},
|
||
},
|
||
{
|
||
"sec": "2.3.1", "name": "修改登录密码", "method": "POST",
|
||
"path": "/OutWebApi/api/ModifyPassword",
|
||
"body_json": {"Password": "E2eProbe#2026", "ConfirmAnswer": "e2e-probe"},
|
||
},
|
||
{"sec": "2.4", "name": "获取当前用户信息", "method": "GET", "path": "/OutWebApi/api/GetUserInfo"},
|
||
{"sec": "2.5.1.1", "name": "获取当前用户项目列表", "method": "GET", "path": "/OutWebApi/api/GetProjectList"},
|
||
{
|
||
"sec": "2.6", "name": "获取APP端任务列表", "method": "GET",
|
||
"path": "/OutWebApi/api/GetUserPlanGroup",
|
||
"query": {"PlanDate": "2026-09-04"},
|
||
},
|
||
{"sec": "2.7", "name": "获取项目监测计划任务列表", "method": "GET", "path": "/OutWebApi/api/GetPlanGroup"},
|
||
{
|
||
"sec": "2.9", "name": "获取测组包含的监测点列表", "method": "GET",
|
||
"path": "/OutWebApi/api/GetPointBySurveyGroup",
|
||
"query": {"GroupID": "e2e-probe-group"},
|
||
},
|
||
{
|
||
"sec": "2.10", "name": "获取项目监测计划统计表", "method": "GET",
|
||
"path": "/OutWebApi/api/GetPlanStatistical",
|
||
"query": {"ProjectID": "e2e-probe-project"},
|
||
},
|
||
{
|
||
"sec": "2.11", "name": "获取指定监测计划的测组统计表", "method": "GET",
|
||
"path": "/OutWebApi/api/GetSurveyGroupList",
|
||
"query": {"PlanID": "e2e-probe-plan"},
|
||
},
|
||
{
|
||
"sec": "2.12", "name": "监测点统计表", "method": "GET",
|
||
"path": "/OutWebApi/api/GetPointStatistics",
|
||
"query": {"PlanID": "e2e-probe-plan", "GroupID": "e2e-probe-group"},
|
||
},
|
||
{
|
||
"sec": "2.13", "name": "数据解析（原始文件上传）", "method": "POST_MULTIPART",
|
||
"path": "/OutWebApi/api/AnalysisMonitorData",
|
||
"multipart_fields": {
|
||
"ProjectID": "e2e-probe-project",
|
||
"WorkPointID": "e2e-probe-workpoint",
|
||
"SurveyGroupID": "e2e-probe-group",
|
||
"PlanID": "e2e-probe-plan",
|
||
"MonitorDate": "2026-09-04",
|
||
"WorkInfo": "e2e-probe",
|
||
"Equipment": "e2e-probe",
|
||
"Weather": "晴",
|
||
"PointsData": json.dumps(
|
||
[{"Point": "e2e-probe-point", "PointName": "e2e-probe-point",
|
||
"CurrentValue": "0", "CurrentChangeValue": "0", "IsInitValue": "false"}],
|
||
ensure_ascii=False),
|
||
},
|
||
"multipart_file": ("E2eProbe.txt", "e2e probe placeholder file\n"),
|
||
},
|
||
{
|
||
"sec": "2.14", "name": "获取基准点", "method": "GET",
|
||
"path": "/OutWebApi/api/ControlPoint",
|
||
"query": {"ProjectID": "e2e-probe-project"},
|
||
},
|
||
{"sec": "2.15", "name": "获取工点数据", "method": "GET", "path": "/OutWebApi/api/GetWorkPoint"},
|
||
]
|
||
|
||
# 与 core/network/AuthInterceptor 注入头保持同名。
|
||
HEADER_SECRET_KEY = "SecretKey"
|
||
HEADER_SYSTEM_CODE = "SystemCode"
|
||
HEADER_TOKEN = "Token"
|
||
|
||
|
||
def load_keys():
|
||
"""解析 docs/SystemCode_secertkey.txt（格式：`SystemCode:Mobile` / `SecretKey:<hex>`）。"""
|
||
keys = {}
|
||
for line in KEY_FILE.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if ":" not in line:
|
||
continue
|
||
name, _, value = line.partition(":")
|
||
keys[name.strip()] = value.strip()
|
||
return keys.get("SystemCode", ""), keys.get("SecretKey", "")
|
||
|
||
|
||
def build_request(ep, case, system_code, secret_key):
|
||
"""按端点与用例组构造 urllib 请求；返回 (request, 描述行)。"""
|
||
url = BASE_URL + ep["path"]
|
||
method = ep["method"]
|
||
headers = {"User-Agent": "stec-cmd-e2e-probe/1.0", "Accept": "application/json, text/*;q=0.9"}
|
||
if case == "B":
|
||
headers[HEADER_SYSTEM_CODE] = system_code
|
||
headers[HEADER_SECRET_KEY] = secret_key
|
||
# A/B 两组均不带 Token（负向目标即缺 Token 行为）。
|
||
|
||
data = None
|
||
if method == "GET":
|
||
if ep.get("query"):
|
||
from urllib.parse import urlencode
|
||
url += "?" + urlencode(ep["query"])
|
||
elif method == "POST":
|
||
body = json.dumps(ep["body_json"], ensure_ascii=False).encode("utf-8")
|
||
headers["Content-Type"] = "application/json; charset=utf-8"
|
||
data = body
|
||
elif method == "POST_MULTIPART":
|
||
boundary = "----e2eProbe" + uuid.uuid4().hex[:12]
|
||
parts = []
|
||
for name, value in ep["multipart_fields"].items():
|
||
parts.append(
|
||
f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n".encode("utf-8"))
|
||
filename, content = ep["multipart_file"]
|
||
parts.append(
|
||
(f"--{boundary}\r\nContent-Disposition: form-data; name=\"Files\"; "
|
||
f"filename=\"{filename}\"\r\nContent-Type: text/plain\r\n\r\n").encode("utf-8")
|
||
+ content.encode("utf-8") + b"\r\n")
|
||
parts.append(f"--{boundary}--\r\n".encode("utf-8"))
|
||
headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"
|
||
data = b"".join(parts)
|
||
else:
|
||
raise ValueError(f"未知方法 {method}")
|
||
|
||
req = urllib.request.Request(url, data=data, headers=headers, method="GET" if method == "GET" else "POST")
|
||
auth = {k: headers.get(k) for k in (HEADER_SECRET_KEY, HEADER_SYSTEM_CODE, HEADER_TOKEN)}
|
||
return req, auth
|
||
|
||
|
||
def do_request(req):
|
||
"""发请求并归一返回 (http_status, body_text, err)。连接失败时 status 记 '000'。"""
|
||
ctx = ssl.create_default_context()
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=TIMEOUT_SEC, context=ctx) as resp:
|
||
return str(resp.status), resp.read().decode("utf-8", errors="replace"), None
|
||
except urllib.error.HTTPError as e:
|
||
return str(e.code), e.read().decode("utf-8", errors="replace"), None
|
||
except Exception as e: # 连接失败/超时/TLS
|
||
return "000", "", f"{type(e).__name__}: {e}"
|
||
|
||
|
||
def analyze(ep, case, status, body):
|
||
"""按判定口径分析一个响应，返回结果 dict。"""
|
||
result = {
|
||
"sec": ep["sec"], "name": ep["name"], "case": case,
|
||
"http_status": status,
|
||
"exists": None, "double_layer": None,
|
||
"envelope_code": None, "envelope_code_is_str": None,
|
||
"envelope_msg": None, "envelope_data": None,
|
||
"exception_type": None, "notes": "",
|
||
}
|
||
if status == "000":
|
||
result["exists"] = False
|
||
result["notes"] = "连接失败"
|
||
return result
|
||
result["exists"] = status != "404"
|
||
if status == "404":
|
||
result["notes"] = "端点不存在"
|
||
return result
|
||
|
||
wrapper = None
|
||
if body.strip().startswith("{"):
|
||
try:
|
||
wrapper = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
wrapper = None
|
||
|
||
embedded = None
|
||
if isinstance(wrapper, dict):
|
||
result["exception_type"] = wrapper.get("ExceptionType")
|
||
raw = wrapper.get("ExceptionMessage")
|
||
if isinstance(raw, str):
|
||
try:
|
||
embedded = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
embedded = None
|
||
elif isinstance(raw, dict):
|
||
embedded = raw # 个别端点可能直接内嵌对象
|
||
|
||
if embedded and isinstance(embedded, dict) and "code" in embedded:
|
||
result["double_layer"] = (status == "500")
|
||
code = embedded.get("code")
|
||
result["envelope_code"] = code
|
||
result["envelope_code_is_str"] = isinstance(code, str)
|
||
result["envelope_msg"] = embedded.get("msg")
|
||
result["envelope_data"] = embedded.get("data")
|
||
result["notes"] = (
|
||
"双层结构成立" if result["double_layer"] and result["envelope_code_is_str"]
|
||
else "内嵌信封可解但形态偏离（见 http_status/code 类型）")
|
||
else:
|
||
# 可能是裸业务信封（HTTP 200 + {code,...}）或非 JSON（HTML 错误页等）
|
||
if isinstance(wrapper, dict) and "code" in wrapper:
|
||
result["double_layer"] = False
|
||
code = wrapper.get("code")
|
||
result["envelope_code"] = code
|
||
result["envelope_code_is_str"] = isinstance(code, str)
|
||
result["envelope_msg"] = wrapper.get("msg")
|
||
result["envelope_data"] = wrapper.get("data")
|
||
result["notes"] = "裸业务信封（未走 ASP.NET 异常包装）"
|
||
else:
|
||
result["double_layer"] = False
|
||
result["notes"] = "响应非可解析双层结构（原始体见存档）"
|
||
return result
|
||
|
||
|
||
def save_raw(ep, case, status, auth, body, err):
|
||
stem = f"{case}_{ep['sec'].replace('.', '_')}"
|
||
keys_desc = "无三要素头" if not auth.get(HEADER_SECRET_KEY) else (
|
||
f"{HEADER_SYSTEM_CODE}={auth.get(HEADER_SYSTEM_CODE)}, "
|
||
f"{HEADER_SECRET_KEY}={str(auth.get(HEADER_SECRET_KEY))[:6]}***(脱敏), 无 Token")
|
||
lines = [
|
||
f"# 端点 {ep['sec']} {ep['name']} 用例组 {case}" + ("（缺三要素头）" if case == "A" else "（带密钥缺 Token）"),
|
||
f"# 请求：{ep['method']} {BASE_URL}{ep['path']}",
|
||
f"# 请求头：{keys_desc}",
|
||
f"# HTTP 状态：{status}" + (f" 错误：{err}" if err else ""),
|
||
f"# 存档时间：{time.strftime('%Y-%m-%dT%H:%M:%S%z')}",
|
||
"=" * 60,
|
||
body if body else "(空响应体)",
|
||
]
|
||
(RESPONSE_DIR / f"{stem}.txt").write_text("\n".join(lines), encoding="utf-8")
|
||
return RESPONSE_DIR / f"{stem}.txt"
|
||
|
||
|
||
def write_markdown(results):
|
||
rows = []
|
||
verdict_a = verdict_b = 0
|
||
for r in results:
|
||
ok = (
|
||
r["exists"] is True
|
||
and r["http_status"] == "500"
|
||
and r["double_layer"] is True
|
||
and r["envelope_code_is_str"] is True
|
||
and r["envelope_code"] in VALID_ENVELOPE_CODES
|
||
)
|
||
if r["case"] == "A" and ok:
|
||
verdict_a += 1
|
||
if r["case"] == "B" and ok:
|
||
verdict_b += 1
|
||
code = r["envelope_code"]
|
||
rows.append(
|
||
"| {sec} | {name} | {case} | {st} | {ex} | {dl} | {code} | {isstr} | {msg} | {note} |".format(
|
||
sec=r["sec"], name=r["name"], case=r["case"], st=r["http_status"],
|
||
ex="存在" if r["exists"] else "否",
|
||
dl="成立" if r["double_layer"] else "不成立",
|
||
code="`%s`" % code if code is not None else "—",
|
||
isstr="是" if r["envelope_code_is_str"] else "否",
|
||
msg=(r["envelope_msg"] or "")[:40] if isinstance(r["envelope_msg"], str) else ("" if r["envelope_msg"] is None else str(r["envelope_msg"])[:40]),
|
||
note=r["notes"],
|
||
))
|
||
md = "\n".join([
|
||
"# 契约层负向探测证据表（自动生成于 %s）" % time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"",
|
||
"- 服务器：%s（接口管理规范 V4.5 测试环境）" % BASE_URL,
|
||
"- 用例组 A：缺三要素头（SecretKey/SystemCode/Token 全缺）；用例组 B：带 docs/SystemCode_secertkey.txt 密钥、缺 Token。",
|
||
"- 判定：存在性=HTTP≠404/000；双层结构=HTTP 500 且 ExceptionMessage 内嵌信封 JSON；信封解包=code 为字符串且 ∈ {200,500,429,403}。",
|
||
"- 通过：A 组 %d/15，B 组 %d/15。" % (verdict_a, verdict_b),
|
||
"- 原始响应存档：`docs/e2e-evidence/responses/`（密钥已脱敏）。",
|
||
"",
|
||
"| 章节 | 端点 | 用例 | HTTP | 存在 | 双层结构 | 信封code | code为字符串 | 信封msg | 备注 |",
|
||
"|---|---|---|---|---|---|---|---|---|---|",
|
||
] + rows + [""])
|
||
(EVIDENCE_DIR / "contract_evidence.md").write_text(md, encoding="utf-8")
|
||
return verdict_a, verdict_b
|
||
|
||
|
||
def main():
|
||
RESPONSE_DIR.mkdir(parents=True, exist_ok=True)
|
||
system_code, secret_key = load_keys()
|
||
if not secret_key:
|
||
raise SystemExit("密钥文件解析失败：%s" % KEY_FILE)
|
||
print("密钥加载完成（SystemCode=%s, SecretKey=%s…脱敏）" % (system_code, secret_key[:6]))
|
||
|
||
results = []
|
||
for case in ("A", "B"):
|
||
for ep in ENDPOINTS:
|
||
req, auth = build_request(ep, case, system_code, secret_key)
|
||
status, body, err = do_request(req)
|
||
raw_path = save_raw(ep, case, status, auth, body, err)
|
||
r = analyze(ep, case, status, body)
|
||
if err:
|
||
r["notes"] += f"（{err}）"
|
||
r["raw_file"] = str(raw_path.relative_to(WORKSPACE))
|
||
results.append(r)
|
||
print("[{case}] {sec:<8} {st:<3} 双层={dl} code={code} {name}".format(
|
||
case=case, sec=ep["sec"], st=status,
|
||
dl=r["double_layer"], code=r["envelope_code"], name=ep["name"]))
|
||
time.sleep(REQUEST_INTERVAL_SEC)
|
||
|
||
(EVIDENCE_DIR / "contract_results.json").write_text(
|
||
json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
verdict_a, verdict_b = write_markdown(results)
|
||
print("完成：A 组通过 %d/15，B 组通过 %d/15；证据见 docs/e2e-evidence/" % (verdict_a, verdict_b))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|