88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
"""T+ 配置和读取适配层的纯单元测试。"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy.dialects.mssql.pyodbc import MSDialect_pyodbc
|
|
|
|
from app.core.config import Settings
|
|
from app.services.tplus.mapping import MappingStatus, get_mapping
|
|
from app.services.tplus.reader import TPlusReader
|
|
|
|
|
|
def _settings(**overrides: object) -> Settings:
|
|
params: dict[str, object] = {"_env_file": None, "app_secret_key": "x" * 32}
|
|
params.update(overrides)
|
|
return Settings(**params) # type: ignore[arg-type]
|
|
|
|
|
|
def test_T加数据库URL使用mssql_pyodbc并隐藏密码() -> None:
|
|
settings = _settings(
|
|
t_plus_enabled=True,
|
|
t_plus_database="UFTData472115_000001",
|
|
t_plus_user="reader",
|
|
t_plus_password="p@ss:w/ord",
|
|
)
|
|
url = settings.t_plus_database_url
|
|
assert url.drivername == "mssql+pyodbc"
|
|
assert url.database == "UFTData472115_000001"
|
|
assert "p@ss:w/ord" not in url.render_as_string()
|
|
|
|
|
|
def test_T加未启用时读取器不建立连接() -> None:
|
|
settings = _settings(t_plus_enabled=False)
|
|
assert settings.t_plus_enabled is False
|
|
|
|
|
|
def test_T加默认加密且允许显式隔离环境回退(monkeypatch) -> None:
|
|
monkeypatch.delenv("T_PLUS_ENCRYPT", raising=False)
|
|
monkeypatch.delenv("T_PLUS_TRUST_SERVER_CERTIFICATE", raising=False)
|
|
assert _settings().t_plus_database_url.query["Encrypt"] == "yes"
|
|
assert _settings().t_plus_database_url.query["TrustServerCertificate"] == "yes"
|
|
url = _settings(t_plus_encrypt=False, t_plus_trust_server_certificate=False).t_plus_database_url
|
|
assert url.query["Encrypt"] == "no"
|
|
assert url.query["TrustServerCertificate"] == "no"
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["a;b", "a}b", "{prefix", "中文;UID=other}x"])
|
|
def test_ODBC特殊字符保持字段边界且不泄露密码(value) -> None:
|
|
settings = _settings(
|
|
t_plus_odbc_driver="Driver}Name",
|
|
t_plus_database=value,
|
|
t_plus_user=value,
|
|
t_plus_password=value,
|
|
t_plus_encrypt=True,
|
|
)
|
|
url = settings.t_plus_database_url
|
|
args, _ = MSDialect_pyodbc().create_connect_args(url)
|
|
escaped = "{" + value.replace("}", "}}") + "}"
|
|
assert f"Database={escaped};UID={escaped};PWD={escaped};" in args[0]
|
|
assert "DRIVER={Driver}}Name}" in args[0]
|
|
assert ";Encrypt=yes;" in args[0]
|
|
assert url.render_as_string().split("@", 1)[0].endswith(":***")
|
|
|
|
|
|
def test_T加查询文件覆盖六类实体() -> None:
|
|
reader = TPlusReader.__new__(TPlusReader)
|
|
reader.query_dir = Path(__file__).resolve().parents[3] / "db" / "queries" / "tplus"
|
|
assert reader.inventory() == (
|
|
"personnel",
|
|
"products",
|
|
"customers",
|
|
"businesses",
|
|
"shipments",
|
|
"receipts",
|
|
)
|
|
for entity in reader.inventory():
|
|
assert (reader.query_dir / f"{entity}.sql").is_file()
|
|
|
|
|
|
def test_T加高风险映射保持待确认() -> None:
|
|
assert get_mapping("businesses").status is MappingStatus.PENDING
|
|
assert "成本口径" in get_mapping("businesses").unresolved
|
|
|
|
|
|
def test_未知T加实体被拒绝() -> None:
|
|
with pytest.raises(KeyError):
|
|
get_mapping("unknown")
|