641 lines
28 KiB
Python
641 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
"""FULL55 最终真机验收脚本。
|
||
|
||
这个脚本只负责按最终证据规则分批验收，不修改路由、Hook 或 Agent。
|
||
|
||
安全默认值：
|
||
* 默认只读模式；只读模块每个连续调用两次。
|
||
* dry-run 模式只带 dry_run=true、confirm=false、retry=0。
|
||
* write 模式必须同时提供 --confirm-writes 和独立白名单夹具。
|
||
* 所有请求都不自动重试，所有回执都保存 trace、通道、原始回执和业务回读摘要。
|
||
|
||
真实设备调用由本脚本的 HTTP 传输层承担；本文件本身不在导入或测试时发起请求。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import copy
|
||
import csv
|
||
import hashlib
|
||
import html
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Iterable, Mapping, Protocol
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
if str(SCRIPT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(SCRIPT_DIR))
|
||
|
||
DEFAULT_BASE_URL = "http://open.quwanzhi.com:8899"
|
||
DEFAULT_DEVICE_ID = "3c2d803e58f2c30a744234484c4e393e"
|
||
DEFAULT_AUDIT_CSV = (
|
||
ROOT
|
||
/ "开发文档/10、项目管理/02-测试报告/20260810_FULL55_REPAIR/23_最终证据审计/证据缺口清单.csv"
|
||
)
|
||
DEFAULT_EVIDENCE_DIR = (
|
||
ROOT / "开发文档/10、项目管理/02-测试报告/20260810_FULL55_REPAIR/26_最终验收脚本"
|
||
)
|
||
|
||
READONLY_CONCLUSION = "需补只读"
|
||
DRY_RUN_CONCLUSION = "需补干跑"
|
||
WRITE_CONCLUSION = "需真实写验收"
|
||
VALID_CONCLUSIONS = {READONLY_CONCLUSION, DRY_RUN_CONCLUSION, WRITE_CONCLUSION}
|
||
|
||
|
||
class AcceptanceConfigError(ValueError):
|
||
"""验收参数或夹具不符合安全门禁。"""
|
||
|
||
|
||
class Transport(Protocol):
|
||
def request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
payload: Mapping[str, Any] | None,
|
||
*,
|
||
trace_id: str,
|
||
) -> dict[str, Any]:
|
||
"""返回 status、body、elapsed_ms；不得在此层自动重试。"""
|
||
|
||
|
||
def _load_module_specs() -> dict[int, dict[str, Any]]:
|
||
"""复用已有只读/干跑调用清单，避免复制另一套接口路径。"""
|
||
try:
|
||
from full55_device_qa import MODULES # type: ignore
|
||
except ImportError as exc: # pragma: no cover - 仅用于损坏安装的清晰报错
|
||
raise AcceptanceConfigError(f"找不到现有FULL55调用清单: {exc}") from exc
|
||
# 第52项验收不能把标准文档页当作带证据回执的业务接口；
|
||
# 标准 /openapi.json 保持原样，验收脚本改走专用只读回执入口。
|
||
specs = copy.deepcopy(MODULES)
|
||
specs[52] = {
|
||
**specs[52],
|
||
"calls": [
|
||
("GET", "/api/v3/integration/openapi-evidence", "readonly", {}),
|
||
],
|
||
}
|
||
return specs
|
||
|
||
|
||
def load_audit_rows(path: Path = DEFAULT_AUDIT_CSV) -> list[dict[str, str]]:
|
||
"""读取审计清单，并检查编号必须连续覆盖1到55。"""
|
||
with path.open(encoding="utf-8-sig", newline="") as handle:
|
||
rows = list(csv.DictReader(handle))
|
||
if len(rows) != 55:
|
||
raise AcceptanceConfigError(f"审计清单应有55项，实际{len(rows)}项")
|
||
numbers = [int(row["编号"]) for row in rows]
|
||
if numbers != list(range(1, 56)):
|
||
raise AcceptanceConfigError("审计清单编号不是连续的1到55")
|
||
unknown = {row["审计结论"] for row in rows} - VALID_CONCLUSIONS
|
||
if unknown:
|
||
raise AcceptanceConfigError(f"审计清单出现未知结论: {sorted(unknown)}")
|
||
return rows
|
||
|
||
|
||
def audit_index(rows: Iterable[Mapping[str, str]]) -> dict[int, dict[str, str]]:
|
||
return {int(row["编号"]): dict(row) for row in rows}
|
||
|
||
|
||
def selected_ids(value: str | None, *, allowed: set[int]) -> list[int]:
|
||
"""解析 1,2,5-8；不传时返回允许的全部编号。"""
|
||
if not value or value.strip().lower() == "all":
|
||
return sorted(allowed)
|
||
result: set[int] = set()
|
||
for part in value.split(","):
|
||
token = part.strip()
|
||
if not token:
|
||
continue
|
||
if "-" in token:
|
||
start_text, end_text = token.split("-", 1)
|
||
start, end = int(start_text), int(end_text)
|
||
if start > end:
|
||
raise AcceptanceConfigError(f"模块范围无效: {token}")
|
||
result.update(range(start, end + 1))
|
||
else:
|
||
result.add(int(token))
|
||
unknown = result - allowed
|
||
if unknown:
|
||
raise AcceptanceConfigError(f"所选模块不属于当前模式: {sorted(unknown)}")
|
||
return sorted(result)
|
||
|
||
|
||
def redact(value: Any) -> Any:
|
||
"""保存证据前脱敏，避免把手机号、令牌和密钥写入报告。"""
|
||
if isinstance(value, dict):
|
||
result: dict[str, Any] = {}
|
||
for key, item in value.items():
|
||
if re.search(r"password|token|secret|api[_-]?key|phone|mobile|authorization|base64|image_data|screenshot_data", str(key), re.I):
|
||
result[key] = "[已脱敏]"
|
||
else:
|
||
result[key] = redact(item)
|
||
return result
|
||
if isinstance(value, list):
|
||
return [redact(item) for item in value]
|
||
return value
|
||
|
||
|
||
def _nested_dict(value: Any, *keys: str) -> dict[str, Any]:
|
||
for key in keys:
|
||
if isinstance(value, dict) and isinstance(value.get(key), dict):
|
||
return value[key]
|
||
return {}
|
||
|
||
|
||
def extract_evidence(body: Any) -> dict[str, Any]:
|
||
"""从不同层级回执中统一提取最终验收所需的七类证据。"""
|
||
if not isinstance(body, dict):
|
||
return {
|
||
"trace_id": None,
|
||
"channel_used": None,
|
||
"raw_rpc_receipt": None,
|
||
"readback": {},
|
||
"readback_verified": False,
|
||
"write_performed": None,
|
||
}
|
||
data = body.get("data") if isinstance(body.get("data"), dict) else {}
|
||
raw = (
|
||
body.get("raw_rpc_receipt")
|
||
or body.get("raw_receipt")
|
||
or body.get("raw")
|
||
or data.get("raw_rpc_receipt")
|
||
or data.get("raw_receipt")
|
||
or data.get("raw")
|
||
)
|
||
readback = (
|
||
body.get("readback")
|
||
if isinstance(body.get("readback"), dict)
|
||
else data.get("readback")
|
||
if isinstance(data.get("readback"), dict)
|
||
else {}
|
||
)
|
||
trace = body.get("trace_id") or body.get("trace") or data.get("trace_id") or data.get("trace")
|
||
channel = (
|
||
body.get("channel_used")
|
||
or body.get("channel")
|
||
or data.get("channel_used")
|
||
or data.get("channel")
|
||
)
|
||
verified = (
|
||
readback.get("verified") is True
|
||
or readback.get("verified_no_write") is True
|
||
or body.get("readback_verified") is True
|
||
or body.get("verified") is True
|
||
or data.get("verified") is True
|
||
)
|
||
write_performed = body.get("write_performed", data.get("write_performed"))
|
||
return {
|
||
"trace_id": trace,
|
||
"channel_used": channel,
|
||
"raw_rpc_receipt": raw,
|
||
"readback": readback,
|
||
"readback_verified": verified,
|
||
"write_performed": write_performed,
|
||
}
|
||
|
||
|
||
def evidence_complete(result: Mapping[str, Any], *, require_write: bool = False) -> bool:
|
||
"""判断单次结果是否闭合；HTTP200之外不提升为成功。"""
|
||
ev = extract_evidence(result.get("body", result))
|
||
if result.get("http_status") != 200:
|
||
return False
|
||
if not ev["trace_id"] or not ev["channel_used"] or not ev["raw_rpc_receipt"]:
|
||
return False
|
||
if not ev["readback"] or not ev["readback_verified"]:
|
||
return False
|
||
if require_write and ev["write_performed"] is not True:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _make_controls(payload: Mapping[str, Any], *, mode: str, trace_id: str, key: str) -> dict[str, Any]:
|
||
body = dict(payload)
|
||
if mode == "dry-run":
|
||
body.update({"dry_run": True, "confirm": False, "retry": 0, "idempotency_key": key, "trace_id": trace_id})
|
||
elif mode == "write":
|
||
body.update({"dry_run": False, "confirm": True, "retry": 0, "idempotency_key": key, "trace_id": trace_id})
|
||
elif mode == "confirm-false":
|
||
body.update({"dry_run": False, "confirm": False, "retry": 0, "idempotency_key": key, "trace_id": trace_id})
|
||
else:
|
||
body.setdefault("trace_id", trace_id)
|
||
return body
|
||
|
||
|
||
class HttpTransport:
|
||
"""直连HTTP传输；关闭代理、关闭自动重试，方便追溯每一次请求。"""
|
||
|
||
def __init__(self, base_url: str, *, timeout: float = 25.0):
|
||
self.base_url = base_url.rstrip("/")
|
||
self.timeout = timeout
|
||
self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
|
||
def request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
payload: Mapping[str, Any] | None,
|
||
*,
|
||
trace_id: str,
|
||
) -> dict[str, Any]:
|
||
method = method.upper()
|
||
query: dict[str, Any] = {}
|
||
body: bytes | None = None
|
||
if method == "GET":
|
||
query = dict(payload or {})
|
||
else:
|
||
body = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8")
|
||
url = self.base_url + path
|
||
if query:
|
||
url += "?" + urllib.parse.urlencode(query, doseq=True)
|
||
request = urllib.request.Request(url, method=method, data=body, headers={"Accept": "application/json", "X-FULL55-Trace": trace_id})
|
||
if body is not None:
|
||
request.add_header("Content-Type", "application/json")
|
||
started = time.monotonic()
|
||
try:
|
||
with self.opener.open(request, timeout=self.timeout) as response:
|
||
raw = response.read()
|
||
status = response.status
|
||
except urllib.error.HTTPError as exc:
|
||
raw = exc.read()
|
||
status = exc.code
|
||
except Exception as exc: # 网络错误也保存为可追溯结果
|
||
return {"http_status": None, "body": {"error": f"{type(exc).__name__}: {exc}"}, "elapsed_ms": round((time.monotonic() - started) * 1000, 1)}
|
||
try:
|
||
parsed: Any = json.loads(raw.decode("utf-8", "replace"))
|
||
except Exception:
|
||
parsed = {"_text": raw.decode("utf-8", "replace")[:4000]}
|
||
return {"http_status": status, "body": parsed, "elapsed_ms": round((time.monotonic() - started) * 1000, 1)}
|
||
|
||
|
||
def _path_for(spec: Any, device_id: str, group_id: str | None = None) -> str:
|
||
return str(spec).replace("{device_id}", device_id).replace("{group_id}", group_id or "FULL55_NO_GROUP_FIXTURE")
|
||
|
||
|
||
def _resolve_placeholders(value: Any, device_id: str, group_id: str | None = None) -> Any:
|
||
"""递归替换调用清单里的设备/群占位符，避免把花括号原样发到正式环境。"""
|
||
if isinstance(value, dict):
|
||
return {key: _resolve_placeholders(item, device_id, group_id) for key, item in value.items()}
|
||
if isinstance(value, list):
|
||
return [_resolve_placeholders(item, device_id, group_id) for item in value]
|
||
if isinstance(value, tuple):
|
||
return tuple(_resolve_placeholders(item, device_id, group_id) for item in value)
|
||
if isinstance(value, str):
|
||
return _path_for(value, device_id, group_id)
|
||
return value
|
||
|
||
|
||
def _write_json(path: Path, value: Any) -> None:
|
||
path.write_text(json.dumps(redact(value), ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||
|
||
|
||
def _write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
|
||
fields = [
|
||
"module_id", "module", "mode", "round", "stage", "method", "path", "http_status",
|
||
"trace_id", "channel_used", "raw_present", "readback_verified", "write_performed", "passed", "reason",
|
||
]
|
||
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
||
writer = csv.DictWriter(handle, fieldnames=fields)
|
||
writer.writeheader()
|
||
for row in rows:
|
||
writer.writerow({field: row.get(field, "") for field in fields})
|
||
|
||
|
||
def _write_html(path: Path, summary: Mapping[str, Any], rows: list[dict[str, Any]]) -> None:
|
||
table = []
|
||
for row in rows:
|
||
table.append(
|
||
"<tr>"
|
||
+ "".join(f"<td>{html.escape(str(row.get(key, '')))}</td>" for key in ("module_id", "module", "mode", "round", "stage", "http_status", "passed", "reason"))
|
||
+ "</tr>"
|
||
)
|
||
headings = "".join(f"<th>{html.escape(k)}</th>" for k in ("module_id", "module", "mode", "round", "stage", "http_status", "passed", "reason"))
|
||
body = "\n".join(table)
|
||
path.write_text(
|
||
"<!doctype html><meta charset='utf-8'><title>FULL55最终验收</title>"
|
||
"<style>body{font-family:system-ui, sans-serif;margin:24px}table{border-collapse:collapse;width:100%}th,td{border:1px solid #ddd;padding:6px;text-align:left}th{background:#f4f4f4}.ok{color:green}</style>"
|
||
f"<h1>FULL55最终验收汇总</h1><pre>{html.escape(json.dumps(summary, ensure_ascii=False, indent=2))}</pre>"
|
||
f"<table><thead><tr>{headings}</tr></thead><tbody>{body}</tbody></table>",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def _find_image_bytes(value: Any) -> bytes | None:
|
||
"""从截图回执中找PNG/JPEG内容；找不到时保留结构化等价证据。"""
|
||
if isinstance(value, dict):
|
||
for key, item in value.items():
|
||
if isinstance(item, str) and re.search(r"png|image|screenshot|base64", str(key), re.I):
|
||
text = item.split(",", 1)[1] if item.startswith("data:image/") and "," in item else item
|
||
try:
|
||
decoded = base64.b64decode(text, validate=False)
|
||
except Exception:
|
||
decoded = b""
|
||
if decoded.startswith(b"\x89PNG\r\n\x1a\n") or decoded.startswith(b"\xff\xd8\xff"):
|
||
return decoded
|
||
found = _find_image_bytes(item)
|
||
if found:
|
||
return found
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
found = _find_image_bytes(item)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
|
||
def _save_screenshot(evidence_dir: Path, module_id: int, body: Any) -> dict[str, Any]:
|
||
image = _find_image_bytes(body)
|
||
if not image:
|
||
return {"screenshot_present": False, "screenshot_equivalent": evidence_complete({"http_status": 200, "body": body})}
|
||
path = evidence_dir / f"screenshot_{module_id:02d}.png"
|
||
path.write_bytes(image)
|
||
return {"screenshot_present": True, "screenshot_path": str(path), "bytes": len(image), "sha256": hashlib.sha256(image).hexdigest()}
|
||
|
||
|
||
def _parse_fixture(path: Path, module_ids: list[int]) -> dict[int, dict[str, Any]]:
|
||
try:
|
||
fixture = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception as exc:
|
||
raise AcceptanceConfigError(f"夹具不是有效JSON: {exc}") from exc
|
||
if fixture.get("independent_allowlist") is not True:
|
||
raise AcceptanceConfigError("夹具必须明确 independent_allowlist=true")
|
||
if not fixture.get("fixture_id"):
|
||
raise AcceptanceConfigError("夹具缺少fixture_id")
|
||
modules = fixture.get("modules")
|
||
if not isinstance(modules, dict):
|
||
raise AcceptanceConfigError("夹具缺少modules对象")
|
||
selected: dict[int, dict[str, Any]] = {}
|
||
keys: list[str] = []
|
||
for module_id in module_ids:
|
||
item = modules.get(str(module_id))
|
||
if not isinstance(item, dict):
|
||
raise AcceptanceConfigError(f"夹具没有模块{module_id}的独立配置")
|
||
if not item.get("operations"):
|
||
raise AcceptanceConfigError(f"模块{module_id}没有operations")
|
||
if not isinstance(item.get("targets"), list) or not item.get("targets"):
|
||
raise AcceptanceConfigError(f"模块{module_id}没有targets白名单")
|
||
if not item.get("readback"):
|
||
raise AcceptanceConfigError(f"模块{module_id}没有业务回读定义")
|
||
for operation in item["operations"]:
|
||
key = operation.get("idempotency_key")
|
||
if not key:
|
||
raise AcceptanceConfigError(f"模块{module_id}存在空幂等键")
|
||
keys.append(str(key))
|
||
selected_item = dict(item)
|
||
selected_item["_fixture_id"] = fixture["fixture_id"]
|
||
selected[module_id] = selected_item
|
||
if len(keys) != len(set(keys)):
|
||
raise AcceptanceConfigError("夹具内幂等键重复")
|
||
return selected
|
||
|
||
|
||
def _module_calls(specs: Mapping[int, Mapping[str, Any]], module_id: int, mode: str) -> list[tuple[str, str, str, dict[str, Any]]]:
|
||
calls = list(specs[module_id].get("calls", []))
|
||
if mode == "readonly":
|
||
calls = [call for call in calls if call[2] == "readonly"]
|
||
elif mode == "dry-run":
|
||
calls = [call for call in calls if call[2] == "dry_run"]
|
||
if not calls:
|
||
raise AcceptanceConfigError(f"模块{module_id}在{mode}模式没有可执行调用")
|
||
return calls
|
||
|
||
|
||
class AcceptanceRunner:
|
||
def __init__(
|
||
self,
|
||
*,
|
||
transport: Transport,
|
||
audit_rows: list[dict[str, str]],
|
||
specs: Mapping[int, Mapping[str, Any]],
|
||
device_id: str,
|
||
evidence_dir: Path,
|
||
mode: str,
|
||
module_ids: list[int],
|
||
confirm_writes: bool = False,
|
||
fixture_modules: dict[int, dict[str, Any]] | None = None,
|
||
):
|
||
self.transport = transport
|
||
self.audit = audit_index(audit_rows)
|
||
self.specs = specs
|
||
self.device_id = device_id
|
||
self.evidence_dir = evidence_dir
|
||
self.mode = mode
|
||
self.module_ids = module_ids
|
||
self.confirm_writes = confirm_writes
|
||
self.fixture_modules = fixture_modules or {}
|
||
self.rows: list[dict[str, Any]] = []
|
||
self.last_body: Any = None
|
||
self.started_at = time.time()
|
||
|
||
def _call(
|
||
self,
|
||
module_id: int,
|
||
module_name: str,
|
||
stage: str,
|
||
round_number: int,
|
||
method: str,
|
||
path: str,
|
||
payload: Mapping[str, Any] | None,
|
||
*,
|
||
request_mode: str,
|
||
key: str | None = None,
|
||
require_write: bool = False,
|
||
) -> dict[str, Any]:
|
||
trace = f"full55-final-{module_id:02d}-{stage}-{round_number}-{uuid.uuid4().hex[:12]}"
|
||
body = _resolve_placeholders(dict(payload or {}), self.device_id)
|
||
if request_mode in {"dry-run", "write", "confirm-false"}:
|
||
body = _make_controls(body, mode=request_mode, trace_id=trace, key=key or f"full55-{uuid.uuid4().hex}")
|
||
elif request_mode == "readonly":
|
||
body.setdefault("device_id", self.device_id)
|
||
body.setdefault("trace_id", trace)
|
||
resolved_path = _path_for(path, self.device_id)
|
||
result = self.transport.request(method, resolved_path, body, trace_id=trace)
|
||
self.last_body = result.get("body")
|
||
ev = extract_evidence(result.get("body"))
|
||
passed = evidence_complete(result, require_write=require_write)
|
||
# confirm=false 的拦截本来就应返回4xx；没有发生写入即可视为门禁通过。
|
||
if stage.startswith("confirm_gate"):
|
||
passed = result.get("http_status") in {400, 409, 422} and ev.get("write_performed") is not True
|
||
row = {
|
||
"module_id": module_id,
|
||
"module": module_name,
|
||
"mode": self.mode,
|
||
"round": round_number,
|
||
"stage": stage,
|
||
"method": method,
|
||
"path": resolved_path,
|
||
"http_status": result.get("http_status"),
|
||
"trace_id": ev.get("trace_id") or trace,
|
||
"channel_used": ev.get("channel_used"),
|
||
"raw_present": bool(ev.get("raw_rpc_receipt")),
|
||
"readback_verified": bool(ev.get("readback_verified")),
|
||
"write_performed": ev.get("write_performed"),
|
||
"passed": passed,
|
||
"reason": "证据齐全" if passed else "缺HTTP/trace/channel/raw/readback/verified中的一项或多项",
|
||
"elapsed_ms": result.get("elapsed_ms"),
|
||
"response": redact(result.get("body")),
|
||
}
|
||
self.rows.append(row)
|
||
_write_json(self.evidence_dir / f"module_{module_id:02d}_{stage}_{round_number}.json", row)
|
||
return row
|
||
|
||
def _run_readonly_or_dry_run(self, module_id: int) -> None:
|
||
item = self.audit[module_id]
|
||
module_name = item["模块"]
|
||
request_mode = "readonly" if self.mode == "readonly" else "dry-run"
|
||
for round_number in (1, 2):
|
||
for call_index, (method, path, _call_mode, payload) in enumerate(_module_calls(self.specs, module_id, self.mode), start=1):
|
||
self._call(
|
||
module_id,
|
||
module_name,
|
||
f"{request_mode}_{call_index}",
|
||
round_number,
|
||
method,
|
||
path,
|
||
payload,
|
||
request_mode=request_mode,
|
||
key=f"full55-{request_mode}-{module_id}-{call_index}-{round_number}-{uuid.uuid4().hex}",
|
||
require_write=False,
|
||
)
|
||
|
||
def _run_write_module(self, module_id: int) -> None:
|
||
item = self.audit[module_id]
|
||
module_name = item["模块"]
|
||
fixture = self.fixture_modules[module_id]
|
||
for op_index, operation in enumerate(fixture["operations"], start=1):
|
||
method = str(operation.get("method", "POST")).upper()
|
||
path = str(operation["path"])
|
||
payload = dict(operation.get("payload", {}))
|
||
key = str(operation.get("idempotency_key") or f"full55-write-{module_id}-{op_index}-{uuid.uuid4().hex}")
|
||
self._call(module_id, module_name, f"dry_run_{op_index}", op_index, method, path, payload, request_mode="dry-run", key=key, require_write=False)
|
||
self._call(module_id, module_name, f"confirm_gate_{op_index}", op_index, method, path, payload, request_mode="confirm-false", key=f"{key}:gate", require_write=False)
|
||
self._call(module_id, module_name, f"confirm_once_{op_index}", op_index, method, path, payload, request_mode="write", key=key, require_write=True)
|
||
replay = self._call(module_id, module_name, f"idempotency_replay_{op_index}", op_index, method, path, payload, request_mode="write", key=key, require_write=True)
|
||
replay["replay_required"] = True
|
||
replay["replay_semantics"] = "必须是同键同参复用，不得产生第二次业务写"
|
||
|
||
readback = fixture["readback"]
|
||
self._call(
|
||
module_id,
|
||
module_name,
|
||
f"business_readback_{op_index}",
|
||
op_index,
|
||
str(readback.get("method", "GET")).upper(),
|
||
str(readback["path"]),
|
||
dict(readback.get("payload", {})),
|
||
request_mode="readonly",
|
||
require_write=False,
|
||
)
|
||
# 一个写模块一张截图；截图失败会被记录，不会自动重试。
|
||
screenshot = fixture.get("screenshot", {"method": "POST", "path": f"/api/v3/devices/{self.device_id}/screenshot", "payload": {}})
|
||
shot = self._call(module_id, module_name, "screenshot", 1, str(screenshot.get("method", "POST")), str(screenshot["path"]), dict(screenshot.get("payload", {})), request_mode="readonly")
|
||
shot.update(_save_screenshot(self.evidence_dir, module_id, self.last_body))
|
||
shot["screenshot_required"] = True
|
||
_write_json(self.evidence_dir / f"module_{module_id:02d}_screenshot_1.json", shot)
|
||
|
||
def run(self) -> dict[str, Any]:
|
||
self.evidence_dir.mkdir(parents=True, exist_ok=True)
|
||
for module_id in self.module_ids:
|
||
if self.mode == "write":
|
||
self._run_write_module(module_id)
|
||
else:
|
||
self._run_readonly_or_dry_run(module_id)
|
||
summary = {
|
||
"script": "full55_final_device_acceptance",
|
||
"mode": self.mode,
|
||
"device_calls_allowed_by_script": True,
|
||
"real_write_allowed": self.mode == "write" and self.confirm_writes,
|
||
"retry": 0,
|
||
"selected_modules": self.module_ids,
|
||
"selected_count": len(self.module_ids),
|
||
"evidence_rule": {"readonly_rounds": 2, "dry_run_confirm": False, "write_confirm_count": 1, "idempotency_replay_count": 1},
|
||
"fixture_id": self.fixture_modules and sorted({str(v.get("_fixture_id", "")) for v in self.fixture_modules.values()}) or None,
|
||
"started_at": self.started_at,
|
||
"finished_at": time.time(),
|
||
"record_count": len(self.rows),
|
||
"passed_records": sum(bool(row["passed"]) for row in self.rows),
|
||
"failed_records": sum(not bool(row["passed"]) for row in self.rows),
|
||
}
|
||
_write_json(self.evidence_dir / "full55_final_acceptance.json", {"summary": summary, "records": self.rows})
|
||
_write_csv(self.evidence_dir / "full55_final_acceptance.csv", self.rows)
|
||
_write_html(self.evidence_dir / "full55_final_acceptance.html", summary, self.rows)
|
||
return {"summary": summary, "records": self.rows}
|
||
|
||
|
||
def _validate_selection(mode: str, module_ids: list[int], rows: list[dict[str, str]], confirm_writes: bool, fixture_file: Path | None) -> dict[int, dict[str, Any]] | None:
|
||
index = audit_index(rows)
|
||
expected = {
|
||
"readonly": {n for n, row in index.items() if row["审计结论"] == READONLY_CONCLUSION},
|
||
"dry-run": {n for n, row in index.items() if row["审计结论"] in {DRY_RUN_CONCLUSION, WRITE_CONCLUSION}},
|
||
"write": {n for n, row in index.items() if row["审计结论"] == WRITE_CONCLUSION},
|
||
}[mode]
|
||
if not set(module_ids) <= expected:
|
||
raise AcceptanceConfigError(f"{mode}模式允许模块{sorted(expected)}，实际选择{module_ids}")
|
||
if mode != "write" and confirm_writes:
|
||
raise AcceptanceConfigError("只有write模式可以使用--confirm-writes")
|
||
if mode == "write":
|
||
if not confirm_writes:
|
||
raise AcceptanceConfigError("write模式必须显式提供--confirm-writes")
|
||
if fixture_file is None:
|
||
raise AcceptanceConfigError("write模式必须提供--fixture-file")
|
||
return _parse_fixture(fixture_file, module_ids)
|
||
return None
|
||
|
||
|
||
def run_acceptance(
|
||
*,
|
||
mode: str = "readonly",
|
||
modules: str | None = None,
|
||
evidence_dir: Path = DEFAULT_EVIDENCE_DIR,
|
||
base_url: str = DEFAULT_BASE_URL,
|
||
device_id: str = DEFAULT_DEVICE_ID,
|
||
audit_csv: Path = DEFAULT_AUDIT_CSV,
|
||
confirm_writes: bool = False,
|
||
fixture_file: Path | None = None,
|
||
transport: Transport | None = None,
|
||
) -> dict[str, Any]:
|
||
if mode not in {"readonly", "dry-run", "write"}:
|
||
raise AcceptanceConfigError(f"不支持模式: {mode}")
|
||
rows = load_audit_rows(audit_csv)
|
||
index = audit_index(rows)
|
||
allowed = {n for n, row in index.items() if row["审计结论"] == READONLY_CONCLUSION} if mode == "readonly" else ({n for n, row in index.items() if row["审计结论"] in {DRY_RUN_CONCLUSION, WRITE_CONCLUSION}} if mode == "dry-run" else {n for n, row in index.items() if row["审计结论"] == WRITE_CONCLUSION})
|
||
module_ids = selected_ids(modules, allowed=allowed)
|
||
fixture_modules = _validate_selection(mode, module_ids, rows, confirm_writes, fixture_file)
|
||
runner = AcceptanceRunner(transport=transport or HttpTransport(base_url), audit_rows=rows, specs=_load_module_specs(), device_id=device_id, evidence_dir=evidence_dir, mode=mode, module_ids=module_ids, confirm_writes=confirm_writes, fixture_modules=fixture_modules)
|
||
return runner.run()
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="FULL55最终真机验收：默认只读，写入需要双重开关")
|
||
parser.add_argument("--mode", choices=("readonly", "dry-run", "write"), default="readonly", help="readonly默认连续两次；dry-run只零写；write受夹具和显式开关保护")
|
||
parser.add_argument("--modules", help="模块编号，如1,2,5-8；不填则选择当前模式全部模块")
|
||
parser.add_argument("--evidence-dir", type=Path, default=DEFAULT_EVIDENCE_DIR, help="证据输出目录")
|
||
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="SDK地址")
|
||
parser.add_argument("--device-id", default=DEFAULT_DEVICE_ID, help="设备编号")
|
||
parser.add_argument("--audit-csv", type=Path, default=DEFAULT_AUDIT_CSV, help="最终证据缺口清单")
|
||
parser.add_argument("--confirm-writes", action="store_true", help="write模式的第二道开关")
|
||
parser.add_argument("--fixture-file", type=Path, help="write模式的独立白名单夹具JSON")
|
||
return parser
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
args = build_parser().parse_args(argv)
|
||
try:
|
||
result = run_acceptance(mode=args.mode, modules=args.modules, evidence_dir=args.evidence_dir, base_url=args.base_url, device_id=args.device_id, audit_csv=args.audit_csv, confirm_writes=args.confirm_writes, fixture_file=args.fixture_file)
|
||
except AcceptanceConfigError as exc:
|
||
print(f"参数/夹具门禁失败：{exc}", file=sys.stderr)
|
||
return 2
|
||
print(json.dumps(result["summary"], ensure_ascii=False, indent=2))
|
||
return 0 if result["summary"]["failed_records"] == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
raise SystemExit(main())
|