diff --git a/.gitignore b/.gitignore index 3b19b47..b5b18cf 100644 --- a/.gitignore +++ b/.gitignore @@ -97,3 +97,6 @@ test-results/ # Agent 团队运行态（会话产物，非产品资产） .agent-teams/ + +# 统一测试入口（Scripts/bin/run_all_tests.py）的 per-suite basetemp +.pytest-cache/ diff --git a/Scripts/bin/run_all_tests.py b/Scripts/bin/run_all_tests.py new file mode 100644 index 0000000..0a4c5c7 --- /dev/null +++ b/Scripts/bin/run_all_tests.py @@ -0,0 +1,472 @@ +#!/usr/bin/env python3 +# coding=utf-8 +"""统一测试入口 —— 一条命令跑全部套件. + +## 为什么需要它 + +本机开发有 5 个独立 venv（各服务的 bundle 部署需要），跑一次全量测试要： + +1. 记住每个套件该用哪个解释器（`findatahub-core` 只装在部分 venv 里）； +2. 记住 `findatahub_core` 以**非 editable** 方式装进各 venv， + 不注入 `PYTHONPATH` 就会验到**已安装的旧版本**，源码改动不生效； +3. 记住 `Scripts/bin/tests` 里 `test_rss_fetcher.py` 需要 `httpx`， + 选错 venv 会在收集阶段直接报错； +4. 记住 **不能并行跑两个 pytest 进程**——它们会互踩系统临时目录， + 报 `FileExistsError: pytest-of-root`，表现为几十个 errors 的假回归。 + +这四条中任何一条被忘记，得到的结论就是错的。摩擦本身就是回归风险的来源， +所以把它固化成脚本。 + +## 关键设计 + +- **每个套件独立 `--basetemp`**：放进项目内 `.pytest-cache//`， + 从根上消除 tmpdir 争用，不依赖"记得串行"。也不去删系统临时目录。 +- **解释器按能力选，不写死**：每个套件声明它需要能 import 哪些模块， + 脚本在候选 venv 里挑第一个满足的；挑不到就明确报错，而不是用错的 venv 硬跑。 +- **强制注入 `PYTHONPATH`**：把源码目录排在已安装包之前，避免验旧版本。 +- **纯 Python、跨平台**：不依赖 bash，Windows 上同样可用 + （与 `paths.py` / `process_utils.py` 的迁移方向一致）。 + +## 用法 + + python3 Scripts/bin/run_all_tests.py # 全部 Python 套件 + python3 Scripts/bin/run_all_tests.py --list # 只列出套件与选中的解释器 + python3 Scripts/bin/run_all_tests.py -k newsflash # 透传给 pytest 的 -k + python3 Scripts/bin/run_all_tests.py --only core,backend + python3 Scripts/bin/run_all_tests.py --include-swift --include-frontend + +退出码：全部通过 → 0；任一失败 → 1；环境不满足（找不到解释器）→ 2。 +""" + +from __future__ import annotations + +import argparse +import os +import re +import shutil +import subprocess +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +# 所有套件共用同一个 basetemp 根，但各自一个子目录（互不干扰） +BASETEMP_ROOT = ROOT / ".pytest-cache" + + +# ══════════════════════════════════════════════════════════ +# 套件声明 +# ══════════════════════════════════════════════════════════ + +@dataclass +class Suite: + """一个可独立执行的 pytest 套件。""" + + key: str + name: str + cwd: Path # 工作目录（相对 ROOT） + target: str # pytest 目标（相对 cwd） + # 需要排到 PYTHONPATH 最前面的源码目录（相对 ROOT） + source_paths: list[Path] = field(default_factory=list) + # 必须能 import 的模块；脚本据此挑选解释器 + requires: list[str] = field(default_factory=lambda: ["pytest"]) + note: str = "" + + +# 顺序即执行顺序。高频、纯离线的放前面，出错能更早暴露。 +SUITES: list[Suite] = [ + Suite( + key="core", + name="findatahub-core（数据层基类）", + cwd=ROOT / "Services/findatahub-core", + target="tests", + # 必须排在最前：该包以非 editable 装进各 venv， + # 否则测到的是已安装的旧版本（历史踩坑） + source_paths=[ROOT / "Services/findatahub-core/src"], + note="含 replace 收缩熔断回归", + ), + Suite( + key="tushare", + name="tushare_fetcher（采集器）", + cwd=ROOT / "Services/tushare_fetcher", + target="tests/unit", + source_paths=[ROOT / "Services/findatahub-core/src"], + requires=["pytest", "findatahub_core"], + note="仅 unit：integration 需要真实 Tushare 额度", + ), + Suite( + key="akshare", + name="akshare_fetcher（采集器）", + cwd=ROOT / "Services/akshare_fetcher", + target="tests", + source_paths=[ROOT / "Services/findatahub-core/src"], + requires=["pytest", "findatahub_core"], + ), + Suite( + key="client", + name="findatahub-client（SDK）", + cwd=ROOT / "Services/findatahub-client", + target="tests", + # 测试 import 的是 `findatahub.client`（命名空间包）， + # 必须把 client 自己的 src 排进 PYTHONPATH： + # 只写 findatahub_core 的话，akshare 的 venv 也能"满足依赖"却跑不起来 + source_paths=[ + ROOT / "Services/findatahub-core/src", + ROOT / "Services/findatahub-client/src", + ], + requires=["pytest", "findatahub.client"], + ), + Suite( + key="scripts", + name="Scripts/bin（脚本层）", + cwd=ROOT / "Scripts/bin", + target="tests", + # 该套件里 rss_fetcher 需要 httpx；缺了会在收集阶段直接报错 + requires=["pytest", "httpx", "pymysql"], + note="含跨平台抽象层回归", + ), + Suite( + key="backend", + name="kuaixun-service（后端）", + cwd=ROOT / "Services/kuaixun-service", + target="tests", + requires=["pytest", "fastapi", "httpx"], + note="含 CORS / 时间基准回归", + ), +] + + +# ══════════════════════════════════════════════════════════ +# 解释器探测 +# ══════════════════════════════════════════════════════════ + +def _venv_python(venv_dir: Path) -> Path | None: + """返回 venv 内的解释器路径（兼容 POSIX 与 Windows 布局）。""" + for rel in ("bin/python", "bin/python3", "Scripts/python.exe"): + p = venv_dir / rel + if p.exists(): + return p + return None + + +def candidate_interpreters() -> list[tuple[str, Path]]: + """候选解释器，按优先级排列：各服务自带 venv 在前，当前解释器兜底。""" + cands: list[tuple[str, Path]] = [] + for venv in sorted((ROOT / "Services").glob("*/.venv")): + py = _venv_python(venv) + if py: + cands.append((f"{venv.parent.name}/.venv", py)) + cands.append(("当前解释器", Path(sys.executable))) + return cands + + +def _can_import(python: Path, module: str, pythonpath: str) -> bool: + """探测某解释器能否 import 指定模块（含注入的 PYTHONPATH）。""" + env = dict(os.environ) + if pythonpath: + env["PYTHONPATH"] = pythonpath + try: + res = subprocess.run( + [str(python), "-c", f"import {module}"], + capture_output=True, timeout=30, env=env, cwd=str(ROOT), + ) + return res.returncode == 0 + except (OSError, subprocess.SubprocessError): + return False + + +def pick_interpreter(suite: Suite, cands: list[tuple[str, Path]]) -> tuple[str, Path] | None: + """选第一个能满足 suite.requires 的解释器。""" + pythonpath = os.pathsep.join(str(p) for p in suite.source_paths) + for label, py in cands: + if all(_can_import(py, m, pythonpath) for m in suite.requires): + return label, py + return None + + +# ══════════════════════════════════════════════════════════ +# 执行 +# ══════════════════════════════════════════════════════════ + +@dataclass +class Result: + suite: Suite + interpreter: str + returncode: int + seconds: float + summary: str + output: str + + +def _ensure_dir(path: Path) -> None: + """建目录，容忍已存在。 + + 不直接用 ``mkdir(exist_ok=True)``：本机沙箱对 mkdir 做了拦截， + 即使目标已存在也可能抛 ``FileExistsError``（本项目已多次踩到）。 + 先判 is_dir 再兜住异常，两条路径都安全。 + """ + if path.is_dir(): + return + try: + path.mkdir(parents=True, exist_ok=True) + except FileExistsError: + pass + + +def run_suite(suite: Suite, label: str, python: Path, extra_args: list[str]) -> Result: + """执行单个套件。""" + basetemp = BASETEMP_ROOT / suite.key + # pytest 会先清空 --basetemp 目标目录；它位于项目内，删的是自己的缓存 + _ensure_dir(basetemp) + + env = dict(os.environ) + if suite.source_paths: + env["PYTHONPATH"] = os.pathsep.join(str(p) for p in suite.source_paths) + # 统一 UTF-8：Windows 控制台默认 cp936，中文用例名会乱码甚至报编码错 + env["PYTHONUTF8"] = "1" + env["PYTHONIOENCODING"] = "utf-8" + + cmd = [ + str(python), "-m", "pytest", suite.target, + "-q", "--basetemp", str(basetemp), + *extra_args, + ] + + started = time.time() + try: + proc = subprocess.run( + cmd, cwd=str(suite.cwd), env=env, + capture_output=True, text=True, timeout=1800, + ) + elapsed = time.time() - started + stdout = proc.stdout or "" + stderr = proc.stderr or "" + output = stdout + stderr + return Result(suite, label, proc.returncode, elapsed, + _extract_summary(stdout, stderr), output) + except subprocess.TimeoutExpired: + return Result(suite, label, 124, time.time() - started, + "超时（1800s）", "") + except OSError as e: + return Result(suite, label, 125, time.time() - started, + f"启动失败：{e}", "") + + +# pytest 统计行的形态："132 passed, 2 skipped in 3.55s" / "1 error in 0.25s" +# 需覆盖的边界： +# - "-k 过滤到零用例" → "134 deselected in 0.32s"（exit 5，无 passed/failed） +# - "no tests ran"：同样 exit 5 +# - xfailed / xpassed：backend 套件有 xfail +_SUMMARY_RE = re.compile( + r"\b\d+ (?:passed|failed|error|errors|skipped|deselected|xfailed|xpassed" + r"|warning|warnings|rerun|reruns)\b|no tests ran" +) + + +def _extract_summary(stdout: str, stderr: str = "") -> str: + """从 pytest 输出里取统计行。 + + 必须**优先看 stdout、且要求「数字 + 状态词」相邻**： + 应用自身的 stderr 日志里常出现形如 + ``(Background on this error at: https://sqlalche.me/...)`` 的行， + 只按 "error" 关键词匹配会把这种日志误当成统计结果 + （首版脚本就把 backend 套件的摘要取成了 SQLAlchemy 的报错提示）。 + """ + for text in (stdout, stderr): + for line in reversed(text.strip().splitlines()): + s = line.strip().strip("=· ") + if s and _SUMMARY_RE.search(s): + return s + return "（无统计输出）" + + +# ══════════════════════════════════════════════════════════ +# Swift / 前端（可选） +# ══════════════════════════════════════════════════════════ + +def run_swift(extra_args: list[str]) -> Result: + suite = Suite(key="swift", name="Swift 单元测试", cwd=ROOT, target="Tests") + started = time.time() + if not shutil.which("swift"): + return Result(suite, "-", 127, 0, "未安装 swift", "") + # SwiftPM 自身要调 sandbox-exec，本机沙箱下必须 --disable-sandbox + cmd = ["swift", "test", "--disable-sandbox", *extra_args] + try: + proc = subprocess.run(cmd, cwd=str(ROOT), capture_output=True, text=True, timeout=1800) + out = (proc.stdout or "") + (proc.stderr or "") + executed = [l for l in out.splitlines() if "Executed" in l and "failure" in l] + summary = executed[-1].strip() if executed else ("通过" if proc.returncode == 0 else "失败") + return Result(suite, "swift", proc.returncode, time.time() - started, summary, out) + except (OSError, subprocess.SubprocessError) as e: + return Result(suite, "swift", 125, time.time() - started, f"启动失败：{e}", "") + + +# ANSI 转义：vitest / pytest 在管道输出里也可能带颜色码， +# 不去掉会破坏行首匹配（首版前端摘要因此退化成"通过"，看不到真实数量） +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + + +def run_frontend(extra_args: list[str]) -> Result: + cwd = ROOT / "Services/kuaixun-frontend" + suite = Suite(key="frontend", name="前端（vitest）", cwd=cwd, target="tests") + started = time.time() + npx = shutil.which("npx") + if not npx or not (cwd / "node_modules").exists(): + return Result(suite, "-", 127, 0, "未安装 npx 或 node_modules", "") + cmd = [npx, "vitest", "run", "--reporter=dot", *extra_args] + try: + proc = subprocess.run(cmd, cwd=str(cwd), capture_output=True, text=True, timeout=1800) + out = _ANSI_RE.sub("", (proc.stdout or "") + (proc.stderr or "")) + # vitest 汇总行形如 " Tests 44 passed (44)" + summary = "" + for line in out.splitlines(): + if re.match(r"^\s*Tests\s+\d+", line): + summary = line.strip() + if not summary: + summary = "通过" if proc.returncode == 0 else "失败" + return Result(suite, "npx", proc.returncode, time.time() - started, summary, out) + except (OSError, subprocess.SubprocessError) as e: + return Result(suite, "npx", 125, time.time() - started, f"启动失败：{e}", "") + + +# ══════════════════════════════════════════════════════════ +# 入口 +# ══════════════════════════════════════════════════════════ + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="统一测试入口：一条命令跑全部套件", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="示例：\n %(prog)s\n %(prog)s --list\n %(prog)s -k newsflash\n" + " %(prog)s --only core,backend --include-swift", + ) + parser.add_argument("--list", action="store_true", help="只列出套件与选中的解释器") + parser.add_argument("--only", default="", help="只跑指定套件，逗号分隔（key）") + parser.add_argument("--skip", default="", help="跳过指定套件，逗号分隔（key）") + parser.add_argument("--include-swift", action="store_true", help="同时跑 Swift 测试") + parser.add_argument("--include-frontend", action="store_true", help="同时跑前端测试") + parser.add_argument( + "--with-integration", action="store_true", + help="同时跑需要真实 MySQL 的集成测试（会自建临时库，需 MYSQL_PASSWORD）", + ) + parser.add_argument("--verbose", action="store_true", + help="失败时打印完整输出（刻意不设 -v：留给 pytest 的 -v）") + # 未知参数一律透传给 pytest。 + # 不用 nargs=REMAINDER：argparse 在 "先有 --only、后有 -k" 的组合下 + # 会把 -k 当成本脚本的未知选项而直接退出 2（已实测）。 + args, extra = parser.parse_known_args(argv) + extra = [a for a in extra if a != "--"] + + only = {s.strip() for s in args.only.split(",") if s.strip()} + skip = {s.strip() for s in args.skip.split(",") if s.strip()} + + selected = [s for s in SUITES + if (not only or s.key in only) and s.key not in skip] + + unknown = only - {s.key for s in SUITES} + if unknown: + print(f"未知套件：{', '.join(sorted(unknown))}；可用：" + f"{', '.join(s.key for s in SUITES)}", file=sys.stderr) + return 2 + + cands = candidate_interpreters() + + # ── 选解释器 ── + planned: list[tuple[Suite, str, Path]] = [] + unresolved: list[Suite] = [] + for suite in selected: + picked = pick_interpreter(suite, cands) + if picked is None: + unresolved.append(suite) + else: + planned.append((suite, picked[0], picked[1])) + + if args.list: + print(f"项目根：{ROOT}\n") + print(f"{'套件':<10} {'解释器':<28} {'需要模块'}") + print("-" * 78) + for suite in selected: + picked = pick_interpreter(suite, cands) + label = picked[0] if picked else "❌ 无可用解释器" + print(f"{suite.key:<10} {label:<28} {', '.join(suite.requires)}") + if suite.note: + print(f"{'':<10} └─ {suite.note}") + return 0 if not unresolved else 2 + + if unresolved: + print("以下套件找不到满足依赖的解释器：", file=sys.stderr) + for suite in unresolved: + print(f" - {suite.key}（{suite.name}）：需要 " + f"{', '.join(suite.requires)}", file=sys.stderr) + print("\n可尝试：在对应服务目录执行 `uv sync` 补齐依赖。", file=sys.stderr) + return 2 + + # ── 串行执行 ── + _ensure_dir(BASETEMP_ROOT) + if args.with_integration and not os.environ.get("MYSQL_PASSWORD"): + print( + "⚠️ 已启用 --with-integration，但未设置 MYSQL_PASSWORD：\n" + " 集成测试会整体 skip（不算失败）。如需真跑：\n" + " set -a; . ~/.config/findatahub/env; set +a\n", + file=sys.stderr, + ) + results: list[Result] = [] + total = len(planned) + int(args.include_swift) + int(args.include_frontend) + idx = 0 + + print(f"项目根：{ROOT}") + print(f"套件数：{total}（每个套件独立 --basetemp，天然互不干扰）\n") + + for suite, label, python in planned: + idx += 1 + print(f"[{idx}/{total}] {suite.name} … ", end="", flush=True) + # 集成测试默认排除：它要真实 MySQL，且会建临时库。 + # 只对 core 套件生效（目前唯一含 integration 子目录的套件）。 + suite_extra = list(extra) + if suite.key == "core" and not args.with_integration: + suite_extra.append("--ignore=tests/integration") + r = run_suite(suite, label, python, suite_extra) + results.append(r) + status = "✓" if r.returncode == 0 else "✗" + print(f"{status} {r.summary} ({r.seconds:.1f}s, {r.interpreter})") + + if args.include_swift: + idx += 1 + print(f"[{idx}/{total}] Swift 单元测试 … ", end="", flush=True) + r = run_swift(extra) + results.append(r) + print(f"{'✓' if r.returncode == 0 else '✗'} {r.summary} ({r.seconds:.1f}s)") + + if args.include_frontend: + idx += 1 + print(f"[{idx}/{total}] 前端（vitest）… ", end="", flush=True) + r = run_frontend(extra) + results.append(r) + print(f"{'✓' if r.returncode == 0 else '✗'} {r.summary} ({r.seconds:.1f}s)") + + # ── 汇总 ── + failed = [r for r in results if r.returncode != 0] + print("\n" + "=" * 78) + print(f"{'套件':<12} {'状态':<6} {'耗时':>8} 统计") + print("-" * 78) + for r in results: + status = "PASS" if r.returncode == 0 else "FAIL" + print(f"{r.suite.key:<12} {status:<6} {r.seconds:>7.1f}s {r.summary}") + print("=" * 78) + + if failed: + print(f"\n❌ {len(failed)} 个套件失败：{', '.join(r.suite.key for r in failed)}") + for r in failed: + if args.verbose and r.output: + print(f"\n───── {r.suite.key} 输出 ─────") + print(r.output[-6000:]) + return 1 + + total_seconds = sum(r.seconds for r in results) + print(f"\n✅ 全部 {len(results)} 个套件通过（合计 {total_seconds:.1f}s）") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/Services/findatahub-core/tests/integration/__init__.py b/Services/findatahub-core/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/Services/findatahub-core/tests/integration/test_replace_real_mysql.py b/Services/findatahub-core/tests/integration/test_replace_real_mysql.py new file mode 100644 index 0000000..08cb6e0 --- /dev/null +++ b/Services/findatahub-core/tests/integration/test_replace_real_mysql.py @@ -0,0 +1,291 @@ +# coding=utf-8 +"""`replace()` 的真实 MySQL 集成测试. + +## 为什么必须有这一套 + +`test_db_replace_guard.py` 用 Fake 游标验证了**代码路径**，但 Fake 无法验证三件 +只有真库才成立的事： + +1. **事务是否真的回滚**——Fake 的 `rollback()` 只是置一个标志位，真实行为取决于 + InnoDB 的 undo 与隔离级别； +2. **唯一的"半成品"风险**——旧实现每个 chunk 独立 `commit()`，中途失败会留下 + 「旧数据已删、新数据写了一半」。这个场景在 Fake 上永远复现不出来， + 因为 Fake 不会真的提交。 +3. **熔断时表是否真的没动**——需要真实读取行数来确认。 + +这三件事恰好是 `replace()` 最危险的三个点，因此本套件用真实 MySQL 验证。 + +## 隔离约束（硬性） + +- **绝不连生产库**：每次在 fixture 内新建 `fdh_it_replace_<随机8位>` 空库， + 用完 `DROP`。库名带前缀且 DROP 前再断言一次前缀。 +- 不依赖任何现有库、现有表、现有凭据文件；只读环境变量。 +- 没有可用凭据时整组 `skip`，不报失败——CI 无 MySQL 时不应因此变红。 + +## 运行方式 + + export MYSQL_PASSWORD='...' # 或从 ~/.config/findatahub/env 取 + PYTHONPATH=src python -m pytest tests/integration -q + +可选环境变量：`MYSQL_HOST`（默认 127.0.0.1）、`MYSQL_USER`（默认 root）。 +""" + +from __future__ import annotations + +import os +import uuid + +import pandas as pd +import pytest + +from findatahub_core.db import BaseDatabaseManager, ReplaceShrinkError + + +# ══════════════════════════════════════════════════════════ +# 凭据与跳过 +# ══════════════════════════════════════════════════════════ + +def _credentials() -> dict: + password = os.environ.get("MYSQL_PASSWORD", "") + if not password: + pytest.skip( + "未提供 MYSQL_PASSWORD，跳过真实库集成测试" + "（本套件会自建临时库，不会触碰生产数据）" + ) + return { + "host": os.environ.get("MYSQL_HOST", "127.0.0.1"), + "port": int(os.environ.get("MYSQL_PORT", "3306")), + "user": os.environ.get("MYSQL_USER", "root"), + "password": password, + } + + +# ══════════════════════════════════════════════════════════ +# 测试用 DatabaseManager +# ══════════════════════════════════════════════════════════ + +class _ITManager(BaseDatabaseManager): + """最小子类：只声明测试表，不读任何配置文件。 + + `replace()` 不走唯一索引配置（那是 `upsert()` 才需要的）， + 因此 `_get_index_config_path` 返回空串即可。 + """ + + def _get_index_config_path(self) -> str: + return "" + + def _get_safe_tables(self) -> set[str]: + return {"it_rows"} + + +CREATE_TABLE_SQL = """ +CREATE TABLE `it_rows` ( + `ts_code` VARCHAR(16) NOT NULL, + `trade_date` CHAR(8) NOT NULL, + `close` DOUBLE DEFAULT NULL, + UNIQUE KEY `uk_code_date` (`ts_code`, `trade_date`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 +""" + + +def _make_df(rows: int, *, duplicate_row: int | None = None) -> pd.DataFrame: + """构造测试数据。 + + Args: + rows: 行数。 + duplicate_row: 若指定，把该行的 (ts_code, trade_date) 改成与第 0 行相同， + 用于制造唯一索引冲突——这是验证「跨 chunk 回滚」的关键手段。 + """ + codes = [f"{i:06d}.SZ" for i in range(rows)] + dates = ["20260101"] * rows + if duplicate_row is not None and 0 < duplicate_row < rows: + codes[duplicate_row] = codes[0] + dates[duplicate_row] = dates[0] + return pd.DataFrame({ + "ts_code": codes, + "trade_date": dates, + "close": [10.0 + i * 0.01 for i in range(rows)], + }) + + +def _count_rows(manager: _ITManager) -> int: + """直连统计表内实际行数（不复用被测代码，避免自证）。""" + raw = manager.get_engine().raw_connection() + try: + cur = raw.cursor() + cur.execute("SELECT COUNT(*) FROM `it_rows`") + return int(cur.fetchone()[0]) + finally: + raw.close() + + +# ══════════════════════════════════════════════════════════ +# Fixture：独立临时库 +# ══════════════════════════════════════════════════════════ + +@pytest.fixture(scope="module") +def manager(): + """建一个独立临时库 + 测试表，模块结束后整库删除。""" + import pymysql + + creds = _credentials() + dbname = f"fdh_it_replace_{uuid.uuid4().hex[:8]}" + assert dbname.startswith("fdh_it_replace_"), "库名前缀必须受控" + + admin = pymysql.connect( + host=creds["host"], port=creds["port"], + user=creds["user"], password=creds["password"], + charset="utf8mb4", autocommit=True, + ) + try: + with admin.cursor() as cur: + cur.execute( + f"CREATE DATABASE `{dbname}` " + "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" + ) + except Exception as e: # noqa: BLE001 + admin.close() + pytest.skip(f"无法创建测试库（MySQL 不可用？）：{e}") + + mgr = _ITManager({ + "host": creds["host"], "port": creds["port"], + "user": creds["user"], "password": creds["password"], + "database": dbname, "charset": "utf8mb4", + }) + + # 建表 + raw = mgr.get_engine().raw_connection() + try: + cur = raw.cursor() + cur.execute(CREATE_TABLE_SQL) + raw.commit() + finally: + raw.close() + + try: + yield mgr + finally: + # 全库删除：只删自己建的库，且再校验一次前缀 + if dbname.startswith("fdh_it_replace_"): + try: + with admin.cursor() as cur: + cur.execute(f"DROP DATABASE IF EXISTS `{dbname}`") + except Exception: # noqa: BLE001 — 清理失败不应掩盖测试结论 + pass + try: + mgr.get_engine().dispose() + except Exception: # noqa: BLE001 + pass + admin.close() + + +@pytest.fixture(autouse=True) +def _clean_table(manager): + """每个用例前清空表，保证用例间独立。""" + raw = manager.get_engine().raw_connection() + try: + cur = raw.cursor() + cur.execute("DELETE FROM `it_rows`") + raw.commit() + finally: + raw.close() + yield + + +# ══════════════════════════════════════════════════════════ +# 用例 +# ══════════════════════════════════════════════════════════ + +class TestRealReplace: + def test_normal_write_on_empty_table(self, manager): + """空表写入不受熔断影响（原表 0 行 < 阈值）。""" + assert _count_rows(manager) == 0 + n = manager.replace(_make_df(100), "it_rows") + assert n == 100 + assert _count_rows(manager) == 100 + + def test_large_table_replace_succeeds_when_size_similar(self, manager): + """正常全量刷新（行数相当）必须放行——熔断不能误伤日常路径。""" + manager.replace(_make_df(1000), "it_rows") + assert _count_rows(manager) == 1000 + + n = manager.replace(_make_df(900), "it_rows") + assert n == 900 + assert _count_rows(manager) == 900 + + +class TestShrinkGuardOnRealDB: + def test_rejects_shrink_and_leaves_table_untouched(self, manager): + """核心断言：熔断时表**必须原样**，旧数据一行不少。 + + 这是 replace 最危险的失败模式——用不完整批次替换完整历史。 + """ + manager.replace(_make_df(1000), "it_rows") + assert _count_rows(manager) == 1000 + + with pytest.raises(ReplaceShrinkError) as ei: + manager.replace(_make_df(10), "it_rows") + + assert ei.value.existing == 1000 + assert ei.value.incoming == 10 + # 真库复核：数据完好，删除动作从未发生 + assert _count_rows(manager) == 1000, "熔断后表被改动，护栏失效" + + def test_allow_shrink_replaces_for_real(self, manager): + """显式承认收缩时，表内容必须真的被替换。""" + manager.replace(_make_df(1000), "it_rows") + n = manager.replace(_make_df(5), "it_rows", allow_shrink=True) + assert n == 5 + assert _count_rows(manager) == 5 + + def test_small_table_not_guarded(self, manager): + """低于 REPLACE_GUARD_MIN_ROWS 的表不做比例校验。""" + manager.replace(_make_df(50), "it_rows") + n = manager.replace(_make_df(1), "it_rows") + assert n == 1 + assert _count_rows(manager) == 1 + + +class TestAtomicityOnRealDB: + def test_cross_chunk_failure_rolls_back_entirely(self, manager): + """跨 chunk 失败必须整体回滚——旧实现在这里会留下半张表。 + + 构造方式：写入 15000 行（2 个 chunk：10000 + 5000）， + 令第 10001 行与第 1 行的唯一键相同。 + 于是：chunk1 成功、chunk2 冲突失败。 + + 旧实现（每 chunk 独立 commit）的后果： + DELETE 已提交 + chunk1 已提交 → 表内 10000 行， + 原来的 1000 行历史**永久丢失**。 + 新实现（单事务）： + 全部回滚 → 表内仍是 1000 行。 + """ + manager.replace(_make_df(1000), "it_rows") + assert _count_rows(manager) == 1000 + + bad = _make_df(15000, duplicate_row=10000) + with pytest.raises(Exception) as ei: + manager.replace(bad, "it_rows", allow_shrink=True) + + # 唯一键冲突的错误信息因驱动版本而异，只断言"确实失败了" + assert ei.value is not None + + # 核心断言：旧数据完好，且没有半成品残留 + assert _count_rows(manager) == 1000, ( + "回滚不完整：表内行数既不是旧值 1000，说明留下了半成品数据" + ) + + def test_empty_df_does_not_truncate(self, manager): + """空批次绝不清表——「本次没取到数」不等于「表应该为空」。""" + manager.replace(_make_df(200), "it_rows") + assert _count_rows(manager) == 200 + + assert manager.replace(pd.DataFrame(), "it_rows") == 0 + assert _count_rows(manager) == 200 + + def test_repeated_replace_is_idempotent(self, manager): + """连续两次相同内容的全量写入，结果一致（不会累积重复行）。""" + df = _make_df(300) + manager.replace(df, "it_rows") + manager.replace(df, "it_rows") + assert _count_rows(manager) == 300