72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
|
"""SQLite 后端回归测试（tmp_path，无需外部服务）。"""
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlmodel import Field, SQLModel, select
|
||
|
|
|
||
|
|
from chongming_database import Database
|
||
|
|
|
||
|
|
|
||
|
|
class SqliteItem(SQLModel, table=True):
|
||
|
|
__tablename__ = "backend_test_item"
|
||
|
|
|
||
|
|
id: int | None = Field(default=None, primary_key=True)
|
||
|
|
name: str = Field(max_length=50)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_sqlite_crud(tmp_path: Path):
|
||
|
|
db = Database(tmp_path / "nested" / "test.db")
|
||
|
|
await db.initialize()
|
||
|
|
try:
|
||
|
|
async with db.get_session() as session:
|
||
|
|
session.add(SqliteItem(name="a"))
|
||
|
|
await session.commit()
|
||
|
|
async with db.get_session() as session:
|
||
|
|
rows = (await session.execute(select(SqliteItem))).scalars().all()
|
||
|
|
assert [r.name for r in rows] == ["a"]
|
||
|
|
finally:
|
||
|
|
await db.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_sqlite_initialize_is_idempotent(tmp_path: Path):
|
||
|
|
"""重复 initialize（模拟多进程各自初始化）不应报错。"""
|
||
|
|
db = Database(tmp_path / "test.db")
|
||
|
|
await db.initialize()
|
||
|
|
try:
|
||
|
|
await db.initialize()
|
||
|
|
finally:
|
||
|
|
await db.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_get_session_before_initialize_raises(tmp_path: Path):
|
||
|
|
db = Database(tmp_path / "test.db")
|
||
|
|
try:
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
db.get_session()
|
||
|
|
finally:
|
||
|
|
await db.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_from_config_sqlite_end_to_end(tmp_path: Path):
|
||
|
|
db = Database.from_config(
|
||
|
|
{"type": "sqlite", "database_path": str(tmp_path / "cfg.db")}
|
||
|
|
)
|
||
|
|
await db.initialize()
|
||
|
|
try:
|
||
|
|
async with db.get_session() as session:
|
||
|
|
await session.execute(select(SqliteItem))
|
||
|
|
finally:
|
||
|
|
await db.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_generate_schemas_false_skips_ddl(tmp_path: Path):
|
||
|
|
"""generate_schemas=False 时只建引擎不建表，会话仍可用。"""
|
||
|
|
db = Database(tmp_path / "noschema.db")
|
||
|
|
await db.initialize(generate_schemas=False)
|
||
|
|
try:
|
||
|
|
async with db.get_session() as session:
|
||
|
|
await session.execute(select(1))
|
||
|
|
finally:
|
||
|
|
await db.dispose()
|