"""Ai-DOP 第三方对接模拟器 · 本地控制服务（任务书 P1-B / P1-G）。
- 只绑定 127.0.0.1，页面与 API 仅本机可用；
- Secret/密码/Token 只从环境变量读取，任何 API 不回显；
- 所有对外目标 host 走白名单硬阻断（默认拒绝）；
- 运行记录（RunStore）写入前统一脱敏；
- 内嵌管理 Mock API（providers.mock_api），默认端口 8018，可启停/切鉴权/注入错误；
- 三层报告：代码层 SUPPORTED/UNSUPPORTED、配置层 READY/NOT_*、数据层 NOT_RUN/ACCEPTED/TRANSFORM_PENDING/RUN_FAILED。
启动：python tools/integration-simulator/app.py （详见 README.md）
"""
from __future__ import annotations
import asyncio
import json
import os
import socket
import sys
import threading
import time
import uuid
from pathlib import Path
SIM_ROOT = Path(__file__).resolve().parent
if str(SIM_ROOT) not in sys.path:
sys.path.insert(0, str(SIM_ROOT))
import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from catalog import sample_adapters as sa
from clients import aidop_probe
from clients.inbound_client import InboundClient
from clients.redaction import mask_access_key, redact_obj
from load import controls as load_controls
from load import metrics as load_metrics
from load import runner_p3
from providers import fault_catalog, mock_api, mock_db
DEFAULT_CONTROL_PORT = 8017
MAX_RUNS = 200
# 状态枚举（任务书 §4）
CONFIG_STATES = ("READY", "CONFIG_NOT_REGISTERED", "CONFIG_NOT_ENABLED",
"GRANT_MISSING", "SOURCE_UNREACHABLE", "RUN_FAILED")
# 入站实体授权探测缓存（60s TTL；探测用 GET schema，无副作用）
_AUTH_PROBE_CACHE: dict = {"at": 0.0, "data": None}
_AUTH_PROBE_TTL = 60.0
def _env(name: str) -> str:
return os.environ.get(name, "") or ""
def _secrets() -> list[str]:
return [v for v in (_env("AIDOP_SIM_INBOUND_SECRET"), _env("AIDOP_SIM_MYSQL_PASSWORD"),
_env("AIDOP_SIM_ADMIN_TOKEN")) if v]
def _access_keys() -> list[str]:
"""已知 AccessKey 值：自由文本中出现时只保留后四位（任务书 §24）。"""
ak = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
return [ak] if ak else []
def load_config() -> dict:
cfg = {
"aidopBaseUrl": "http://127.0.0.1:5005",
"mockApiPort": 8018,
"mysqlDatabase": mock_db.SIM_DB_DEFAULT,
"allowedTargetHosts": ["127.0.0.1", "localhost"],
}
cfg_file = SIM_ROOT / "config.json"
if cfg_file.exists():
import json
with cfg_file.open("r", encoding="utf-8") as f:
cfg.update(json.load(f))
return cfg
CONFIG = load_config()
MOCK_STATE = mock_api.MockApiState()
MOCK_APP = mock_api.create_app(MOCK_STATE)
# ---- 第三阶段压力测试全局（默认关闭，重启即清）----
LOAD_GUARD = load_controls.LoadGuard()
LOAD_METRICS = load_metrics.LoadMetrics()
LOAD_RUNNER: runner_p3.P3Runner | None = None
LOAD_REPORTS = None # 在 ReportStore 定义后创建
DB_GEN = {"thread": None, "result": None, "error": ""}
class RunStore:
def __init__(self, limit: int = MAX_RUNS):
self.limit = limit
self.runs: list[dict] = []
def add(self, run_id: str, object_code: str, channel: str, request_summary: dict,
response: object, layers: dict, followup_sql: list[str] | None = None,
gaps: list[str] | None = None, scenario_id: str | None = None,
nodes: list[dict] | None = None) -> dict:
entry = {
"runId": run_id,
"time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
"objectCode": object_code,
"channel": channel,
"scenarioId": scenario_id,
"nodes": nodes,
"request": redact_obj(request_summary, _secrets(), _access_keys()),
"response": redact_obj(response, _secrets(), _access_keys()),
"layers": layers,
"followupSql": followup_sql or [],
"knownGaps": gaps or [],
}
self.runs.append(entry)
if len(self.runs) > self.limit:
del self.runs[: len(self.runs) - self.limit]
return entry
def recent(self) -> list[dict]:
return list(reversed(self.runs[-50:]))
RUNS = RunStore()
app = FastAPI(title="Ai-DOP Integration Simulator", version="1.0.0")
# ---------------------------------------------------------------------------
# Mock API 内嵌启停
# ---------------------------------------------------------------------------
class _MockServerHandle:
def __init__(self):
self.server: uvicorn.Server | None = None
self.task: asyncio.Task | None = None
self.last_error: str = ""
MOCK_HANDLE = _MockServerHandle()
def _port_in_use(port: int) -> bool:
"""真实探测本机端口是否已被占用。
不依赖 uvicorn 内部属性：0.30+ 的 Server 已无 startup_error，
读它会抛 AttributeError 并把已启动的服务误报成 500。
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.settimeout(0.3)
return probe.connect_ex(("127.0.0.1", int(port))) == 0
async def _start_mock_api() -> dict:
port = int(CONFIG["mockApiPort"])
if MOCK_HANDLE.task and not MOCK_HANDLE.task.done():
return {"running": True, "port": port, "alreadyRunning": True}
# 端口预检必须在建 task 之前：否则失败路径会留下一个已赋值的句柄，
# 使 _mock_running() 与实际监听状态不一致。
if _port_in_use(port):
MOCK_HANDLE.last_error = f"端口 {port} 已被占用（可能是另一个模拟器实例）"
return {"running": False, "error": MOCK_HANDLE.last_error}
config = uvicorn.Config(MOCK_APP, host="127.0.0.1", port=port, log_level="warning")
server = uvicorn.Server(config)
MOCK_HANDLE.server = server
MOCK_HANDLE.task = asyncio.get_running_loop().create_task(server.serve())
# 轮询库的公开状态位，上限 ~2s：比固定 sleep 既不会过早返回未就绪，也不会白等。
for _ in range(40):
await asyncio.sleep(0.05)
if server.started:
MOCK_HANDLE.last_error = ""
return {"running": True, "port": port}
if server.should_exit or MOCK_HANDLE.task.done():
break
MOCK_HANDLE.last_error = f"Mock API 在端口 {port} 上启动失败"
failed_task = MOCK_HANDLE.task
MOCK_HANDLE.server = None
MOCK_HANDLE.task = None
if failed_task and not failed_task.done():
failed_task.cancel()
return {"running": False, "error": MOCK_HANDLE.last_error}
async def _stop_mock_api() -> dict:
if MOCK_HANDLE.server and MOCK_HANDLE.task and not MOCK_HANDLE.task.done():
MOCK_HANDLE.server.should_exit = True
try:
await asyncio.wait_for(MOCK_HANDLE.task, timeout=5)
except asyncio.TimeoutError:
MOCK_HANDLE.task.cancel()
return {"running": False}
return {"running": False, "alreadyStopped": True}
def _mock_running() -> bool:
return bool(MOCK_HANDLE.task and not MOCK_HANDLE.task.done() and
MOCK_HANDLE.server and MOCK_HANDLE.server.started)
# ---------------------------------------------------------------------------
# 基础模型
# ---------------------------------------------------------------------------
class RowsPayload(BaseModel):
objectCode: str
runId: str | None = None
rows: list[dict] | None = None
class InboundExecPayload(BaseModel):
objectCode: str | None = None
entityCode: str | None = None
operation: str = "push" # schema|push|receipt|bulk|snapshot_open|snapshot_commit|digest
runId: str | None = None
rows: list[dict] | None = None
contractVersion: str | None = None
# 用例与操作参数
case: str | None = None # happy|missing-idem|tamper-body|expired-ts|nonce-replay|same-key-replay|same-key-diff-body|bad-signature
syncBatchId: str | None = None
snapshotId: str | None = None
force: bool = False
bizDate: str | None = None
class ScenarioPayload(BaseModel):
scenarioId: str
runId: str | None = None
channelOverride: dict[str, str] | None = None
class ReportPayload(BaseModel):
"""全链路报告草稿（任务书 F5 §10）：只保存测试人员填写的摘要。"""
runId: str
channel: str # DB_SYNC | API_PULL | API_INBOUND
environment: str = "" # 环境与测试租户
objectCode: str = "" # 对象
simPrep: str = "" # 模拟端准备结果
manualSteps: str = "" # 人工执行步骤
simEvidence: str = "" # 模拟端证据
aidopConn: str = "" # Ai-DOP 测连结果
flowIds: str = "" # 同步/入站流水编号
stagingCheck: str = "" # 贴源核验结果
targetResult: str = "" # 目标业务结果
conclusion: str = "" # 结论
gaps: str = "" # 未完成项
cleanup: str = "" # 清理提示
class FaultApplyPayload(BaseModel):
scenarioId: str
objectCode: str | None = None # DB/API 场景默认 ITEM
entityCode: str | None = None # N3-15 未启用 entity 场景的目标实体码
class FaultReportPayload(BaseModel):
"""异常测试报告（任务书 §15 八列）。「是否符合预期」JSON 键为 pass。"""
scenarioId: str
simBehavior: str = ""
expected: str = ""
actualResponse: str = ""
aidopFlowStatus: str = ""
targetChanged: str = ""
conform: str = Field(default="", alias="pass")
note: str = ""
class LoadPlanPayload(BaseModel):
"""压力测试计划（§16 十项 + §17 参数 + 通道专属参数，§20 安全限制）。
校验由 load.controls.validate_plan 执行；本模型只负责字段承接。
"""
channel: str # DB_SYNC | API_PULL | API_INBOUND
objectCode: str = ""
entityCode: str = ""
runId: str | None = None
# §16 执行前必填十项
environment: str = ""
tenant: str = ""
owner: str = ""
timeWindow: str = ""
maxDataRows: int = 0
maxConcurrency: int = 0
maxQps: int = 0
maxDurationSeconds: int = 0
stopThreshold: str = ""
cleanupPlan: str = ""
# 安全
confirmed: bool = False # 共享环境负责人确认
escalationConfirm: bool = False # 提高上限二次确认
# §17 通用参数
warmupSeconds: int = 0
batchSize: int = 1
requestIntervalMs: int = 0
timeoutSeconds: int = 30
maxRetries: int = 0
sampleMode: str = "fixed" # fixed | random
maxConsecutiveErrors: int = 0
maxTimeouts: int = 0
maxRateLimited: int = 0
errorRate: float = 0.0 # P2 Mock 随机 500 比例 / P3 错误数据比例
# P1 专属（§18）
size: str = "" # S | M | L | XL（记录档位，行数以 maxDataRows 为准）
incrementRatio: float = 1.0
fieldWidth: int = 0
# P2 专属（§19）
pageSize: int = 0
totalRowsLimit: int = 0
randomDelay: float = 0.0
rateLimit: float = 0.0
# P3 专属（§20）
mode: str = "push" # push | bulk
rampSteps: list[dict] | None = None # 阶梯升压 [{duration,qps,concurrency}]
idemReplay: bool = False # 幂等重复
class LoadReportPayload(BaseModel):
"""压力报告人工观察部分（自动指标与计划由后端快照填充，§18 验收/§17 末句）。"""
runId: str = ""
channel: str = ""
aidopResource: str = "" # Ai-DOP CPU/内存/队列/数据库人工观察结果
aidopFlow: str = "" # Ai-DOP 同步/入站流水核对结果
conclusion: str = ""
notes: str = ""
CHANNELS = ("DB_SYNC", "API_PULL", "API_INBOUND")
CHANNEL_LABELS = {
"DB_SYNC": "方式甲 DB_SYNC（数据库同步）",
"API_PULL": "方式乙 API_PULL（接口拉取）",
"API_INBOUND": "方式丙 API_INBOUND（接口推送）",
}
# 报告章节：与任务书 §10 的字段清单一一对应
REPORT_FIELDS = [
("runId", "运行编号 runId"),
("channel", "集成方式"),
("environment", "环境与测试租户"),
("objectCode", "对象"),
("simPrep", "模拟端准备结果"),
("manualSteps", "人工执行步骤"),
("simEvidence", "模拟端证据"),
("aidopConn", "Ai-DOP 测连结果"),
("flowIds", "同步/入站流水编号"),
("stagingCheck", "贴源核验结果"),
("targetResult", "目标业务结果"),
("conclusion", "结论"),
("gaps", "未完成项"),
("cleanup", "清理提示"),
]
class ReportStore:
"""全链路报告存储（任务书 F5）。
- 只保存测试人员填写的摘要字段，不保存数据库密码、Secret、JWT、连接串或完整敏感 payload；
- 保存前统一过 redact_obj，导出内容与内存一致；
- 内存态、不落盘：控制台重启即清空，避免凭据残留。
"""
def __init__(self, limit: int = 50):
self.limit = limit
self.items: list[dict] = []
def save(self, fields: dict, secrets: list[str] | None = None,
access_keys: list[str] | None = None) -> dict:
entry = {
"id": uuid.uuid4().hex[:12],
"createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
}
entry.update(redact_obj(dict(fields), secrets, access_keys))
self.items.append(entry)
if len(self.items) > self.limit:
del self.items[: len(self.items) - self.limit]
return entry
def list(self) -> list[dict]:
return list(reversed(self.items[-20:]))
def get(self, report_id: str) -> dict | None:
return next((r for r in self.items if r["id"] == report_id), None)
def delete(self, report_id: str) -> bool:
before = len(self.items)
self.items = [r for r in self.items if r["id"] != report_id]
return len(self.items) < before
def build_report_markdown(entry: dict) -> str:
"""报告导出 Markdown（任务书 F5 §10 字段 + 脱敏声明）。"""
lines = ["# Ai-DOP 第三方系统模拟器 · 全链路测试报告", ""]
for key, label in REPORT_FIELDS:
value = str(entry.get(key) or "").strip()
if key == "channel":
value = CHANNEL_LABELS.get(value, value)
lines.append(f"## {label}")
lines.append(value or "（未填写）")
lines.append("")
lines.append("---")
lines.append(f"报告 ID：{entry.get('id', '')} · 创建时间：{entry.get('createdAt', '')}")
lines.append("")
lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
"数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
return "\n".join(lines)
REPORTS = ReportStore()
# 压力报告复用 ReportStore（保存前脱敏、内存态不落盘）
LOAD_REPORTS = ReportStore(limit=20)
class FaultReportStore(ReportStore):
"""异常测试报告存储（任务书 §15 N4）。复用脱敏与内存态约定。
每场景一行最终结论：同 scenarioId 重复保存 = 原位更新（保留原 id），
避免误点/重复点保存产生重复行。
"""
def save(self, fields: dict, secrets: list[str] | None = None,
access_keys: list[str] | None = None) -> dict:
scenario_id = fields.get("scenarioId")
existing = next((r for r in self.items if r.get("scenarioId") == scenario_id), None)
if existing is None:
entry = super().save(fields, secrets, access_keys)
replaced = False
else:
entry = existing
entry.update(redact_obj(dict(fields), secrets, access_keys))
entry["updatedAt"] = time.strftime("%Y-%m-%dT%H:%M:%S")
replaced = True
scenario = fault_catalog.find_scenario(entry.get("scenarioId", ""))
entry["scenarioTitle"] = scenario["title"] if scenario else ""
entry["channel"] = scenario["channel"] if scenario else ""
entry["replaced"] = replaced
return entry
FAULT_REPORTS = FaultReportStore()
def build_fault_report_markdown(entry: dict) -> str:
"""异常报告导出 Markdown（任务书 §15 八列 + 脱敏声明）。"""
lines = ["# Ai-DOP 第三方系统模拟器 · 异常测试报告", ""]
lines.append(f"- 场景：{entry.get('scenarioId', '')} {entry.get('scenarioTitle', '')}")
lines.append(f"- 集成方式：{fault_catalog.FAULT_CHANNEL_LABELS.get(entry.get('channel', ''), entry.get('channel', ''))}")
lines.append(f"- 创建时间：{entry.get('createdAt', '')}")
lines.append("")
lines.append("| 项目 | 内容 |")
lines.append("|---|---|")
for key, label in fault_catalog.FAULT_REPORT_FIELDS:
value = str(entry.get(key) or "").strip() or "（未填写）"
lines.append(f"| {label} | {value.replace(chr(10), '
')} |")
lines.append("")
lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
"数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# 公共辅助
# ---------------------------------------------------------------------------
def _require_obj(object_code: str) -> dict:
obj = sa.find_object(object_code)
if not obj:
raise HTTPException(status_code=404, detail=f"unknown objectCode: {object_code}")
return obj
def _build_inbound_client() -> InboundClient:
base = CONFIG["aidopBaseUrl"]
try:
aidop_probe.assert_host_allowed(base, CONFIG["allowedTargetHosts"])
except aidop_probe.HostNotAllowedError as ex:
raise HTTPException(status_code=400, detail=str(ex))
access_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
secret = _env("AIDOP_SIM_INBOUND_SECRET")
missing = [n for n, v in (("AIDOP_SIM_INBOUND_ACCESS_KEY", access_key),
("AIDOP_SIM_INBOUND_SECRET", secret)) if not v]
if missing:
raise HTTPException(status_code=400,
detail=f"缺少环境变量: {', '.join(missing)}（Secret 只从环境变量读取，不落盘）")
return InboundClient(base, access_key, secret)
def _resolve_entity(payload: InboundExecPayload) -> tuple[dict | None, str]:
"""解析本次执行的对象与 entityCode。仅给 entityCode 时允许无目录对象直发（rows 必填）。"""
if payload.objectCode:
obj = _require_obj(payload.objectCode)
else:
obj = _require_obj_by_entity(payload.entityCode or "")
if payload.entityCode:
entity = payload.entityCode.upper()
else:
if not obj:
raise HTTPException(status_code=400, detail="need objectCode or entityCode")
inbound = obj.get("apiInbound") or {}
if not inbound.get("supported"):
raise HTTPException(status_code=400,
detail=f"{obj['objectCode']} 不支持 API_INBOUND: {inbound.get('reason', '无契约')}")
entity = inbound["entityCode"]
return obj, entity
def _require_obj_by_entity(entity_code: str) -> dict | None:
for obj in sa.load_catalog()["objects"]:
if (obj.get("apiInbound") or {}).get("entityCode") == entity_code.upper():
return obj
return None
def _first_required_field(entity: str) -> str:
"""契约实体第一个必填字段（N3 缺必填字段/类型错误/部分行拒绝场景用）。"""
spec = sa.INBOUND_SPECS.get(entity) or []
for field, required, _ in spec:
if required:
return field
return "Descr"
def _layers(code: str, config: str, data: str) -> dict:
return {"code": code, "config": config, "data": data}
def _followup_sql(obj: dict, channel: str, run_id: str) -> list[str]:
prefix = sa.biz_prefix(run_id)
db = (obj.get("dbSync") or {}).get("table")
sqls = [
"SELECT source_code, status, COUNT(*) cnt, MAX(end_time) latest FROM mdp_sync_log "
"WHERE source_code LIKE 'SIM_MODE3%' GROUP BY source_code, status;",
"SELECT COUNT(*) FROM mdp_inbound_request WHERE create_time >= CURDATE();",
]
if channel == "DB_SYNC" and db:
sqls.insert(0, f"-- 模拟库核对：\nSELECT COUNT(*), MAX(sourceUpdatedAt) FROM `{db}` WHERE bizKey LIKE '{prefix}%';")
sqls.append(f"SELECT * FROM `{db.replace('sim_', 'mdp_stg_sim_')}` WHERE raw_data LIKE '%{prefix}%' LIMIT 20;")
if channel == "API_INBOUND":
entity = (obj.get("apiInbound") or {}).get("entityCode")
sqls.append(f"SELECT status, accepted_rows, rejected_rows, sync_batch_id FROM mdp_inbound_request "
f"WHERE entity_code='{entity}' ORDER BY id DESC LIMIT 10;")
sqls.append("-- 标准层/KPI 下游核验属第二阶段自动对账范围；第一阶段报告 TRANSFORM_PENDING，不猜测成功。")
return sqls
# ---------------------------------------------------------------------------
# 状态与目录
# ---------------------------------------------------------------------------
@app.get("/api/status")
def status():
inbound_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
return {
"controlPort": DEFAULT_CONTROL_PORT,
"aidopBaseUrl": CONFIG["aidopBaseUrl"],
"allowedTargetHosts": CONFIG["allowedTargetHosts"],
"aidopReachable": None, # 由 /api/precheck 实测，避免每次状态查询打真实请求
"mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"],
"authMode": MOCK_STATE.settings.auth_mode,
"forceStatus": MOCK_STATE.settings.force_status,
"lastError": MOCK_HANDLE.last_error},
"mysql": {
"database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT,
"passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD")),
},
"inbound": {
"accessKeyLoaded": bool(inbound_key),
"accessKeyMasked": mask_access_key(inbound_key),
"secretLoaded": bool(_env("AIDOP_SIM_INBOUND_SECRET")),
},
"adminTokenLoaded": bool(_env("AIDOP_SIM_ADMIN_TOKEN")),
"objectCounts": _channel_counts(),
}
def _channel_counts() -> dict:
counts = {"DB_SYNC": 0, "API_PULL": 0, "API_INBOUND": 0, "total": 0}
for obj in sa.load_catalog()["objects"]:
counts["total"] += 1
if (obj.get("dbSync") or {}).get("supported"):
counts["DB_SYNC"] += 1
if (obj.get("apiPull") or {}).get("supported"):
counts["API_PULL"] += 1
if (obj.get("apiInbound") or {}).get("supported"):
counts["API_INBOUND"] += 1
return counts
@app.get("/api/catalog")
def catalog():
return sa.load_catalog()
@app.get("/api/scenarios")
def scenarios():
return sa.load_scenarios()
@app.get("/api/sample/{object_code}")
def get_sample(object_code: str, runId: str | None = None):
obj = _require_obj(object_code)
run_id = runId or sa.new_run_id()
inbound = obj.get("apiInbound") or {}
result: dict = {"objectCode": obj["objectCode"], "runId": run_id,
"canonical": sa.load_canonical_rows(obj),
"dbSync": None, "apiPull": None, "apiInbound": None}
if (obj.get("dbSync") or {}).get("supported"):
result["dbSync"] = sa.adapt_db_rows(obj, run_id)
if (obj.get("apiPull") or {}).get("supported"):
result["apiPull"] = sa.adapt_api_rows(obj, run_id)
if inbound.get("supported"):
adapted = sa.adapt_inbound_rows(obj, run_id)
result["apiInbound"] = {
"entityCode": inbound["entityCode"],
"contractVersion": inbound.get("contractVersion", "v1"),
"rows": adapted["rows"],
"missingRequired": adapted["missing_required"],
"envelope": sa.inbound_envelope(adapted["rows"]),
}
return result
# ---------------------------------------------------------------------------
# Mock API 控制
# ---------------------------------------------------------------------------
@app.post("/api/mock-api/start")
async def mock_api_start():
return await _start_mock_api()
@app.post("/api/mock-api/stop")
async def mock_api_stop():
return await _stop_mock_api()
@app.get("/api/mock-api/settings")
def mock_api_settings():
s = MOCK_STATE.settings
return {"running": _mock_running(), "authMode": s.auth_mode, "delaySeconds": s.delay_seconds,
"forceStatus": s.force_status, "responsePath": s.response_path,
"apiKeyHeader": s.api_key_header, "overrides": sorted(MOCK_STATE.overrides.keys())}
@app.post("/api/mock-api/control")
def mock_api_control(payload: dict):
"""统一控制入口：auth_mode/token/basic/apikey/delay/force_status/response_path/override。"""
s = MOCK_STATE.settings
if "auth_mode" in payload:
mode = str(payload["auth_mode"]).upper()
if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"):
raise HTTPException(status_code=400, detail=f"unknown auth_mode: {mode}")
s.auth_mode = mode
for key, attr in (("token", "token"), ("basic_user", "basic_user"),
("basic_password", "basic_password"), ("api_key_header", "api_key_header"),
("api_key_value", "api_key_value"), ("response_path", "response_path")):
if payload.get(key):
setattr(s, attr, str(payload[key]))
if "delay_seconds" in payload:
s.delay_seconds = max(0.0, float(payload["delay_seconds"]))
if "force_status" in payload:
s.force_status = int(payload["force_status"]) if payload["force_status"] else None
if payload.get("override_path") and isinstance(payload.get("override_rows"), list):
MOCK_STATE.overrides[payload["override_path"]] = payload["override_rows"]
if payload.get("clear_overrides"):
MOCK_STATE.overrides.clear()
return {"ok": True, "settings": mock_api_settings()}
@app.get("/api/mock-api/audit")
def mock_api_audit():
return {"entries": redact_obj(MOCK_STATE.audit[-50:], _secrets())}
# ---------------------------------------------------------------------------
# 模拟 MySQL 控制
# ---------------------------------------------------------------------------
@app.post("/api/db/ensure")
def db_ensure(payload: RowsPayload):
"""初始化模拟 schema 并确保对象源表存在（DDL 由样例推断）。"""
obj = _require_obj(payload.objectCode)
db_spec = obj.get("dbSync") or {}
if not db_spec.get("supported"):
raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC")
if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
raise HTTPException(status_code=400, detail="缺少环境变量 AIDOP_SIM_MYSQL_PASSWORD")
run_id = payload.runId or sa.new_run_id()
rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id)
try:
result = mock_db.init_schema({db_spec["table"]: {
"rows": rows, "businessKey": db_spec.get("businessKey"),
"incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}})
return {"ok": True, "runId": run_id, **result}
except mock_db.BlockedDatabaseError as ex:
raise HTTPException(status_code=400, detail=str(ex))
except Exception as ex:
raise HTTPException(status_code=502, detail=f"MySQL 不可达或执行失败: {type(ex).__name__}: {ex}")
@app.post("/api/db/load")
def db_load(payload: RowsPayload):
obj = _require_obj(payload.objectCode)
db_spec = obj.get("dbSync") or {}
if not db_spec.get("supported"):
raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC")
run_id = payload.runId or sa.new_run_id()
rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id)
try:
count = mock_db.upsert_rows(db_spec["table"], rows)
stats = mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt"))
except Exception as ex:
raise HTTPException(status_code=502, detail=f"装载失败（先执行 /api/db/ensure？）: {type(ex).__name__}: {ex}")
layers = _layers("SUPPORTED", "READY", "ACCEPTED")
RUNS.add(run_id, obj["objectCode"], "DB_SYNC",
{"action": "load", "table": db_spec["table"], "rowCount": count},
stats, layers, _followup_sql(obj, "DB_SYNC", run_id),
gaps=["Ai-DOP 侧同步须由 MdpDbPullExecutor 真实执行；本操作只装载第三方模拟源库"])
return {"ok": True, "runId": run_id, "loaded": count, "stats": stats, "layers": layers}
@app.post("/api/db/reset")
def db_reset(payload: RowsPayload):
obj = _require_obj(payload.objectCode)
table = (obj.get("dbSync") or {}).get("table")
if not table:
raise HTTPException(status_code=400, detail="object has no dbSync table")
try:
mock_db.reset_table(table)
return {"ok": True, "table": table}
except Exception as ex:
raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")
@app.post("/api/db/update")
def db_update(payload: dict):
obj = _require_obj(payload.get("objectCode", ""))
db_spec = obj.get("dbSync") or {}
table = db_spec.get("table")
updates = payload.get("updates") or {}
where = payload.get("where") or {}
if not table or not updates or not where:
raise HTTPException(status_code=400, detail="need objectCode/updates/where")
try:
affected = mock_db.update_rows(table, updates, where,
db_spec.get("incrementColumn", "sourceUpdatedAt"),
payload.get("newIncrement"))
return {"ok": True, "table": table, "affected": affected}
except Exception as ex:
raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")
@app.get("/api/db/stats/{object_code}")
def db_stats(object_code: str):
obj = _require_obj(object_code)
db_spec = obj.get("dbSync") or {}
if not db_spec.get("table"):
raise HTTPException(status_code=400, detail="object has no dbSync table")
try:
return mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt"))
except Exception as ex:
raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")
@app.get("/api/db/aidop-config/{object_code}")
def db_aidop_config(object_code: str):
obj = _require_obj(object_code)
base = f"http://127.0.0.1:{CONFIG['mockApiPort']}"
return mock_db.generate_aidop_config(obj, base)
# ---------------------------------------------------------------------------
# 预检与三层报告（P1-G）
# ---------------------------------------------------------------------------
@app.get("/api/precheck")
def precheck():
base = CONFIG["aidopBaseUrl"]
reach = aidop_probe.check_reachable(base, CONFIG["allowedTargetHosts"])
inbound_ready = bool(_env("AIDOP_SIM_INBOUND_ACCESS_KEY") and _env("AIDOP_SIM_INBOUND_SECRET"))
env_missing = [n for n in ("AIDOP_SIM_INBOUND_ACCESS_KEY", "AIDOP_SIM_INBOUND_SECRET",
"AIDOP_SIM_MYSQL_PASSWORD") if not _env(n)]
return {
"aidopBaseUrl": base,
"aidop": reach,
"mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"]},
"mysql": {"database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT,
"passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD"))},
"inboundCredentialsLoaded": inbound_ready,
"envMissing": env_missing,
"testSources": mock_db.SIM_SOURCES,
"note": "DB_SYNC/API_PULL 的来源、实体、租户登记须人工在 Ai-DOP 配置页核对（生成清单见 /api/db/aidop-config/{objectCode}）",
}
@app.get("/api/precheck/inbound")
def precheck_inbound():
"""26 个代码契约逐一 schema 预检（真实签名，只读）。"""
client = _build_inbound_client()
codes = sorted({(o.get("apiInbound") or {}).get("entityCode")
for o in sa.load_catalog()["objects"]
if (o.get("apiInbound") or {}).get("supported")})
results = aidop_probe.precheck_all_inbound(client, codes, CONFIG["allowedTargetHosts"])
return {"accessKey": mask_access_key(client.key), "count": len(results), "results": results}
@app.get("/api/precheck/inbound/{entity_code}")
def precheck_inbound_one(entity_code: str):
client = _build_inbound_client()
return aidop_probe.precheck_inbound_entity(client, entity_code.upper(), CONFIG["allowedTargetHosts"])
# ---------------------------------------------------------------------------
# API_INBOUND 执行（P1-E）
# ---------------------------------------------------------------------------
@app.get("/api/inbound/auth-probe")
def inbound_auth_probe(refresh: int = 0):
"""探测全部入站实体在 Ai-DOP 侧的授权状态（GET schema，只读无副作用）。
Secret 不离开后端进程；结果缓存 60 秒，refresh=1 强制重测。
前端实体下拉据此只展示已授权实体，避免选到 403 未授权实体。
"""
global _AUTH_PROBE_CACHE
now = time.monotonic()
if not refresh and _AUTH_PROBE_CACHE["data"] is not None \
and now - _AUTH_PROBE_CACHE["at"] < _AUTH_PROBE_TTL:
return {"ok": True, "cached": True, **_AUTH_PROBE_CACHE["data"]}
reach = aidop_probe.check_reachable(CONFIG["aidopBaseUrl"], CONFIG["allowedTargetHosts"],
timeout=5)
if not reach.get("reachable"):
# Ai-DOP 不可达：不逐个探测（避免 26 次连接等待），全部如实标记为不可用
entities = {inbound["entityCode"]: False
for obj in sa.load_catalog()["objects"]
if (inbound := ((obj or {}).get("apiInbound") or {})) and inbound.get("supported")}
data = {"entities": entities,
"probedAt": time.strftime("%Y-%m-%d %H:%M:%S"),
"probeErrors": [f"Ai-DOP 不可达: {reach.get('error')}"]}
_AUTH_PROBE_CACHE.update(at=time.monotonic(), data=data)
return {"ok": True, "cached": False, **data}
client = _build_inbound_client()
entities: dict[str, bool] = {}
probe_errors: list[str] = []
for obj in sa.load_catalog()["objects"]:
inbound = (obj or {}).get("apiInbound") or {}
if not inbound.get("supported"):
continue
entity = inbound["entityCode"]
try:
status, _, _ = client.op_schema(entity,
contract_version=inbound.get("contractVersion"))
entities[entity] = status == 200
if status not in (200, 403):
probe_errors.append(f"{entity}: HTTP {status}")
except Exception as ex: # noqa: BLE001 —— 单实体失败不阻断整体探测
probe_errors.append(f"{entity}: {type(ex).__name__}")
data = {"entities": entities, "probedAt": time.strftime("%Y-%m-%d %H:%M:%S"),
"probeErrors": probe_errors}
_AUTH_PROBE_CACHE.update(at=time.monotonic(), data=data)
return {"ok": True, "cached": False, **data}
@app.post("/api/inbound/exec")
def inbound_exec(payload: InboundExecPayload):
obj, entity = _resolve_entity(payload)
client = _build_inbound_client()
run_id = payload.runId or sa.new_run_id()
contract_version = payload.contractVersion
if obj is None:
# 仅按 entityCode 直发（无目录样例），rows 必填
rows = payload.rows
if rows is None and payload.operation in ("push", "bulk"):
raise HTTPException(status_code=400, detail="entityCode-only exec requires rows")
envelope = sa.inbound_envelope(rows or [])
else:
inbound = obj.get("apiInbound") or {}
contract_version = contract_version or inbound.get("contractVersion")
rows = payload.rows if payload.rows is not None else sa.adapt_inbound_rows(obj, run_id)["rows"]
envelope = sa.inbound_envelope(rows)
object_code = obj["objectCode"] if obj else entity
op = payload.operation
result: dict = {"runId": run_id, "entityCode": entity, "operation": op, "case": payload.case}
def record(status: int, body, headers, layers: dict, extra: dict | None = None):
result.update({"httpStatus": status, "response": body, "requestHeaders": headers, **(extra or {})})
RUNS.add(run_id, object_code, "API_INBOUND",
{"operation": op, "case": payload.case, "entityCode": entity,
"rowCount": len(rows or []), "contractVersion": contract_version},
{"httpStatus": status, "body": body, "headers": headers},
layers, _followup_sql(obj or {"objectCode": entity, "apiInbound": {"entityCode": entity}},
"API_INBOUND", run_id))
return result
# 预检（配置层）
pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], contract_version)
config_state = pre["configState"]
result["precheck"] = {"configState": config_state, "httpStatus": pre["httpStatus"]}
if op == "schema":
status, body, headers = client.op_schema(entity, contract_version)
layers = _layers("SUPPORTED", aidop_probe.classify_inbound_precheck(status, body), "NOT_RUN")
return record(status, body, headers, layers)
if config_state != "READY" and op in ("push", "bulk", "snapshot_open"):
layers = _layers("SUPPORTED", config_state, "NOT_RUN")
raise HTTPException(status_code=409, detail={
"message": f"配置层 {config_state}：普通推数只对 READY 实体执行（不伪造成功）",
"precheck": pre, "layers": layers})
if op == "push":
case = payload.case or "happy"
if case == "happy":
status, body, headers = client.op_push(entity, envelope, contract_version=contract_version)
elif case == "missing-idem":
status, body, headers = client.case_missing_idem(entity, envelope)
elif case == "tamper-body":
status, body, headers = client.case_tamper_body(entity, envelope)
elif case == "expired-ts":
status, body, headers = client.case_expired_ts(entity, envelope)
elif case == "bad-signature":
status, body, headers = client.case_bad_signature(entity, envelope)
elif case == "nonce-replay":
(s1, d1), (s2, d2) = client.case_nonce_replay(entity, envelope)
status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {}
elif case == "same-key-replay":
(s1, d1), (s2, d2) = client.case_same_key_replay(entity, envelope)
status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {}
elif case == "same-key-diff-body":
env2 = sa.inbound_envelope([{**r, "sourceUpdatedAt": sa.now_iso()} for r in (rows or [])])
(s1, d1), (s2, d2) = client.case_same_key_diff_body(entity, envelope, env2)
status, body, headers = s2, {"first": [s1, d1], "conflict": [s2, d2]}, {}
# ---- 第二阶段 N3 新增用例（被 Ai-DOP 正确拒绝 = 测试通过）----
elif case == "missing-access-key":
status, body, headers = client.case_missing_access_key(entity, envelope)
elif case == "wrong-access-key":
status, body, headers = client.case_wrong_access_key(entity, envelope)
elif case == "missing-required-field":
status, body, headers = client.case_missing_required_field(
entity, envelope, _first_required_field(entity))
elif case == "wrong-field-type":
# 用结构敌对值（嵌套对象）而非同类标量：字符串列塞数字可能被宽容转换，
# 必须用无法映射到标量列的值才能触发类型校验拒绝
status, body, headers = client.case_wrong_field_type(
entity, envelope, _first_required_field(entity), {"__sim_type_fault__": True})
elif case == "partial-reject":
env = json.loads(json.dumps(envelope, ensure_ascii=False))
if len(env["data"]["list"]) < 2:
# 样例仅 1 行时复制一行，保证「1 好 1 坏」部分行拒绝可演示
env["data"]["list"].append(
json.loads(json.dumps(env["data"]["list"][0], ensure_ascii=False)))
env["data"]["list"][1].pop(_first_required_field(entity), None)
status, body, headers = client.op_push(entity, env, contract_version=contract_version)
elif case == "stale":
status, body, headers = client.case_stale_rows(entity, envelope)
elif case == "unknown-entity":
status, body, headers = client.case_unknown_entity(envelope)
else:
raise HTTPException(status_code=400, detail=f"unknown case: {case}")
data_state = "ACCEPTED" if status == 202 else ("RUN_FAILED" if status >= 400 else "NOT_RUN")
layers = _layers("SUPPORTED", config_state, data_state)
return record(status, body, headers, layers,
extra={"gaps": ["HTTP 202 仅代表摄取受理；贴源/标准层核验见 followupSql（TRANSFORM_PENDING）"]})
if op == "receipt":
if not payload.syncBatchId:
raise HTTPException(status_code=400, detail="receipt requires syncBatchId")
status, body, headers = client.op_receipt(payload.syncBatchId)
layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED")
return record(status, body, headers, layers)
if op == "bulk":
# 后端 NDJSON 分块：每行须为完整信封（data.list）+ 根级 idempotencyKey（MdpInboundReceiveService.BulkAsync）
bulk_rows = [{"idempotencyKey": f"{run_id}-L{i + 1}",
"data": {"list": [r]}} for i, r in enumerate(rows or [])]
if payload.case == "bulk-partial-failure":
# N3-16：坏行缺必填字段，其余行完好（观察逐行结果与计数）
field = _first_required_field(entity)
if bulk_rows:
if len(bulk_rows) < 2:
# 样例仅 1 行时复制一行（新幂等键），保证「部分失败」可演示
extra = json.loads(json.dumps(bulk_rows[0], ensure_ascii=False))
extra["idempotencyKey"] = f"{run_id}-L{len(bulk_rows) + 1}"
bulk_rows.append(extra)
bulk_rows[-1]["data"]["list"][0].pop(field, None)
status, body, headers = client.op_bulk(entity, bulk_rows, contract_version=contract_version)
layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 202 else "RUN_FAILED")
return record(status, body, headers, layers)
if op == "snapshot_open":
status, body, headers = client.op_snapshot_open(entity)
layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 201) else "RUN_FAILED")
return record(status, body, headers, layers)
if op == "snapshot_commit":
if not payload.snapshotId:
raise HTTPException(status_code=400, detail="snapshot_commit requires snapshotId")
status, body, headers = client.op_snapshot_commit(entity, payload.snapshotId, payload.force)
layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 202) else "RUN_FAILED")
return record(status, body, headers, layers)
if op == "digest":
biz_date = payload.bizDate or time.strftime("%Y-%m-%d")
status, body, headers = client.op_digest(entity, biz_date)
layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED")
return record(status, body, headers, layers)
raise HTTPException(status_code=400, detail=f"unknown operation: {op}")
# ---------------------------------------------------------------------------
# 场景执行（P1-F）
# ---------------------------------------------------------------------------
@app.post("/api/scenarios/run")
def scenario_run(payload: ScenarioPayload):
scenario = next((s for s in sa.load_scenarios()["scenarios"]
if s["scenarioId"] == payload.scenarioId), None)
if not scenario:
raise HTTPException(status_code=404, detail=f"unknown scenario: {payload.scenarioId}")
run_id = payload.runId or sa.new_run_id()
overrides = payload.channelOverride or {}
nodes_out = []
for node in scenario["nodes"]:
obj = sa.find_object(node["objectCode"])
channel = overrides.get(node["objectCode"], node["channel"])
node_result = {"objectCode": node["objectCode"], "channel": channel,
"relationKey": node.get("relationKey")}
if not obj:
node_result.update({"state": "CODE_UNSUPPORTED", "message": "目录中无此对象"})
nodes_out.append(node_result)
continue
spec = {"DB_SYNC": obj.get("dbSync"), "API_PULL": obj.get("apiPull"),
"API_INBOUND": obj.get("apiInbound")}.get(channel)
if not spec or not spec.get("supported"):
reason = (spec or {}).get("reason") or node.get("inboundNote") or "该通道不支持"
node_result.update({"state": "CODE_UNSUPPORTED", "message": reason})
nodes_out.append(node_result)
continue
try:
if channel == "DB_SYNC":
rows = sa.adapt_db_rows(obj, run_id)
if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
raise RuntimeError("缺少 AIDOP_SIM_MYSQL_PASSWORD")
mock_db.init_schema({spec["table"]: {"rows": rows, "businessKey": spec.get("businessKey"),
"incrementColumn": spec.get("incrementColumn")}})
count = mock_db.upsert_rows(spec["table"], rows)
node_result.update({"state": "ACCEPTED", "loaded": count, "table": spec["table"]})
elif channel == "API_PULL":
if not _mock_running():
raise RuntimeError("Mock API 未启动（先 POST /api/mock-api/start）")
rows = sa.adapt_api_rows(obj, run_id)
MOCK_STATE.overrides[spec["path"]] = rows
node_result.update({"state": "ACCEPTED", "overridePath": spec["path"], "rowCount": len(rows),
"note": "已在 Mock API 挂载适配样例；由 Ai-DOP MdpApiPullExecutor 真实拉取"})
else:
client = _build_inbound_client()
entity = spec["entityCode"]
version = node.get("contractVersion") or spec.get("contractVersion")
pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], version)
if pre["configState"] != "READY":
node_result.update({"state": pre["configState"],
"message": f"配置层 {pre['configState']}，未推数（不伪造成功）"})
else:
rows = sa.adapt_inbound_rows(obj, run_id)["rows"]
status, body, _ = client.op_push(entity, sa.inbound_envelope(rows),
contract_version=version)
node_result.update({"state": "ACCEPTED" if status == 202 else "RUN_FAILED",
"httpStatus": status, "response": body})
except HTTPException:
raise
except aidop_probe.HostNotAllowedError as ex:
node_result.update({"state": "SOURCE_UNREACHABLE", "message": str(ex)})
except Exception as ex:
node_result.update({"state": "RUN_FAILED", "message": f"{type(ex).__name__}: {ex}"})
nodes_out.append(node_result)
unsupported = [n["objectCode"] for n in nodes_out if n["state"] == "CODE_UNSUPPORTED"]
layers = _layers("SUPPORTED",
"READY" if all(n["state"] not in ("SOURCE_UNREACHABLE",) for n in nodes_out) else "SOURCE_UNREACHABLE",
"ACCEPTED" if all(n["state"] in ("ACCEPTED", "CODE_UNSUPPORTED") for n in nodes_out) else "RUN_FAILED")
entry = RUNS.add(run_id, ",".join(n["objectCode"] for n in nodes_out), f"SCENARIO:{payload.scenarioId}",
{"scenario": scenario["name"]}, {"nodes": nodes_out}, layers,
_followup_sql({"objectCode": "SCENARIO", "dbSync": {}}, "SCENARIO", run_id),
gaps=([f"以下节点该通道不支持：{', '.join(unsupported)}（链路不完整，不得宣称全链路跑通）"]
if unsupported else []),
scenario_id=payload.scenarioId, nodes=nodes_out)
return entry
# ---------------------------------------------------------------------------
# 运行记录
# ---------------------------------------------------------------------------
@app.get("/api/runs")
def runs():
return {"runs": RUNS.recent()}
# ---------------------------------------------------------------------------
# 异常测试（第二阶段 N1-N4，任务书 §12-§15）
# ---------------------------------------------------------------------------
@app.get("/api/fault/catalog")
def fault_catalog_route():
return {"scenarios": fault_catalog.FAULT_SCENARIOS,
"counts": fault_catalog.scenario_counts(),
"reportFields": fault_catalog.FAULT_REPORT_FIELDS}
def _apply_db_fault(directive: dict, object_code: str) -> dict:
"""执行 N1 模拟库异常预置（只操作模拟库，硬阻断保证不触碰 Ai-DOP 库）。"""
obj = _require_obj(object_code)
db_spec = obj.get("dbSync") or {}
table = db_spec.get("table")
if not table:
raise HTTPException(status_code=400, detail=f"{object_code} 无 dbSync 表（N1 场景需要）")
if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
raise HTTPException(status_code=400,
detail="缺少 AIDOP_SIM_MYSQL_PASSWORD 环境变量：模拟库操作需要 MySQL 凭据")
rows = sa.adapt_db_rows(obj, sa.new_run_id())
spec = {"rows": rows, "businessKey": db_spec.get("businessKey"),
"incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}
action = directive["type"]
try:
if action == "db_drop_table":
mock_db.drop_table(table)
return {"ok": True, "table": table, "note": f"已删除源表 {table}（Ai-DOP 侧保持配置不变再触发同步）"}
if action == "db_load_null_bizkey":
mock_db.drop_table(table) # 从正常结构出发，保证场景独立可复现
mock_db.init_schema({table: spec})
null_row = {k: (None if k == "bizKey" else v) for k, v in rows[0].items()}
mock_db.insert_rows_raw(table, [null_row])
return {"ok": True, "table": table, "note": f"已向 {table} 追加一行 bizKey=NULL 的行"}
if action == "db_load_duplicate_bizkey":
# 必须先 DROP：init_schema 的 DDL 是 CREATE TABLE IF NOT EXISTS，
# 表已存在时不会重建，旧唯一键会残留导致重复行插入报 1062。
mock_db.drop_table(table)
mock_db.init_schema({table: {**spec, "unique_key": False}})
mock_db.insert_rows_raw(table, [rows[0], dict(rows[0])])
return {"ok": True, "table": table,
"note": f"{table} 已重建为无唯一键表并装载两条相同 bizKey 的行"}
if action == "db_update_no_advance":
mock_db.drop_table(table) # 从正常结构出发，保证场景独立可复现
mock_db.init_schema({table: spec})
mock_db.upsert_rows(table, rows)
mock_db.update_rows(table, {"itemName": "N1-08 更新后不推进增量列"},
{"bizKey": rows[0]["bizKey"]}, spec["incrementColumn"],
advance_increment=False)
return {"ok": True, "table": table, "note": f"已更新 {table} 一行但未推进增量列"}
if action == "db_slow_view":
mock_db.init_schema({table: spec})
view = f"v_slow_{table}"
mock_db.create_slow_view(view, table, float(directive.get("sleepSeconds", 0.5)))
return {"ok": True, "view": view, "table": table,
"note": f"已创建慢查询视图 {view}；请测试人员在 Ai-DOP 侧把 source_table_name 改为该视图名"}
except HTTPException:
raise
except Exception as ex:
raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")
def _apply_mock_api_fault(directive: dict) -> dict:
s = MOCK_STATE.settings
settings = directive.get("settings") or {}
if settings.get("reset"):
s.reset_faults()
for key in ("force_status", "delay_seconds", "timeout_seconds", "empty_response",
"invalid_json", "ignore_cursor", "no_sort"):
if key in settings:
setattr(s, key, settings[key])
for key in ("content_type", "response_path"):
if key in settings:
setattr(s, key, settings[key])
return {"ok": True, "settings": mock_api_settings(),
"mockRunning": _mock_running(),
"note": "" if _mock_running() else "提示：Mock API 未启动，先点「启动 Mock API」"}
def _apply_mock_api_override(directive: dict, object_code: str) -> dict:
obj = _require_obj(object_code)
path = (obj.get("apiPull") or {}).get("path") or "/api/item"
rows = sa.adapt_api_rows(obj, sa.new_run_id())
mode = directive.get("mode")
if mode == "missing_bizkey":
for r in rows:
r.pop("bizKey", None)
elif mode == "duplicate_bizkey":
rows = rows + [dict(rows[0])]
MOCK_STATE.overrides[path] = rows
return {"ok": True, "overridePath": path, "rowCount": len(rows),
"mode": mode,
"note": "已挂载覆盖行；Ai-DOP 拉取该 path 时生效（恢复默认时清除）"}
@app.post("/api/fault/apply")
def fault_apply(payload: FaultApplyPayload):
scenario = fault_catalog.find_scenario(payload.scenarioId)
if not scenario:
raise HTTPException(status_code=404, detail=f"unknown scenarioId: {payload.scenarioId}")
if scenario["kind"] == "guide":
return {"applied": False, "guide": True, "scenario": scenario,
"note": "此场景需测试人员在 Ai-DOP 侧人工操作（模拟器不修改 Ai-DOP 配置）"}
directive = scenario["apply"]
dtype = directive["type"]
if dtype.startswith("db_"):
return {"applied": True, "scenario": scenario,
** _apply_db_fault(directive, payload.objectCode or "ITEM")}
if dtype == "mock_api":
return {"applied": True, "scenario": scenario,
**_apply_mock_api_fault(directive)}
if dtype == "mock_api_override":
return {"applied": True, "scenario": scenario,
**_apply_mock_api_override(directive, payload.objectCode or "ITEM")}
if dtype == "inbound_case":
case = directive["case"]
object_code = payload.objectCode or "ITEM"
if case == "disabled-entity":
entity = (payload.entityCode or "S5_INVENTORY_TXN").upper()
try:
pre = precheck_inbound_one(entity)
except HTTPException as ex:
pre = {"httpStatus": ex.status_code, "configState": "UNREACHABLE",
"message": ex.detail}
return {"applied": True, "scenario": scenario, "entityCode": entity,
"precheck": pre,
"note": "请测试人员先在 Ai-DOP 侧关闭该实体 inbound 开关，再用「API_INBOUND」页签向它推送；"
"被 Ai-DOP 正确拒绝 = 测试通过"}
if case == "snapshot-commit-failure":
exec_payload = InboundExecPayload(objectCode=object_code,
operation="snapshot_commit",
snapshotId="SNAP-NOT-EXIST-0001")
elif case == "bulk-partial-failure":
exec_payload = InboundExecPayload(objectCode=object_code, operation="bulk",
case="bulk-partial-failure")
elif case == "unknown-entity":
exec_payload = InboundExecPayload(objectCode=object_code, operation="push",
case="unknown-entity")
else:
exec_payload = InboundExecPayload(objectCode=object_code, operation="push", case=case)
try:
result = inbound_exec(exec_payload)
except HTTPException as ex:
# 仅「Ai-DOP 预检拦截」（409 + dict detail）才属于异常场景的「实际响应」：
# 正确拒绝 = 测试通过。模拟器自身的配置缺失（缺环境变量/白名单拦截等 400）
# 必须如实透传，不得包装成 expectedReject，否则会把「工具未发出请求」
# 误报成「Ai-DOP 拒绝了请求」。
if ex.status_code == 409 and isinstance(ex.detail, dict):
return {"applied": True, "scenario": scenario, "expectedReject": True,
"httpStatus": ex.status_code, "detail": ex.detail,
"note": "Ai-DOP 已在预检阶段拒绝（正确拒绝属于测试通过，不属于工具失败）"}
raise
return {"applied": True, "scenario": scenario, "result": result}
raise HTTPException(status_code=400, detail=f"unknown apply directive: {dtype}")
@app.post("/api/fault/recover")
def fault_recover(payload: dict):
"""恢复默认：channel 可传 DB_SYNC|API_PULL|API_INBOUND，缺省全部。"""
channel = payload.get("channel")
object_code = payload.get("objectCode") or "ITEM"
notes: list[str] = []
if channel in (None, "API_PULL"):
MOCK_STATE.settings.reset_faults()
cleared = len(MOCK_STATE.overrides)
MOCK_STATE.overrides.clear()
notes.append(f"Mock API 异常注入已全部恢复默认；清除 {cleared} 个覆盖行挂载"
f"（场景编排挂载需重跑场景恢复）")
if channel in (None, "DB_SYNC"):
obj = sa.find_object(object_code)
db_spec = (obj or {}).get("dbSync") or {}
table = db_spec.get("table")
if not table:
notes.append(f"{object_code} 无 dbSync 表，跳过模拟库恢复")
elif not _env("AIDOP_SIM_MYSQL_PASSWORD"):
notes.append("缺少 AIDOP_SIM_MYSQL_PASSWORD：模拟库恢复未执行"
"（设置环境变量后重试；或按场景说明人工恢复）")
else:
try:
rows = sa.adapt_db_rows(obj, sa.new_run_id())
spec = {"rows": rows, "businessKey": db_spec.get("businessKey"),
"incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}
# 先 DROP 再建：init_schema 的 DDL 是 CREATE TABLE IF NOT EXISTS，
# 不先删则 N1-07 的无唯一键表结构会残留，恢复后仍是重复键表。
mock_db.drop_table(table)
mock_db.init_schema({table: spec}) # 重建带唯一键的正常表
mock_db.upsert_rows(table, rows)
mock_db.drop_view(f"v_slow_{table}")
notes.append(f"{table} 已重建为正常表并装载样例行；慢查询视图已删除")
except Exception as ex:
notes.append(f"模拟库恢复失败：{type(ex).__name__}: {ex}")
if channel in (None, "API_INBOUND"):
notes.append("方式丙无状态注入无需恢复；Ai-DOP 侧实体开关由测试人员自行恢复")
return {"ok": True, "notes": notes}
@app.post("/api/fault-report/save")
def fault_report_save(payload: FaultReportPayload):
if not fault_catalog.find_scenario(payload.scenarioId):
raise HTTPException(status_code=400, detail=f"unknown scenarioId: {payload.scenarioId}")
entry = FAULT_REPORTS.save(payload.model_dump(by_alias=True), _secrets(), _access_keys())
return {"ok": True, "report": entry}
@app.get("/api/fault-report/list")
def fault_report_list():
return {"reports": FAULT_REPORTS.list()}
@app.get("/api/fault-report/export/{report_id}")
def fault_report_export(report_id: str):
entry = FAULT_REPORTS.get(report_id)
if not entry:
raise HTTPException(status_code=404, detail="report not found")
md = build_fault_report_markdown(entry)
filename = f"aidop-sim-fault-report-{entry['scenarioId']}-{report_id}.md"
return Response(content=md, media_type="text/markdown; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'})
@app.delete("/api/fault-report/{report_id}")
def fault_report_delete(report_id: str):
if not FAULT_REPORTS.delete(report_id):
raise HTTPException(status_code=404, detail="report not found")
return {"ok": True}
# ---------------------------------------------------------------------------
# 压力测试（第三阶段 §16-§21）
# ---------------------------------------------------------------------------
def _apply_p2_load(plan: dict) -> dict:
"""P2：把压力参数注入 Mock API（§19 配置每次返回行数/总量/延迟/错误比例/限流阈值）。"""
s = MOCK_STATE.settings
s.load_mode = True
s.page_size = int(plan.get("pageSize") or 0)
s.total_rows_limit = int(plan.get("totalRowsLimit") or 0)
s.random_delay = float(plan.get("randomDelay") or 0)
s.error_rate = float(plan.get("errorRate") or 0)
s.rate_limit = float(plan.get("rateLimit") or 0)
MOCK_STATE.load_stats.reset()
return {"ok": True, "settings": mock_api_settings(), "mockRunning": _mock_running(),
"note": "" if _mock_running() else "提示：Mock API 未启动，先点「启动 Mock API」"}
def _stop_p2_load() -> dict:
"""P2：实时停止压力模式，恢复正常模式（§19）。"""
s = MOCK_STATE.settings
s.load_mode = False
s.page_size = 0
s.total_rows_limit = 0
s.random_delay = 0.0
s.error_rate = 0.0
s.rate_limit = 0.0
return {"ok": True, "note": "Mock 压力模式已停止，恢复正常模式"}
def _run_p1_generation(obj: dict, plan: dict, run_id: str) -> None:
"""P1 后台线程：批量生成档位数据（§18 批量生成/耗时行数/增量比例/字段宽度/一键停止）。"""
DB_GEN.update({"result": None, "error": ""})
table = (obj.get("dbSync") or {}).get("table")
rows = sa.adapt_db_rows(obj, run_id)
started = time.monotonic()
try:
result = mock_db.bulk_generate(
table, rows, int(plan["maxDataRows"]),
increment_ratio=float(plan.get("incrementRatio") or 1.0),
field_width=int(plan.get("fieldWidth") or 0),
batch_size=1000, stop_event=LOAD_GUARD.stop)
LOAD_METRICS.mark_start()
LOAD_METRICS.record(200, time.monotonic() - started,
rows=result["loaded"], batches=result["batches"])
LOAD_METRICS.mark_finish()
DB_GEN["result"] = result
except Exception as ex: # noqa: BLE001
LOAD_METRICS.mark_start()
LOAD_METRICS.record(500, time.monotonic() - started)
LOAD_METRICS.mark_finish()
DB_GEN["error"] = f"{type(ex).__name__}: {ex}"
finally:
LOAD_GUARD.finish()
@app.get("/api/load/limits")
def load_limits():
"""上限与必填项（前端展示与预校验）。"""
return {"requiredFields": list(load_controls.REQUIRED_PLAN_FIELDS),
"defaultLimits": load_controls.DEFAULT_LIMITS,
"hardLimits": load_controls.HARD_LIMITS,
"sizes": mock_db.LOAD_SIZES,
"guard": LOAD_GUARD.snapshot()}
@app.post("/api/load/plan")
def load_plan(payload: LoadPlanPayload):
"""提交压力计划：§16 十项必填 + 上限校验 + 负责人确认（默认不改变运行状态）。"""
if LOAD_GUARD.state in ("running", "stopping"):
raise HTTPException(status_code=409,
detail=f"当前状态 {LOAD_GUARD.state}：先停止本轮压测再提交新计划")
result = LOAD_GUARD.submit_plan(payload.model_dump())
if not result["ok"]:
raise HTTPException(status_code=400, detail={"errors": result["errors"]})
return result
@app.post("/api/load/start")
def load_start(payload: dict):
"""显式开启压测（§16：必须由测试人员显式开启）。按通道启动对应执行器。"""
if LOAD_GUARD.state != "ready":
raise HTTPException(status_code=409,
detail=f"当前状态 {LOAD_GUARD.state}：须先提交并通过压力计划校验")
plan = LOAD_GUARD.plan or {}
channel = plan["channel"]
run_id = LOAD_GUARD.start(plan.get("runId") or None)
try:
if channel == "API_INBOUND":
client = _build_inbound_client()
entity = (plan.get("entityCode") or payload.get("entityCode") or "").upper()
obj = _require_obj_by_entity(entity) or _require_obj(plan.get("objectCode") or "")
inbound = (obj or {}).get("apiInbound") or {}
if not inbound.get("supported"):
raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 API_INBOUND")
entity = entity or inbound["entityCode"]
rows = sa.adapt_inbound_rows(obj, run_id)["rows"]
global LOAD_RUNNER
LOAD_RUNNER = runner_p3.P3Runner(
client, entity, rows, plan, LOAD_GUARD, LOAD_METRICS,
contract_version=inbound.get("contractVersion"))
LOAD_RUNNER.start()
elif channel == "DB_SYNC":
obj = _require_obj(plan.get("objectCode") or "")
db_spec = obj.get("dbSync") or {}
if not db_spec.get("supported") or not db_spec.get("table"):
raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC")
if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
raise HTTPException(status_code=400, detail="缺少环境变量 AIDOP_SIM_MYSQL_PASSWORD")
thread = threading.Thread(target=_run_p1_generation, args=(obj, plan, run_id),
name=f"p1-gen-{run_id}", daemon=True)
DB_GEN["thread"] = thread
LOAD_GUARD.register_worker(thread)
thread.start()
elif channel == "API_PULL":
if not _mock_running():
raise HTTPException(status_code=400, detail="Mock API 未启动（先 POST /api/mock-api/start）")
_apply_p2_load(plan)
# P2 无后台负载：由 Ai-DOP 侧发压，压力模式由 /api/load/stop 实时停止
else:
raise HTTPException(status_code=400, detail=f"unknown channel: {channel}")
except HTTPException:
LOAD_GUARD.finish("启动失败")
raise
return {"ok": True, "runId": run_id, "channel": channel, "state": LOAD_GUARD.state}
@app.post("/api/load/stop")
def load_stop(payload: dict):
"""立即停止（§21）：P1/P3 置停止信号等待在途收尾；P2 恢复 Mock 正常模式。"""
plan = LOAD_GUARD.plan or {}
reason = str(payload.get("reason") or "manual")
p2_recovered = False
if plan.get("channel") == "API_PULL":
_stop_p2_load()
p2_recovered = True
result = LOAD_GUARD.request_stop(reason)
if plan.get("channel") == "API_PULL":
# P2 无后台线程：直接收尾（含无残留断言）
result.update(LOAD_GUARD.finish(reason))
return {"ok": True, "p2Recovered": p2_recovered, **result}
@app.get("/api/load/status")
def load_status():
"""实时状态（§21 实时基础指标 + 错误摘要）。"""
guard = LOAD_GUARD.snapshot()
out = {"guard": guard, "metrics": LOAD_METRICS.snapshot()}
channel = guard.get("planChannel")
if channel == "API_PULL":
out["p2"] = {"loadMode": MOCK_STATE.settings.load_mode,
"stats": MOCK_STATE.load_stats.snapshot()}
if channel == "DB_SYNC":
out["p1"] = {"result": DB_GEN.get("result"), "error": DB_GEN.get("error")}
try:
out["p1"]["db"] = mock_db.db_status()
except Exception: # noqa: BLE001 —— 模拟库不可达时只影响统计展示
out["p1"]["db"] = None
return out
@app.get("/api/load/aidop-checklist/{channel}")
def load_aidop_checklist(channel: str):
"""Ai-DOP 人工执行和观察清单（§18 末项；§17 末句资源观察）。"""
common = [
"确认压力计划已获负责人审批，在当前时间窗口内执行",
"打开 Ai-DOP 监控工具，观察 CPU、内存、队列深度、数据库连接数",
"记录 mdp_sync_log / mdp_inbound_request 流水号与状态",
]
if channel == "DB_SYNC":
steps = [
"在 Ai-DOP 登记 SIM_MODE3_DB 数据源并测连（连接指向模拟库）",
"确认实体 source_table_name 指向目标表，增量列与模拟库一致",
"人工触发首次全量同步，观察分批拉取与耗时",
"全量完成后核对 Ai-DOP 贴源表行数与模拟库源端行数（分页无丢行、无重复）",
"修改模拟库数据后触发增量，核对增量值与批次",
"同步中断后恢复，观察断点续拉状态是否可解释",
]
elif channel == "API_PULL":
steps = [
"在 Ai-DOP 登记 SIM_MODE3_API 数据源并测连（指向 Mock API :8018）",
"人工触发拉取，观察多批 cursor 推进与批次耗时",
"核对请求量与模拟器统计（请求量/QPS/429/5xx 可对应）",
"观察间歇 429/5xx/延迟下 Ai-DOP 的重试与错误提示",
"压力结束后停止 Mock 压力模式，确认可恢复正常拉取",
]
elif channel == "API_INBOUND":
steps = [
"观察 mdp_inbound_request 流水：受理数、拒收数与原因分类",
"核对贴源表 mdp_stg_* 新增行数与发送行数",
"观察 429 限流响应与恢复后继续受理的情况",
"压测结束后确认无持续入站请求（模拟器停止后 Ai-DOP 侧连接回落）",
]
else:
raise HTTPException(status_code=404, detail="unknown channel")
return {"channel": channel, "steps": common + steps}
LOAD_REPORT_FIELDS = [
("runId", "运行编号 runId"),
("channel", "集成方式"),
("environment", "环境"),
("tenant", "测试租户"),
("owner", "负责人"),
("timeWindow", "时间窗口"),
("maxDataRows", "最大数据量"),
("maxConcurrency", "最大并发"),
("maxQps", "最大 QPS"),
("maxDurationSeconds", "最大持续时间"),
("stopThreshold", "停止阈值"),
("cleanupPlan", "清理方案"),
("startedAt", "实际开始时间"),
("finishedAt", "实际结束时间"),
("stopReason", "停止原因"),
("aidopResource", "Ai-DOP 资源人工观察"),
("aidopFlow", "Ai-DOP 流水核对"),
("conclusion", "结论"),
("notes", "备注"),
]
def build_load_report_markdown(entry: dict) -> str:
"""压力报告导出 Markdown（§17 指标全列 + §16 计划 + 人工观察 + 脱敏声明）。"""
lines = ["# Ai-DOP 第三方系统模拟器 · 压力测试报告", ""]
metrics = entry.get("metrics") or {}
guard = entry.get("guard") or {}
lines.append(f"- 集成方式：{CHANNEL_LABELS.get(entry.get('channel', ''), entry.get('channel', ''))}")
lines.append(f"- runId：{entry.get('runId', '')}")
lines.append("")
lines.append("## 测试指标（§17）")
lines.append("| 指标 | 值 |")
lines.append("|---|---|")
for key, label in load_metrics.METRIC_FIELDS:
value = metrics.get(key)
if isinstance(value, dict):
value = ", ".join(f"{k}:{v}" for k, v in value.items()) or "—"
lines.append(f"| {label} | {value if value not in (None, '') else '—'} |")
lines.append("")
lines.append("## 计划与确认（§16）")
lines.append("| 项目 | 内容 |")
lines.append("|---|---|")
for key, label in LOAD_REPORT_FIELDS:
value = entry.get(key)
if key in ("startedAt", "finishedAt", "stopReason"):
value = guard.get(key, value)
value = str(value or "").strip() or "（未填写）"
lines.append(f"| {label} | {value.replace(chr(10), '
')} |")
lines.append("")
lines.append(f"报告 ID：{entry.get('id', '')} · 创建时间：{entry.get('createdAt', '')}")
lines.append("")
lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
"数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
return "\n".join(lines)
@app.post("/api/load/report")
def load_report_save(payload: LoadReportPayload):
"""保存压力报告：自动快照计划与指标 + 人工观察字段，保存前脱敏。"""
guard = LOAD_GUARD.snapshot()
fields = dict(payload.model_dump())
fields.update({
"metrics": LOAD_METRICS.snapshot(),
"guard": guard,
"environment": (LOAD_GUARD.plan or {}).get("environment", ""),
"tenant": (LOAD_GUARD.plan or {}).get("tenant", ""),
"owner": (LOAD_GUARD.plan or {}).get("owner", ""),
"timeWindow": (LOAD_GUARD.plan or {}).get("timeWindow", ""),
"maxDataRows": (LOAD_GUARD.plan or {}).get("maxDataRows", ""),
"maxConcurrency": (LOAD_GUARD.plan or {}).get("maxConcurrency", ""),
"maxQps": (LOAD_GUARD.plan or {}).get("maxQps", ""),
"maxDurationSeconds": (LOAD_GUARD.plan or {}).get("maxDurationSeconds", ""),
"stopThreshold": (LOAD_GUARD.plan or {}).get("stopThreshold", ""),
"cleanupPlan": (LOAD_GUARD.plan or {}).get("cleanupPlan", ""),
})
entry = LOAD_REPORTS.save(fields, _secrets(), _access_keys())
return {"ok": True, "report": entry}
@app.get("/api/load/report/list")
def load_report_list():
return {"reports": LOAD_REPORTS.list()}
@app.get("/api/load/report/export/{report_id}")
def load_report_export(report_id: str):
entry = LOAD_REPORTS.get(report_id)
if not entry:
raise HTTPException(status_code=404, detail="report not found")
md = build_load_report_markdown(entry)
filename = f"aidop-sim-load-report-{report_id}.md"
return Response(content=md, media_type="text/markdown; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'})
@app.delete("/api/load/report/{report_id}")
def load_report_delete(report_id: str):
if not LOAD_REPORTS.delete(report_id):
raise HTTPException(status_code=404, detail="report not found")
return {"ok": True}
# ---------------------------------------------------------------------------
# 全链路报告（F5）
# ---------------------------------------------------------------------------
@app.post("/api/report/save")
def report_save(payload: ReportPayload):
if payload.channel not in CHANNELS:
raise HTTPException(status_code=400, detail=f"unknown channel: {payload.channel}")
if not payload.runId.strip():
raise HTTPException(status_code=400, detail="runId required")
entry = REPORTS.save(payload.model_dump(), _secrets(), _access_keys())
return {"ok": True, "report": entry}
@app.get("/api/report/list")
def report_list():
return {"reports": REPORTS.list()}
@app.get("/api/report/evidence/{channel}")
def report_evidence(channel: str):
"""按方式取最近运行记录（已脱敏），供测试人员一键导入「模拟端证据」。"""
if channel not in CHANNELS:
raise HTTPException(status_code=404, detail="unknown channel")
entries = [r for r in RUNS.runs if r["channel"] == channel][-5:]
return {"entries": list(reversed(entries))}
@app.get("/api/report/export/{report_id}")
def report_export(report_id: str):
entry = REPORTS.get(report_id)
if not entry:
raise HTTPException(status_code=404, detail="report not found")
md = build_report_markdown(entry)
filename = f"aidop-sim-report-{entry['runId']}.md"
return Response(content=md, media_type="text/markdown; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'})
@app.delete("/api/report/{report_id}")
def report_delete(report_id: str):
if not REPORTS.delete(report_id):
raise HTTPException(status_code=404, detail="report not found")
return {"ok": True}
# ---------------------------------------------------------------------------
# 静态页面
# ---------------------------------------------------------------------------
@app.get("/")
def index():
return FileResponse(SIM_ROOT / "static" / "index.html")
app.mount("/static", StaticFiles(directory=SIM_ROOT / "static"), name="static")
def main() -> None:
print(f"[simulator] control panel: http://127.0.0.1:{DEFAULT_CONTROL_PORT}/")
print(f"[simulator] aidop target : {CONFIG['aidopBaseUrl']} (allowed hosts: {CONFIG['allowedTargetHosts']})")
uvicorn.run(app, host="127.0.0.1", port=DEFAULT_CONTROL_PORT, log_level="info")
if __name__ == "__main__":
main()