196 lines
8.6 KiB
Python
196 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""WP-CONSOLE-18 本机临时 Mongo 真实留存验收。
|
||
|
||
这个脚本只接受本机 Mongo 和带 ``wp_console_18_test_`` 前缀的数据库名。
|
||
它会初始化总控平台的 13 个集合、70 个索引，写入少量测试资料，
|
||
断开连接后重新连接读取，再回滚本任务索引，最后删除自己的临时数据库。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
from pymongo import MongoClient
|
||
|
||
APP_DIR = Path(__file__).resolve().parents[1] / "app"
|
||
if str(APP_DIR) not in sys.path:
|
||
sys.path.insert(0, str(APP_DIR))
|
||
|
||
from services.control_plane_indexes import ( # noqa: E402
|
||
CONTROL_PLANE_COLLECTIONS,
|
||
CONTROL_PLANE_INDEXES,
|
||
apply_control_plane_indexes,
|
||
rollback_control_plane_indexes,
|
||
)
|
||
|
||
DB_PREFIX = "wp_console_18_test_"
|
||
LOCAL_HOSTS = {"127.0.0.1", "localhost", "::1"}
|
||
DB_RE = re.compile(r"^wp_console_18_test_[a-z0-9][a-z0-9_-]{3,80}$")
|
||
|
||
|
||
def _guard_target(uri: str, database: str) -> None:
|
||
parsed = urlparse(uri)
|
||
host = (parsed.hostname or "").lower()
|
||
if host not in LOCAL_HOSTS:
|
||
raise ValueError("只允许连接本机Mongo")
|
||
if parsed.port == 27017:
|
||
raise ValueError("默认27017可能是正式库，本任务只允许独立测试端口")
|
||
if not DB_RE.fullmatch(database):
|
||
raise ValueError(f"数据库名必须以{DB_PREFIX}开头并带随机后缀")
|
||
|
||
|
||
def _utc_now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _wait_ping(client: MongoClient, timeout_seconds: float = 8.0) -> None:
|
||
deadline = time.monotonic() + timeout_seconds
|
||
last_error: Exception | None = None
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
client.admin.command("ping")
|
||
return
|
||
except Exception as exc: # pragma: no cover - 真实环境重试分支
|
||
last_error = exc
|
||
time.sleep(0.25)
|
||
raise RuntimeError(f"Mongo在规定时间内没有回应: {last_error}")
|
||
|
||
|
||
def _index_count(db) -> int:
|
||
return sum(
|
||
max(0, len(db[name].index_information()) - 1)
|
||
for name in CONTROL_PLANE_COLLECTIONS
|
||
if name in db.list_collection_names()
|
||
)
|
||
|
||
|
||
def _fixture_documents(run_id: str) -> dict[str, dict]:
|
||
now = _utc_now()
|
||
return {
|
||
"device_groups": {"group_id": f"grp-{run_id}", "name": "WP-CONSOLE-18测试分组", "device_ids": ["device-test-18"], "updated_at": now},
|
||
"device_releases": {"release_id": f"rel-{run_id}", "version": "18.0.0-test", "status": "ready", "created_at": now},
|
||
"integration_applications": {"app_id": f"app-{run_id}", "name": "WP-CONSOLE-18测试应用", "organization": "wp_console_18_test", "status": "active", "created_at": now},
|
||
"authorization_grants": {"grant_id": f"grant-{run_id}", "app_id": f"app-{run_id}", "status": "active", "device_ids": ["device-test-18"], "created_at": now},
|
||
"integration_usage_records": {"usage_id": f"usage-{run_id}", "app_id": f"app-{run_id}", "interface_scope": "console.read", "result": "success", "status_code": 200, "occurred_at": now},
|
||
"integration_documents": {"document_id": f"doc-{run_id}", "title": "WP-CONSOLE-18测试知识库条目", "category": "test", "visibility": "private", "status": "published", "updated_at": now},
|
||
"ai_tasks": {"task_id": f"task-{run_id}", "action": "console_test_read", "actor": "wp-console-18-test", "status": "completed", "trace_id": f"trace-{run_id}", "created_at": now, "updated_at": now},
|
||
}
|
||
|
||
|
||
def _write_fixtures(db, run_id: str) -> dict[str, str]:
|
||
documents = _fixture_documents(run_id)
|
||
for collection, document in documents.items():
|
||
db[collection].insert_one(document)
|
||
return {collection: document[next(key for key in document if key.endswith("_id"))] for collection, document in documents.items()}
|
||
|
||
|
||
def _read_fixtures(db, ids: dict[str, str]) -> dict[str, bool]:
|
||
result: dict[str, bool] = {}
|
||
for collection, expected_id in ids.items():
|
||
# Mongo会自动补一个内部的_id；读回时必须使用业务自己的编号。
|
||
key = next(
|
||
key for key in db[collection].find_one({})
|
||
if key != "_id" and key.endswith("_id")
|
||
)
|
||
result[collection] = db[collection].find_one({key: expected_id}) is not None
|
||
return result
|
||
|
||
|
||
def run(uri: str, database: str, report_dir: Path) -> dict:
|
||
_guard_target(uri, database)
|
||
report_dir.mkdir(parents=True, exist_ok=True)
|
||
run_id = uuid.uuid4().hex[:12]
|
||
result: dict = {
|
||
"task_id": "WP-CONSOLE-18",
|
||
"run_id": run_id,
|
||
"uri_host": urlparse(uri).hostname,
|
||
"database": database,
|
||
"target_guard": "local_only_test_database",
|
||
"started_at": _utc_now(),
|
||
"connection_reopened": False,
|
||
"cross_reopen_readback": False,
|
||
"business_data_preserved_after_rollback": False,
|
||
"cleaned_up": False,
|
||
}
|
||
client = MongoClient(uri, serverSelectionTimeoutMS=5000, connectTimeoutMS=5000)
|
||
try:
|
||
_wait_ping(client)
|
||
db = client[database]
|
||
first = apply_control_plane_indexes(db)
|
||
second = apply_control_plane_indexes(db)
|
||
result["first_init"] = {key: first[key] for key in ("collection_count", "index_count", "created_count", "existing_count")}
|
||
result["second_init"] = {key: second[key] for key in ("collection_count", "index_count", "created_count", "existing_count")}
|
||
if first["collection_count"] != 13 or first["index_count"] != 70 or first["created_count"] != 70:
|
||
raise AssertionError("首次初始化没有得到13个集合和70个索引")
|
||
if second["created_count"] != 0 or second["existing_count"] != 70:
|
||
raise AssertionError("第二次初始化没有做到幂等")
|
||
result["index_count_after_second_init"] = _index_count(db)
|
||
fixture_ids = _write_fixtures(db, run_id)
|
||
result["fixture_ids"] = fixture_ids
|
||
result["fixture_write_count"] = len(fixture_ids)
|
||
client.close()
|
||
result["connection_reopened"] = True
|
||
|
||
client = MongoClient(uri, serverSelectionTimeoutMS=5000, connectTimeoutMS=5000)
|
||
_wait_ping(client)
|
||
db = client[database]
|
||
readback = _read_fixtures(db, fixture_ids)
|
||
result["readback_after_reopen"] = readback
|
||
result["cross_reopen_readback"] = all(readback.values())
|
||
if not result["cross_reopen_readback"]:
|
||
raise AssertionError("断开重连后测试资料没有全部读回")
|
||
|
||
preview = rollback_control_plane_indexes(db, dry_run=True)
|
||
rollback = rollback_control_plane_indexes(db)
|
||
result["rollback_preview"] = {key: preview[key] for key in ("planned_count", "dropped_count")}
|
||
result["rollback"] = {key: rollback[key] for key in ("planned_count", "dropped_count")}
|
||
result["remaining_custom_indexes"] = _index_count(db)
|
||
preserved = _read_fixtures(db, fixture_ids)
|
||
result["readback_after_rollback"] = preserved
|
||
result["business_data_preserved_after_rollback"] = all(preserved.values())
|
||
if rollback["dropped_count"] != 70 or result["remaining_custom_indexes"] != 0 or not result["business_data_preserved_after_rollback"]:
|
||
raise AssertionError("回滚没有只移除本任务索引或业务数据未保留")
|
||
result["result"] = "PASS"
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
# 只删除本任务自己生成的临时数据库，不删除任何集合或正式库。
|
||
cleanup = MongoClient(uri, serverSelectionTimeoutMS=5000, connectTimeoutMS=5000)
|
||
try:
|
||
_wait_ping(cleanup)
|
||
cleanup.drop_database(database)
|
||
result["cleaned_up"] = True
|
||
finally:
|
||
cleanup.close()
|
||
result["finished_at"] = _utc_now()
|
||
(report_dir / "RESULT.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return result
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="WP-CONSOLE-18临时Mongo真实留存验收")
|
||
parser.add_argument("--uri", required=True)
|
||
parser.add_argument("--database", required=True)
|
||
parser.add_argument("--report-dir", type=Path, required=True)
|
||
args = parser.parse_args()
|
||
try:
|
||
result = run(args.uri, args.database, args.report_dir)
|
||
except Exception as exc:
|
||
print(json.dumps({"task_id": "WP-CONSOLE-18", "result": "BLOCKED_OR_FAILED", "error": str(exc)}, ensure_ascii=False))
|
||
return 2
|
||
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|