119 lines
3.4 KiB
Python
119 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import JSON
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.api.ai_steward_api import routes_conflicts
|
|
from app.core.database import get_db
|
|
from app.dependencies.auth import get_current_user
|
|
from app.dependencies.tenant import get_tenant_id
|
|
from app.models.base import Base
|
|
from app.models.cognitive_conflict import CognitiveConflict
|
|
from app.models.truth_node import TruthNode
|
|
from app.models.user import User
|
|
|
|
|
|
def _use_sqlite_json_types() -> None:
|
|
User.__table__.c.assigned_domains.type = JSON()
|
|
TruthNode.__table__.c.content.type = JSON()
|
|
CognitiveConflict.__table__.c.involved_user_ids.type = JSON()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
_use_sqlite_json_types()
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(
|
|
lambda sync_conn: Base.metadata.create_all(
|
|
sync_conn,
|
|
tables=[
|
|
User.__table__,
|
|
TruthNode.__table__,
|
|
CognitiveConflict.__table__,
|
|
],
|
|
)
|
|
)
|
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with maker() as session:
|
|
yield session
|
|
await engine.dispose()
|
|
|
|
|
|
def _build_app(*, db_session) -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(routes_conflicts.router)
|
|
|
|
async def _fake_get_db():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = _fake_get_db
|
|
app.dependency_overrides[get_tenant_id] = lambda: 1
|
|
app.dependency_overrides[get_current_user] = lambda: SimpleNamespace(
|
|
id=9001,
|
|
tenant_id=1,
|
|
name="管理员",
|
|
username="admin",
|
|
role="REVIEWER",
|
|
)
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cognitive_conflicts_return_display_type(db_session) -> None:
|
|
user = User(
|
|
tenant_id=1,
|
|
username="tester",
|
|
email="tester@example.com",
|
|
password_hash="hash",
|
|
name="测试用户",
|
|
role="REVIEWER",
|
|
status="active",
|
|
is_active=True,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
|
|
truth_node = TruthNode(
|
|
tenant_id=1,
|
|
node_type="charter",
|
|
parent_id=None,
|
|
project_id=7,
|
|
signal_id=None,
|
|
created_by=user.id,
|
|
title="项目宪章",
|
|
content={},
|
|
version=1,
|
|
status="active",
|
|
)
|
|
db_session.add(truth_node)
|
|
await db_session.flush()
|
|
|
|
db_session.add(
|
|
CognitiveConflict(
|
|
tenant_id=1,
|
|
truth_node_id=truth_node.id,
|
|
conflict_type="skill_mismatch",
|
|
description="技能要求与成员能力不匹配",
|
|
involved_user_ids=[user.id],
|
|
status="active",
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
app = _build_app(db_session=db_session)
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/ai-steward/cognitive-conflicts")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload[0]["type"] == "skill_mismatch"
|
|
assert payload[0]["display_type"] == "技能错配"
|