# -*- coding: utf-8 -*- """本机 CI stage：验收门 = 真实 PostgreSQL16 + Redis 就绪 → `pytest -m pg` 全绿。 背景：工程已纳入 git（Gitea gonsun/dpb，2026-08-06 起）；CI stage 落地为本机驱动脚本。 v8 验收标准「CI 跑通 pytest 全绿」= 本脚本 EXIT=0。 step 0 探测真实 PG16（backend/env/.env.dev 的 DATABASE_*，可用 TC_CI_DB_* 覆盖）+ Redis （REDIS_*）连通。本机 PG/Redis 未就绪且 docker 可用时，自动 `docker compose -f docker/test-compose.yaml up -d` 拉起测试依赖栈 （postgres:16:5433 + redis:7:6380，预建 dpb_test）并注入 TC_CI_* 覆盖后重探。 仍失败给出明确指引并 exit 非 0。 step 1 backend 目录下 `pytest -m pg`（dpb python + -X utf8）跑 10 套统计引擎正式 tc 套件 （backend/tests/test_stat_analysis_tc.py 包装层，subprocess 调真实脚本）。 step 2 汇总 EXIT=0 → 「CI 全绿」；日志写 Temp/claude/ci_logs/。 说明： - 全程真实 PG16 + Redis（本工程无 SQLite）；PG/Redis 缺失时 step 0 拦截，不会误绿。 - 与 run_regression.py 并存：run_regression = 手动全量回归（e2e 探针 + tc 套件）； run_ci = 验收门（只跑 pytest -m pg 的 10 套正式 tc）。 - 解释器默认 sys.executable（须用 dpb env python 运行）；可用 DPB_PYTHON 覆盖。 """ import asyncio import os import subprocess import sys sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.abspath(__file__)) BACKEND = os.path.join(ROOT, "backend") ENV_FILE = os.path.join(BACKEND, "env", ".env.dev") LOG_DIR = os.path.join(ROOT, "Temp", "claude", "ci_logs") PY = os.environ.get("DPB_PYTHON", sys.executable) def _env_cfg() -> dict: cfg = {} if os.path.isfile(ENV_FILE): for line in open(ENV_FILE, encoding="utf-8").read().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) cfg[k.strip()] = v.strip().strip("'\"") for key, envvar in { "DATABASE_HOST": "TC_CI_DB_HOST", "DATABASE_PORT": "TC_CI_DB_PORT", "DATABASE_USER": "TC_CI_DB_USER", "DATABASE_PASSWORD": "TC_CI_DB_PASSWORD", "DATABASE_NAME": "TC_CI_DB_NAME", "REDIS_HOST": "TC_CI_REDIS_HOST", "REDIS_PORT": "TC_CI_REDIS_PORT", "REDIS_DB_NAME": "TC_CI_REDIS_DB", }.items(): if envvar in os.environ and os.environ[envvar] != "": cfg[key] = os.environ[envvar] return { "DATABASE_HOST": cfg.get("DATABASE_HOST", "localhost"), "DATABASE_PORT": int(cfg.get("DATABASE_PORT", "5432")), "DATABASE_USER": cfg.get("DATABASE_USER", "dpb"), "DATABASE_PASSWORD": cfg.get("DATABASE_PASSWORD", ""), "DATABASE_NAME": cfg.get("DATABASE_NAME", "dpb"), "REDIS_HOST": cfg.get("REDIS_HOST", "localhost"), "REDIS_PORT": int(cfg.get("REDIS_PORT", "6379")), "REDIS_DB_NAME": int(cfg.get("REDIS_DB_NAME", "1")), } def _maybe_start_compose() -> bool: """本机 PG/Redis 未就绪时，用 docker compose 拉起测试依赖栈（postgres:16 + redis）。 测试栈端口与宿主机常驻服务错开（5433/6380），预建测试库 dpb_test。 返回是否成功拉起（docker 缺失/失败返回 False，交由 step 0 给出指引）。 """ import shutil if shutil.which("docker") is None: return False compose = os.path.join(ROOT, "docker", "test-compose.yaml") try: subprocess.run( ["docker", "compose", "-f", compose, "up", "-d"], cwd=ROOT, capture_output=True, timeout=120, check=True, ) return True except Exception as exc: # noqa: BLE001 print(f" [warn] docker compose 拉起测试依赖失败: {str(exc)[:120]}") return False async def _probe(cfg: dict) -> tuple[bool, bool, str, str]: import asyncpg import redis.asyncio as aioredis db_ok = redis_ok = False db_err = redis_err = "" try: conn = await asyncpg.connect( host=cfg["DATABASE_HOST"], port=cfg["DATABASE_PORT"], user=cfg["DATABASE_USER"], password=cfg["DATABASE_PASSWORD"] or None, database=cfg["DATABASE_NAME"], timeout=5) await conn.close() db_ok = True except Exception as exc: # noqa: BLE001 db_err = str(exc)[:120] try: # protocol=2 必须与 app.core.database.redis_connect 一致：本机 WSL redis-server 版本较老， # 不支持 redis-py 7.x 默认 RESP3，需走 RESP2。 r = aioredis.Redis(host=cfg["REDIS_HOST"], port=cfg["REDIS_PORT"], db=cfg["REDIS_DB_NAME"], protocol=2, socket_connect_timeout=5) await r.ping() await r.aclose() redis_ok = True except Exception as exc: # noqa: BLE001 redis_err = str(exc)[:120] return db_ok, redis_ok, db_err, redis_err def main() -> int: print("=" * 60) print("本机 CI stage：pytest -m pg（10 套统计引擎正式 tc，真实 PG16 + Redis）") print("=" * 60) # ---- step 0: 探测基础设施 ---- cfg = _env_cfg() db_ok, redis_ok, db_err, redis_err = asyncio.run(_probe(cfg)) print(f"[step 0] 探测 PostgreSQL16 " f"{cfg['DATABASE_HOST']}:{cfg['DATABASE_PORT']}/{cfg['DATABASE_NAME']} → " f"{'OK' if db_ok else 'FAIL ' + db_err}") print(f"[step 0] 探测 Redis {cfg['REDIS_HOST']}:{cfg['REDIS_PORT']}/{cfg['REDIS_DB_NAME']} → " f"{'OK' if redis_ok else 'FAIL ' + redis_err}") if not (db_ok and redis_ok): print("\n[step 0] 本机 PG/Redis 未就绪，尝试 docker compose 拉起测试依赖栈…") if _maybe_start_compose(): # 测试栈端口与常驻服务错开（5433/6380），预建 dpb_test；注入 TC_CI_* 覆盖 os.environ.update({ "TC_CI_DB_HOST": "127.0.0.1", "TC_CI_DB_PORT": "5433", "TC_CI_DB_USER": "dpb", "TC_CI_DB_PASSWORD": "dpb", "TC_CI_DB_NAME": "dpb_test", "TC_CI_REDIS_HOST": "127.0.0.1", "TC_CI_REDIS_PORT": "6380", "TC_CI_REDIS_DB": "1", }) cfg = _env_cfg() # postgres:16 首次初始化可能需数十秒，轮询等健康 for _ in range(15): db_ok, redis_ok, db_err, redis_err = asyncio.run(_probe(cfg)) if db_ok and redis_ok: break import time time.sleep(5) print(f"[step 0] 容器 PG {cfg['DATABASE_HOST']}:{cfg['DATABASE_PORT']}/{cfg['DATABASE_NAME']} → " f"{'OK' if db_ok else 'FAIL ' + db_err}") print(f"[step 0] 容器 Redis {cfg['REDIS_HOST']}:{cfg['REDIS_PORT']}/{cfg['REDIS_DB_NAME']} → " f"{'OK' if redis_ok else 'FAIL ' + redis_err}") if not (db_ok and redis_ok): print("\n❌ 基础设施未就绪，CI 无法放行。请检查：") print(" · 本机 PostgreSQL 16 服务是否启动（服务名 postgresql-x64-16，端口 5432）") print(" · Redis 是否启动（redis-server，端口 6379；工程 WSL 内运行则确认 WSL 已起）") print(f" · 连接配置 backend/env/.env.dev 的 DATABASE_*/REDIS_* 是否与本地一致") print(" · 或安装 docker 后由本脚本自动拉起 docker/test-compose.yaml 测试依赖栈") print(" · 可用 TC_CI_DB_HOST / TC_CI_DB_PORT 等环境变量临时覆盖探测地址") return 1 # ---- step 1: pytest -m pg ---- os.makedirs(LOG_DIR, exist_ok=True) log_path = os.path.join(LOG_DIR, "ci_pytest_mg.log") env = dict(os.environ) env.setdefault("ENVIRONMENT", "dev") env["PYTHONPATH"] = BACKEND env["PYTHONIOENCODING"] = "utf-8" cmd = [PY, "-X", "utf8", "-m", "pytest", "-m", "pg", "-v"] print(f"[step 1] {' '.join(cmd)} (cwd={BACKEND})") print(f" 日志 → {log_path}") with open(log_path, "w", encoding="utf-8") as log: rc = subprocess.call(cmd, cwd=BACKEND, env=env, stdout=log, stderr=subprocess.STDOUT) # ---- step 2: 汇总 ---- with open(log_path, encoding="utf-8") as log: tail = log.readlines()[-25:] for line in tail: print(" " + line.rstrip()) print("=" * 60) if rc == 0: print("✅ CI 全绿：pytest -m pg 10 套统计引擎正式 tc 全部通过") return 0 print(f"❌ CI 失败：pytest -m pg EXIT={rc}（完整输出见 {log_path}）") return rc if __name__ == "__main__": sys.exit(main())