b8920c0d28
补充业务详情指导成本入口、来源版本和历史封存保护，接入月份锁并完善权限复核。 同步接入读取服务运行状态到页面，补充前端控制、回归测试和服务端只读连接授权校验。 定向隔离测试：89 项通过，1 条既有 Starlette 弃用警告。
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""兼容性脚本与应用使用相同连接策略，失败报告不泄露连接数据。"""
|
|
|
|
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,
|
|
t_plus_allow_dev_sa=False, t_plus_allow_admin_account=False,
|
|
)
|
|
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
|