Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions core/database.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,17 @@ def get_db_config() -> dict:
"journal_mode": "WAL",
"busy_timeout": 10000,
"foreign_keys": "ON",
# 小内存 VPS 下的 IOPS 与内存折中。Tortoise 的 SqliteClient 会把
# credentials 中 file_path 以外的键逐条执行成 PRAGMA,且 SQLite
# 后端只维持单条连接,因此这些设置在进程生命周期内全局生效。
# synchronous=NORMAL:WAL 下应用崩溃不丢已提交事务,
# 仅 OS/断电可能丢最后若干事务,省去每次提交的 fsync。
# temp_store=MEMORY:临时表与排序走内存,小库收益明显;
# 后台大排序的内存峰值需在压测中关注。
# cache_size=-8000:page cache 上限约 8MB(负数单位为 KiB)。
"synchronous": "NORMAL",
"temp_store": "MEMORY",
"cache_size": -8000,
},
}
},
Expand Down
50 changes: 50 additions & 0 deletions tests/test_sqlite_pragmas.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
"""SQLite 连接 PRAGMA 生效性验证。

不 mock 配置字典,而是用 get_db_config() 真起一个文件库读取 PRAGMA——
Tortoise 的 SqliteClient 在 create_connection() 时逐条执行 credentials 中的
PRAGMA,只有真实连接才能验证这一点。
"""
import pytest
from tortoise import Tortoise

import core.database as database
from tests.helpers import close_db


@pytest.fixture
async def file_db(tmp_path, monkeypatch):
monkeypatch.setattr(database, "_DB_FILE", str(tmp_path / "pragma.db"))
if Tortoise._inited:
await Tortoise.close_connections()
await Tortoise.init(config=database.get_db_config())
try:
yield Tortoise.get_connection("default")
finally:
await close_db()


async def _pragma(conn, statement: str):
rows = await conn.execute_query_dict(statement)
return next(iter(rows[0].values()))


@pytest.mark.asyncio
class TestSqlitePragmas:
async def test_wal_and_busy_timeout_kept(self, file_db):
assert str(await _pragma(file_db, "PRAGMA journal_mode")).lower() == "wal"
assert await _pragma(file_db, "PRAGMA busy_timeout") == 10000

async def test_synchronous_normal(self, file_db):
# 1 = NORMAL(WAL 下省 fsync),2 = FULL
assert await _pragma(file_db, "PRAGMA synchronous") == 1

async def test_temp_store_memory(self, file_db):
# 0 = DEFAULT,1 = FILE,2 = MEMORY
assert await _pragma(file_db, "PRAGMA temp_store") == 2

async def test_cache_size_8mb(self, file_db):
# 负值单位为 KiB
assert await _pragma(file_db, "PRAGMA cache_size") == -8000

async def test_foreign_keys_on(self, file_db):
assert await _pragma(file_db, "PRAGMA foreign_keys") == 1