65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""数据库连接与 Repository 集成测试。"""
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from extensions.sqlalchemy import engine
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def test_db_connect():
|
|
with engine.connect() as conn:
|
|
result = conn.execute(select(1))
|
|
assert result.scalar() == 1
|
|
|
|
|
|
def test_repo_get_by_id_robust():
|
|
"""get_by_id 对非法 ID 容错，对已存在 ID 正常返回。"""
|
|
from modules.role.repository import RoleRepository
|
|
repo = RoleRepository()
|
|
try:
|
|
assert repo.get_by_id(None) is None
|
|
assert repo.get_by_id("") is None
|
|
assert repo.get_by_id("abc") is None
|
|
assert repo.get_by_id(0) is None
|
|
assert repo.get_by_id(-1) is None
|
|
first = repo.filter_by().order_by(RoleRepository.model.id.asc()).first()
|
|
assert first is not None
|
|
assert repo.get_by_id(first.id) is not None
|
|
finally:
|
|
repo.close()
|
|
|
|
|
|
def test_repo_exists_by_field():
|
|
"""exists_by_field 唯一性校验（仅未删除，可排除自身）。"""
|
|
from modules.role.models import Role
|
|
from modules.role.repository import RoleRepository
|
|
repo = RoleRepository()
|
|
try:
|
|
role = repo.create(Role(name="唯一名XYZ", code="uniq_xyz", status=1, sort=0))
|
|
assert repo.exists_by_field("code", "uniq_xyz") # 自身命中
|
|
assert not repo.exists_by_field("code", "uniq_xyz", exclude_id=role.id) # 排除自身后无重复
|
|
assert not repo.exists_by_field("code", "__不存在的编码__")
|
|
repo.db.query(Role).filter(Role.id == role.id).delete()
|
|
repo.db.commit()
|
|
finally:
|
|
repo.close()
|
|
|
|
|
|
def test_repo_batch_delete():
|
|
"""batch_delete 逗号分隔软删，返回实际删除条数。"""
|
|
from modules.role.models import Role
|
|
from modules.role.repository import RoleRepository
|
|
repo = RoleRepository()
|
|
try:
|
|
role = repo.create(Role(name="测试删除角色", code="test_del", status=1, sort=0))
|
|
assert repo.batch_delete(f"{role.id}") == 1
|
|
assert repo.get_by_id(role.id) is None # 软删后不可见
|
|
assert repo.get_by_id(role.id, soft_delete=False) is not None # 物理仍在
|
|
assert repo.batch_delete("abc,,0") == 0
|
|
# 清理
|
|
repo.db.query(Role).filter(Role.id == role.id).delete()
|
|
repo.db.commit()
|
|
finally:
|
|
repo.close()
|