from __future__ import annotations import os from datetime import datetime 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 os.environ["DATABASE_URL"] = "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test" from app.api.ai_steward_api import routes_feedback_center 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.feedback_center import FeedbackCenterRecord, FeedbackCenterRecordRevision import app.services.feedback_center.service as feedback_center_service_module def _use_sqlite_json_types() -> None: for table in (FeedbackCenterRecord.__table__, FeedbackCenterRecordRevision.__table__): for column in table.columns: if column.name in {"before_value", "after_value", "ai_snapshot", "reason_candidates", "before_snapshot", "after_snapshot"}: column.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=[ FeedbackCenterRecord.__table__, FeedbackCenterRecordRevision.__table__, ], ) ) maker = async_sessionmaker(engine, expire_on_commit=False) async with maker() as session: yield session await engine.dispose() def _build_app(db_session): app = FastAPI() app.include_router(routes_feedback_center.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="ADMIN", ) return app @pytest.mark.asyncio async def test_feedback_center_summary_list_detail_and_catalog(db_session) -> None: db_session.add_all( [ FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="task_breakdown", source_id=101, object_type="task", object_id=201, field_name="task_object", source_title="任务拆解", object_title="任务拆解", before_value=None, after_value={"title": "任务拆解"}, ai_confidence=0.91, reason_candidates=[{"code": "manual_override", "label": "人工覆盖"}], reason_code="manual_override", review_status="pending", auto_assigned=False, ), FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="qa_feedback", source_id=102, object_type="qa_record", object_id=202, field_name="qa_feedback", source_title="Q&A", object_title="Q&A", before_value={"answer": "旧答案"}, after_value={"rating": "negative", "correct_answer": "新答案"}, ai_confidence=None, reason_candidates=[{"code": "answer_incorrect", "label": "答案不正确"}], reason_code="answer_incorrect", review_status="reviewed", auto_assigned=False, ), FeedbackCenterRecord( tenant_id=1, project_id=22, source_type="signal_feedback", source_id=103, object_type="signal", object_id=203, field_name="strategic_classification", source_title="上线异常告警", object_title="上线异常告警", before_value={"ai_predicted_strategic": True}, after_value={"was_strategic": False}, ai_confidence=0.8, reason_candidates=[{"code": "strategic_judgement_mismatch", "label": "战略判断偏差"}], reason_code="strategic_judgement_mismatch", review_status="accepted", auto_assigned=False, ), ] ) await db_session.commit() app = _build_app(db_session) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: summary_resp = await client.get("/ai-steward/feedback-center/summary", params={"project_id": 11}) list_resp = await client.get("/ai-steward/feedback-center/items", params={"project_id": 11, "limit": 20}) detail_resp = await client.get("/ai-steward/feedback-center/items/1", params={"project_id": 11}) catalog_resp = await client.get( "/ai-steward/feedback-center/reason-catalog", params={"source_type": "task_override", "field_name": "priority"}, ) assert summary_resp.status_code == 200 summary_payload = summary_resp.json() assert summary_payload["code"] == 0 assert summary_payload["data"]["total"] == 2 assert summary_payload["data"]["by_status"]["pending"] == 1 assert summary_payload["data"]["by_status"]["reviewed"] == 1 assert list_resp.status_code == 200 list_payload = list_resp.json() assert list_payload["code"] == 0 assert len(list_payload["data"]["items"]) == 2 assert all(item["project_id"] == 11 for item in list_payload["data"]["items"]) assert detail_resp.status_code == 200 detail_payload = detail_resp.json() assert detail_payload["code"] == 0 assert detail_payload["data"]["record"]["id"] == 1 assert detail_payload["data"]["record"]["source_type"] == "task_breakdown" assert detail_payload["data"]["revisions"] == [] assert catalog_resp.status_code == 200 catalog_payload = catalog_resp.json() assert catalog_payload["code"] == 0 codes = [item["code"] for item in catalog_payload["data"]["items"]] assert "priority_misalignment" in codes @pytest.mark.asyncio async def test_feedback_center_summary_honors_window_and_scope_query_params(db_session, monkeypatch) -> None: fixed_now = datetime(2025, 1, 15, 12, 0, 0) monkeypatch.setattr(feedback_center_service_module, "now_cn_naive", lambda: fixed_now) db_session.add_all( [ FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="task_breakdown", source_id=801, object_type="task", object_id=901, field_name="task_breakdown", source_title="任务拆解", object_title="任务拆解", review_status="accepted", auto_assigned=True, created_at=datetime(2025, 1, 15, 8, 0, 0), reviewed_at=datetime(2025, 1, 15, 8, 30, 0), ), FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="task_override", source_id=802, object_type="task", object_id=901, field_name="priority", source_title="优先级修正", object_title="优先级修正", review_status="reviewed", created_at=datetime(2025, 1, 14, 8, 0, 0), reviewed_at=datetime(2025, 1, 14, 8, 50, 0), ), FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="qa_feedback", source_id=803, object_type="qa_record", object_id=902, field_name="qa_feedback", source_title="QA 反馈", object_title="QA 反馈", review_status="accepted", created_at=datetime(2025, 1, 13, 8, 0, 0), reviewed_at=datetime(2025, 1, 13, 8, 20, 0), ), ] ) await db_session.commit() app = _build_app(db_session) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get( "/ai-steward/feedback-center/summary", params={"project_id": 11, "window_days": 30, "source_group": "task"}, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert payload["data"]["source_group"] == "task" assert payload["data"]["trend"]["window_days"] == 30 assert payload["data"]["trend"]["current"]["total_records"] == 2 assert payload["data"]["trend"]["current"]["task_breakdown_count"] == 1 assert payload["data"]["trend"]["current"]["task_override_count"] == 1 assert payload["data"]["trend"]["current"]["qa_feedback_count"] == 0 assert payload["data"]["insights"] insight_codes = {item["code"] for item in payload["data"]["insights"]} assert "manual_override_pressure" in insight_codes @pytest.mark.asyncio async def test_feedback_center_summary_accepts_scope_params(db_session) -> None: db_session.add_all( [ FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="task_breakdown", source_id=201, object_type="task", object_id=301, field_name="task_object", review_status="pending", ), FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="qa_feedback", source_id=202, object_type="qa_record", object_id=302, field_name="qa_feedback", review_status="accepted", ), ] ) await db_session.commit() app = _build_app(db_session) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: scoped_resp = await client.get( "/ai-steward/feedback-center/summary", params={"project_id": 11, "window_days": 30, "source_group": "qa"}, ) invalid_window = await client.get( "/ai-steward/feedback-center/summary", params={"project_id": 11, "window_days": 14}, ) invalid_group = await client.get( "/ai-steward/feedback-center/summary", params={"project_id": 11, "source_group": "unknown"}, ) assert scoped_resp.status_code == 200 scoped_payload = scoped_resp.json() assert scoped_payload["code"] == 0 assert scoped_payload["data"]["trend"]["window_days"] == 30 assert scoped_payload["data"]["by_source_type"] == {"qa_feedback": 1} assert invalid_window.status_code == 400 assert invalid_group.status_code == 400 @pytest.mark.asyncio async def test_feedback_center_review_updates_record_and_revision(db_session) -> None: db_session.add( FeedbackCenterRecord( tenant_id=1, project_id=11, source_type="task_override", source_id=500, object_type="task", object_id=600, field_name="priority", source_title="优先级修正", object_title="优先级修正", before_value="P2", after_value="P1", ai_confidence=0.62, reason_candidates=[{"code": "priority_misalignment", "label": "优先级错配"}], reason_code="priority_misalignment", review_status="pending", auto_assigned=False, ) ) await db_session.commit() app = _build_app(db_session) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/ai-steward/feedback-center/items/1/review", json={ "review_status": "accepted", "reason_code": "priority_misalignment", "reason_text": "优先级确认需要上调", }, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert payload["data"]["review_status"] == "accepted" assert payload["data"]["reason_code"] == "priority_misalignment" assert payload["data"]["reviewed_by"] == 9001 refreshed = await db_session.get(FeedbackCenterRecord, 1) assert refreshed is not None assert refreshed.review_status == "accepted" assert refreshed.reason_text == "优先级确认需要上调" revisions = ( await db_session.execute( FeedbackCenterRecordRevision.__table__.select().where(FeedbackCenterRecordRevision.record_id == 1) ) ).mappings().all() assert len(revisions) == 1 assert revisions[0]["before_snapshot"]["review_status"] == "pending" assert revisions[0]["after_snapshot"]["review_status"] == "accepted"