fix(tplus): 统一兼容检查连接策略并默认启用加密

backend/tests/unit/test_tplus_compatibility_script.py

@@ -0,0 +1,92 @@
+"""兼容性脚本与应用使用相同连接策略，失败报告不泄露连接数据。"""
+
+import importlib.util
+import json
+import sys
+from datetime import datetime
+from decimal import Decimal
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.core.config import Settings
+
+
+@pytest.fixture
+def script(monkeypatch):
+    # 纯单元测试不需要安装 ODBC 驱动或连接源库。
+    monkeypatch.setitem(sys.modules, "pyodbc", MagicMock())
+    path = Path(__file__).resolve().parents[3] / "scripts/tplus/check_compatibility.py"
+    spec = importlib.util.spec_from_file_location("compatibility_script", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    settings = Settings(_env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True)
+    monkeypatch.setattr(module, "get_settings", lambda: settings)
+    return module
+
+
+def test_脚本复用连接文件校验与加密设置(script, tmp_path):
+    path = tmp_path / "reader.json"
+    payload = {
+        "host": "localhost",
+        "port": 1433,
+        "database": "data;x}",
+        "user": "reader;other}",
+        "password": "test-only;x}",
+        "driver": "TestDriver",
+    }
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    resolved = script.compatibility_settings(path)
+    assert resolved.t_plus_database_url.database == payload["database"]
+    assert resolved.t_plus_database_url.username == payload["user"]
+    assert resolved.t_plus_database_url.query["Encrypt"] == "yes"
+    assert script.get_settings().t_plus_enabled is False
+    payload["user"] = "sa"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    with pytest.raises(RuntimeError, match="专用只读"):
+        script.compatibility_settings(path)
+
+
+@pytest.mark.parametrize(
+    "version,passed",
+    [
+        ("10.50.1600", True),
+        ("10.50.1600.0", True),
+        ("10.50.6000", False),
+        ("16.0.4165.4", False),
+    ],
+)
+def test_同主版本或2022副本不能冒充备份源同版本(script, version, passed):
+    connection = MagicMock()
+    cursor = connection.cursor.return_value
+    cursor.execute.return_value.fetchone.side_effect = [
+        (version,),
+        ("中文核验", Decimal("12345.6789"), datetime(2008, 1, 2, 3, 4, 5)),
+        (0, 0, 0),
+    ]
+    connection.getinfo.return_value = "test-driver"
+    report = script._inspect_connection(connection, "10.50.1600")
+    assert report["compatible"] is passed
+    cursor.close.assert_called_once()
+
+
+def test_连接失败不降级且报告不带驱动异常原文(script, tmp_path, monkeypatch, capsys):
+    report_path = tmp_path / "report.json"
+    monkeypatch.setattr(
+        sys, "argv", ["check", "--connection-file", "unused.json", "--report", str(report_path)]
+    )
+    monkeypatch.setattr(script, "compatibility_settings", lambda path: script.get_settings())
+    engine = MagicMock()
+    engine.raw_connection.side_effect = RuntimeError("PWD=secret-test-only;手机号")
+    create = MagicMock(return_value=engine)
+    monkeypatch.setattr(script, "create_engine", create)
+    assert script.main() == 2
+    create.assert_called_once()
+    engine.raw_connection.assert_called_once()
+    engine.dispose.assert_called_once()
+    assert json.loads(report_path.read_text(encoding="utf-8")) == {
+        "compatible": False,
+        "error_type": "RuntimeError",
+    }
+    assert "secret-test-only" not in capsys.readouterr().out