chore(db): add driver error diagnostics (#6115)

tests/test_db_error_diagnostics.py

@@ -0,0 +1,106 @@
+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