66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""Skill 数据访问测试。"""
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.skills import Skill
|
|
from app.repositories import skills as skills_repo
|
|
|
|
|
|
def _skill(name: str = "coding", **overrides) -> Skill:
|
|
base = {
|
|
"name": name,
|
|
"description": "代码辅助",
|
|
"trigger": "代码,编程",
|
|
"prompt_fragment": "你是代码助手",
|
|
"tool_bindings_json": "[]",
|
|
"risk_level": "L1",
|
|
"status": "enabled",
|
|
}
|
|
base.update(overrides)
|
|
return Skill(**base)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_and_get_skill() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
s = _skill()
|
|
await skills_repo.add_skill(session, s)
|
|
await session.commit()
|
|
sid = s.id
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
fetched = await skills_repo.get_skill(session, sid)
|
|
assert fetched is not None
|
|
assert fetched.name == "coding"
|
|
assert fetched.trigger == "代码,编程"
|
|
assert fetched.status == "enabled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_skills_with_status_filter() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
await skills_repo.add_skill(session, _skill("a", status="enabled"))
|
|
await skills_repo.add_skill(session, _skill("b", status="disabled"))
|
|
await session.commit()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
enabled = await skills_repo.list_skills(session, status="enabled")
|
|
all_skills = await skills_repo.list_skills(session)
|
|
assert len(enabled) == 1
|
|
assert enabled[0].name == "a"
|
|
assert len(all_skills) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_name_unique_conflict() -> None:
|
|
async with AsyncSessionLocal() as session:
|
|
await skills_repo.add_skill(session, _skill("dup"))
|
|
await session.commit()
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
with pytest.raises(IntegrityError):
|
|
await skills_repo.add_skill(session, _skill("dup"))
|
|
await session.commit()
|