44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""数据访问层自检：MySQL 可达且表已建则跑真实 CRUD；否则 SKIP（exit 0）。"""
|
||
import asyncio
|
||
|
||
from sqlalchemy import text
|
||
|
||
from config.database.mysql import get_engine, get_session_factory
|
||
from repositories.sys_config import SysConfigRepo
|
||
|
||
|
||
async def _db_ok() -> bool:
|
||
"""MySQL 可达且 sys_config 表已建（schema.sql 已导入）才算就绪。"""
|
||
try:
|
||
async with get_engine().connect() as conn:
|
||
await conn.execute(text("SELECT 1"))
|
||
async with get_session_factory()() as session:
|
||
await session.execute(text("SELECT 1 FROM sys_config LIMIT 1"))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
async def main():
|
||
try:
|
||
if not await _db_ok():
|
||
print("SKIP: MySQL 不可达或 sys_config 未建表，CRUD 检查延后（先启动 MySQL 并导入 sql/schema.sql）")
|
||
return
|
||
async with get_session_factory()() as session:
|
||
repo = SysConfigRepo(session)
|
||
row = await repo.set_value("test.hnw", "3000000", "自检")
|
||
assert row.config_key == "test.hnw"
|
||
assert await repo.get_value("test.hnw") == "3000000"
|
||
assert await repo.count() > 0
|
||
rows = await repo.list(where=(repo.model.config_key == "test.hnw",))
|
||
assert len(rows) == 1
|
||
assert await repo.delete(rows[0].id)
|
||
assert await repo.get_value("test.hnw") is None # 幂等清理
|
||
print("REPO TESTS PASSED")
|
||
finally:
|
||
import config.database as _db
|
||
await _db.dispose() # 干净收尾，避免连接在事件循环关闭后被 GC
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |