mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-09 23:44:20 +08:00
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
import app.db as db_module
|
|
|
|
|
|
class _SqliteError(Exception):
|
|
"""模拟 sqlite3 异常暴露的扩展错误字段。"""
|
|
|
|
sqlite_errorcode = 266
|
|
sqlite_errorname = "SQLITE_IOERR_READ"
|
|
|
|
|
|
class _PsycopgError(Exception):
|
|
"""模拟 psycopg2 异常暴露的 SQLSTATE 字段。"""
|
|
|
|
pgcode = "40001"
|
|
|
|
|
|
class _AsyncpgError(Exception):
|
|
"""模拟 asyncpg 适配异常暴露的 SQLSTATE 字段。"""
|
|
|
|
sqlstate = "23505"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("error", "expected"),
|
|
[
|
|
(
|
|
_SqliteError("disk I/O error"),
|
|
{
|
|
"error_type": "_SqliteError",
|
|
"error_code": 266,
|
|
"error_name": "SQLITE_IOERR_READ",
|
|
},
|
|
),
|
|
(
|
|
_PsycopgError("serialization failure"),
|
|
{
|
|
"error_type": "_PsycopgError",
|
|
"sqlstate": "40001",
|
|
},
|
|
),
|
|
(
|
|
_AsyncpgError("duplicate key"),
|
|
{
|
|
"error_type": "_AsyncpgError",
|
|
"sqlstate": "23505",
|
|
},
|
|
),
|
|
],
|
|
)
|
|
def test_database_error_metadata_extracts_driver_codes(error, expected) -> None:
|
|
"""诊断元数据应兼容 SQLite、psycopg2 与 asyncpg 的稳定错误字段。"""
|
|
assert db_module._database_error_metadata(error) == expected
|
|
|
|
|
|
def test_database_error_listener_omits_statement_and_parameters(monkeypatch) -> None:
|
|
"""数据库错误日志不得包含 SQL、参数或驱动返回的原始消息。"""
|
|
messages = []
|
|
engine = create_engine("sqlite:///:memory:")
|
|
monkeypatch.setattr("app.db.logger.error", messages.append)
|
|
db_module._register_database_error_logging(engine)
|
|
|
|
with pytest.raises(OperationalError):
|
|
with engine.connect() as connection:
|
|
connection.execute(
|
|
text("SELECT * FROM missing_table WHERE token = :token"),
|
|
{"token": "private-token"},
|
|
)
|
|
|
|
assert len(messages) == 1
|
|
assert "database=sqlite" in messages[0]
|
|
assert "driver=pysqlite" in messages[0]
|
|
assert "error_code=1" in messages[0]
|
|
assert "error_name=SQLITE_ERROR" in messages[0]
|
|
assert "missing_table" not in messages[0]
|
|
assert "private-token" not in messages[0]
|
|
|
|
|
|
def test_async_database_engine_logs_driver_error_metadata(monkeypatch) -> None:
|
|
"""异步 Engine 应通过底层 sync engine 记录驱动错误码。"""
|
|
messages = []
|
|
monkeypatch.setattr("app.db.logger.error", messages.append)
|
|
|
|
async def query_missing_table() -> None:
|
|
async with db_module.AsyncEngine.connect() as connection:
|
|
await connection.execute(text("SELECT * FROM async_missing_table"))
|
|
|
|
with pytest.raises(OperationalError):
|
|
asyncio.run(query_missing_table())
|
|
|
|
assert len(messages) == 1
|
|
assert "database=sqlite" in messages[0]
|
|
assert "driver=aiosqlite" in messages[0]
|
|
assert "error_code=1" in messages[0]
|
|
assert "error_name=SQLITE_ERROR" in messages[0]
|
|
assert "async_missing_table" not in messages[0]
|
|
|
|
|
|
def test_database_error_metadata_ignores_unclassified_errors() -> None:
|
|
"""没有驱动错误码时不应制造无效诊断日志。"""
|
|
assert db_module._database_error_metadata(RuntimeError("plain failure")) is None
|