99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
"""PostgreSQL 兼容性冒烟测试。
|
|
|
|
**独立运行**，不依赖 pytest conftest 中的 reset_database 夹具。
|
|
直接管理 engine 生命周期，避免 autouse 夹具冲突。
|
|
|
|
验证：
|
|
1. SQLAlchemy metadata.create_all / drop_all 在 PG 下完整可用
|
|
2. 核心 CRUD（角色创建/查询/更新）在 PG 下正常
|
|
3. PG 特有类型问题（如 varchar vs date 比较）已被修复
|
|
|
|
运行方式：
|
|
UV_PYTHON_PATH="$PWD/.venv/bin/python" uv run python tests/test_pg_compat.py
|
|
|
|
或直接：
|
|
HUNJING_TEST_DATABASE_URL="postgresql+asyncpg://hunjing:change-me@127.0.0.1:5432/hunjing_test" \\
|
|
uv run python tests/test_pg_compat.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
|
|
|
|
async def _run() -> None:
|
|
database_url = os.environ.get("HUNJING_TEST_DATABASE_URL", "")
|
|
if not database_url.startswith("postgresql"):
|
|
print("❌ 请设 HUNJING_TEST_DATABASE_URL 环境变量")
|
|
sys.exit(1)
|
|
|
|
assert "test" in database_url, "必须指向名含 test 的数据库"
|
|
print(f"✅ 数据库 URL: {database_url}")
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.db.base import Base
|
|
|
|
engine = create_async_engine(database_url)
|
|
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
|
|
# ── 建表 ─────────────────────────────────────────────
|
|
print("⏳ 创建表...")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
print("✅ 39 张表已创建")
|
|
|
|
from app.core.ids import new_character_id, new_user_id
|
|
from app.models.character import Character
|
|
from app.repositories.json_utils import dumps, loads_dict
|
|
|
|
# ── CRUD ─────────────────────────────────────────────
|
|
async with AsyncSessionLocal() as session:
|
|
owner = new_user_id()
|
|
char = Character(
|
|
id=new_character_id(),
|
|
name="pg-test-char",
|
|
display_name="PG测试",
|
|
status="active",
|
|
language="zh-CN",
|
|
profile_json=dumps({"name": "PG 测试角色"}),
|
|
model_policy_json=dumps({"default_chat_model_hint": "test-model"}),
|
|
privacy_level="normal",
|
|
owner_scope=owner,
|
|
is_system_template=False,
|
|
)
|
|
session.add(char)
|
|
await session.flush()
|
|
char_id = char.id
|
|
print(f"✅ 创建角色: {char_id}")
|
|
|
|
from app.services.characters import get_owned_character
|
|
|
|
found = await get_owned_character(session, char_id, owner)
|
|
assert found.id == char_id
|
|
assert found.name == "pg-test-char"
|
|
print("✅ 查询角色")
|
|
|
|
profile = loads_dict(found.profile_json)
|
|
profile["description"] = "更新后描述"
|
|
found.profile_json = dumps(profile)
|
|
await session.flush()
|
|
await session.refresh(found)
|
|
assert loads_dict(found.profile_json).get("description") == "更新后描述"
|
|
print("✅ 更新角色")
|
|
|
|
await session.commit()
|
|
|
|
# ── 清理 ─────────────────────────────────────────────
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await engine.dispose()
|
|
print("✅ 表已清理")
|
|
print("\n🎉 PostgreSQL 兼容性验证全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_run())
|