fix(测试): 清理本次创建的隔离资源并保留失败码

记录数据库与账号的创建结果，只清理本次成功创建的对象。
支持显式保留资源，拒绝误删复用库或重名对象，补充退出码回归。

backend/tests/unit/test_isolated_runner.py

@@ -0,0 +1,88 @@
+"""运行器必须保留失败码，并仅清理本次成功创建的对象。"""
+
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.fixture
+def runner(monkeypatch):
+    path = Path(__file__).resolve().parents[3] / "scripts/run_isolated_tests.py"
+    spec = importlib.util.spec_from_file_location("isolated_runner", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    calls, commands = [], []
+
+    def sql(statement):
+        calls.append(statement)
+        return SimpleNamespace(returncode=0)
+
+    def run(command, **kwargs):
+        commands.append(command)
+        return SimpleNamespace(returncode=0)
+
+    monkeypatch.setattr(module, "run_mysql", sql)
+    monkeypatch.setattr(module.subprocess, "run", run)
+    monkeypatch.setattr(module.sys, "argv", ["run_isolated_tests.py"])
+    return module, calls, commands
+
+
+@pytest.mark.parametrize(
+    "args", [[], ["--keep"], ["--reuse-database", "payroll_test_0123456789ab"]]
+)
+def test_失败码不被清理覆盖(runner, monkeypatch, args):
+    module, calls, _ = runner
+    monkeypatch.setattr(module.sys, "argv", ["runner", *args])
+    monkeypatch.setattr(module.subprocess, "run", lambda *_, **__: SimpleNamespace(returncode=5))
+    assert module.main() == 5
+    drops = [sql for sql in calls if sql.startswith("DROP")]
+    assert bool(drops) == ("--keep" not in args)
+    if "--reuse-database" in args:
+        assert not any(sql.startswith("DROP DATABASE") for sql in drops)
+
+
+def test_只清理成功创建的对象(runner, monkeypatch):
+    module, calls, _ = runner
+
+    def sql(statement):
+        calls.append(statement)
+        return SimpleNamespace(returncode=int(statement.startswith("CREATE USER")))
+
+    monkeypatch.setattr(module, "run_mysql", sql)
+    assert module.main() == 1
+    assert any(sql.startswith("DROP DATABASE") for sql in calls)
+    assert not any(sql.startswith("DROP USER") for sql in calls)
+
+
+def test_创建重名库失败绝不删除原库(runner, monkeypatch):
+    module, calls, _ = runner
+
+    def sql(statement):
+        calls.append(statement)
+        return SimpleNamespace(returncode=1)
+
+    monkeypatch.setattr(module, "run_mysql", sql)
+    assert module.main() == 1
+    assert not any(sql.startswith("DROP") for sql in calls)
+
+
+def test_参数分隔符不传给pytest(runner, monkeypatch):
+    module, _, commands = runner
+    monkeypatch.setattr(module.sys, "argv", ["runner", "--", "tests/integration/test_auth_flow.py"])
+    assert module.main() == 0
+    assert "--" not in commands[-1]
+    assert "tests/integration/test_auth_flow.py" in commands[-1]
+
+
+def test_清理失败不能报成功(runner, monkeypatch):
+    module, calls, _ = runner
+
+    def sql(statement):
+        calls.append(statement)
+        return SimpleNamespace(returncode=int(statement.startswith("DROP DATABASE")))
+
+    monkeypatch.setattr(module, "run_mysql", sql)
+    assert module.main() == 1
+    assert any(sql.startswith("DROP USER") for sql in calls)

scripts/run_isolated_tests.py

@@ -1,7 +1,7 @@
 """使用唯一的本地测试库运行迁移和集成测试，绝不清理原业务库。"""
 
-import os
 import argparse
+import os
 import re
 import secrets
 import subprocess
@@ -9,43 +9,99 @@ import sys
 from pathlib import Path
 
 
+def run_mysql(sql: str) -> subprocess.CompletedProcess:
+    """密码仅在容器环境和标准输入中使用，不回显 SQL 错误内容。"""
+    return subprocess.run(
+        [
+            "docker",
+            "exec",
+            "-i",
+            "ys-mysql-dev",
+            "sh",
+            "-c",
+            'export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"; exec mysql -uroot',
+        ],
+        input=sql.encode(),
+        capture_output=True,
+    )
+
+
 def main():
     parser = argparse.ArgumentParser()
-    parser.add_argument('--reuse-database')
+    parser.add_argument("--reuse-database")
-    parser.add_argument('--migration-roundtrip', action='store_true')
+    parser.add_argument("--migration-roundtrip", action="store_true")
+    parser.add_argument("--keep", action="store_true", help="保留本次隔离库和账号供复查")
     parsed, test_args = parser.parse_known_args()
-    if parsed.reuse_database and not re.fullmatch(r'payroll_test_[0-9a-f]{12}', parsed.reuse_database):
+    if test_args[:1] == ["--"]:
-        raise SystemExit('只允许复用本运行器创建的隔离测试库')
+        test_args = test_args[1:]
+    if parsed.reuse_database and not re.fullmatch(
+        r"payroll_test_[0-9a-f]{12}", parsed.reuse_database
+    ):
+        raise SystemExit("只允许复用本运行器创建的隔离测试库")
     root = Path(__file__).resolve().parents[1]
-    python = root / "backend/.venv-workbench/Scripts/python.exe"
+    python = Path(sys.executable)
-    if not python.exists():
+    if sys.prefix == sys.base_prefix:
-        python = Path(sys.executable)
+        for candidate in (".venv", ".venv-workbench"):
+            local_python = root / "backend" / candidate / "Scripts/python.exe"
+            if local_python.exists():
+                python = local_python
+                break
     suffix = secrets.token_hex(6)
-    database, username = parsed.reuse_database or "payroll_test_" + suffix, "pytest_" + suffix
+    database = parsed.reuse_database or "payroll_test_" + suffix
+    username = "pytest_" + suffix
     password = secrets.token_urlsafe(30)
-    sql = ("" if parsed.reuse_database else f"CREATE DATABASE `{database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; ") + f"CREATE USER '{username}'@'%' IDENTIFIED BY '{password}'; GRANT ALL PRIVILEGES ON `{database}`.* TO '{username}'@'%';"
+    created_database = created_user = False
-    setup = subprocess.run(
+    result_code = 1
-        ["docker", "exec", "-i", "ys-mysql-dev", "sh", "-c", 'export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"; exec mysql -uroot'],
+    try:
-        input=sql.encode(), capture_output=True,
+        if not parsed.reuse_database:
-    )
+            if run_mysql(
-    if setup.returncode:
+                f"CREATE DATABASE `{database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;"
-        raise SystemExit("隔离库创建失败，未执行迁移或测试")
+            ).returncode:
-    env = os.environ | {
+                raise RuntimeError("隔离库创建失败，未执行迁移或测试")
-        "MYSQL_DATABASE": database, "MYSQL_USER": username, "MYSQL_PASSWORD": password,
+            created_database = True
-        "PAYROLL_TEST_DATABASE": database, "REDIS_DB": "15", "PYTHONIOENCODING": "utf-8",
+        if run_mysql(f"CREATE USER '{username}'@'%' IDENTIFIED BY '{password}';").returncode:
-        "UPLOAD_DIR": str(root / ".ai" / "isolated-uploads" / database),
+            raise RuntimeError("隔离账号创建失败，未执行迁移或测试")
-    }
+        created_user = True
-    print("独立测试库：" + database, flush=True)
+        if run_mysql(f"GRANT ALL PRIVILEGES ON `{database}`.* TO '{username}'@'%';").returncode:
-    commands = [["-m", "alembic", "upgrade", "head"]]
+            raise RuntimeError("隔离账号授权失败，未执行迁移或测试")
-    if parsed.migration_roundtrip:
+        env = os.environ | {
-        commands += [["-m", "alembic", "downgrade", "07b8c9d0e1f2"], ["-m", "alembic", "upgrade", "head"]]
+            "MYSQL_DATABASE": database,
-    args = test_args
+            "MYSQL_USER": username,
-    commands.append(["-m", "pytest", *(args or ["tests/integration"]), "-q", "--tb=short"])
+            "MYSQL_PASSWORD": password,
-    for args in commands:
+            "PAYROLL_TEST_DATABASE": database,
-        result = subprocess.run([str(python), *args], cwd=root / "backend", env=env)
+            "REDIS_DB": "15",
-        if result.returncode:
+            "PYTHONIOENCODING": "utf-8",
-            return result.returncode
+            "UPLOAD_DIR": str(root / ".ai" / "isolated-uploads" / database),
-    return 0
+        }
+        print("独立测试库：" + database, flush=True)
+        commands = [["-m", "alembic", "upgrade", "head"]]
+        if parsed.migration_roundtrip:
+            commands += [
+                ["-m", "alembic", "downgrade", "07b8c9d0e1f2"],
+                ["-m", "alembic", "upgrade", "head"],
+            ]
+        commands.append(["-m", "pytest", *(test_args or ["tests/integration"]), "-q", "--tb=short"])
+        for args in commands:
+            result_code = subprocess.run(
+                [str(python), *args], cwd=root / "backend", env=env
+            ).returncode
+            if result_code:
+                break
+    except RuntimeError as exc:
+        print(str(exc), file=sys.stderr)
+    finally:
+        # 只清理本次 CREATE 已成功的对象；重名创建失败及复用库绝不能被删除。
+        if not parsed.keep:
+            cleanup = []
+            if created_database:
+                cleanup.append(f"DROP DATABASE `{database}`;")
+            if created_user:
+                cleanup.append(f"DROP USER '{username}'@'%';")
+            for statement in cleanup:
+                if run_mysql(statement).returncode:
+                    print(f"隔离测试资源清理失败，请核对 {database} / {username}", file=sys.stderr)
+                    result_code = result_code or 1
+    return result_code
 
 
 if __name__ == "__main__":