add: PostgreSQL 支持(asyncpg 驱动,from_url/from_config + 集成测试)
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user