SupplyAgent/tests/api/test_demand_feedback.py

from __future__ import annotations

from datetime import datetime
from types import SimpleNamespace

import pytest
from pydantic import ValidationError

from api.schemas.demand_feedback import CreateDemandFeedbackBody
from api.services import demand_feedback as feedback_service
from supply_infra.db.models.demand_feedback import DemandFeedback


def _body(**changes) -> CreateDemandFeedbackBody:
    values = {
        "client_request_id": "request-123",
        "target_type": "demand",
        "demand_grade_id": 10,
        "feedback_action": "support",
    }
    values.update(changes)
    return CreateDemandFeedbackBody(**values)


def test_feedback_body_validates_target_fields() -> None:
    with pytest.raises(ValidationError):
        _body(target_type="video")
    with pytest.raises(ValidationError):
        _body(
            target_type="hit_content",
            video_id="vid-1",
        )
    with pytest.raises(ValidationError):
        _body(video_id="vid-1")

    body = _body(target_type="video", video_id="vid-1")
    assert body.video_id == "vid-1"


def test_feedback_body_validates_action_content_and_reason() -> None:
    with pytest.raises(ValidationError):
        _body(feedback_action="oppose")
    with pytest.raises(ValidationError):
        _body(feedback_action="correct")
    with pytest.raises(ValidationError):
        _body(reason_code="irrelevant")

    body = _body(
        feedback_action="correct",
        reason_code="wrong_reason",
        content="  正确原因应引用后验样本。  ",
    )
    assert body.content == "正确原因应引用后验样本。"


def test_feedback_model_has_person_but_no_review_or_boolean_state() -> None:
    columns = DemandFeedback.__table__.columns
    assert "feedback_user_id" in columns
    assert "feedback_user_name_snapshot" in columns
    assert "feedback_username_snapshot" in columns
    assert "review_status" not in columns
    assert "has_feedback" not in columns
    assert "is_feedback" not in columns


def test_create_feedback_uses_server_target_and_current_user(monkeypatch) -> None:
    grade = SimpleNamespace(
        id=10,
        biz_dt="20260730",
        demand_name="老年人智能手机使用",
        grade="A",
        reason="需求有明确内容意图",
        video_list=None,
    )
    expansion = SimpleNamespace(
        id=30,
        biz_dt="20260730",
        source_demand_grade_id=10,
        video_id="vid-1",
        point_type="purpose",
        expanded_text="降低老人使用智能设备的门槛",
        point_desc="解释基础操作",
        reason="覆盖当前需求",
    )
    detail = SimpleNamespace(title="教父母使用手机")
    saved: list[DemandFeedback] = []

    class Session:
        def refresh(self, row) -> None:
            row.id = 99
            row.created_at = datetime(2026, 7, 30, 15, 20)

        def rollback(self) -> None:
            raise AssertionError("unexpected rollback")

    class SessionContext:
        def __enter__(self):
            return Session()

        def __exit__(self, *_args):
            return False

    class GradeRepo:
        def __init__(self, _session):
            pass

        def get_by_id(self, grade_id):
            return grade if grade_id == 10 else None

    class ExpansionRepo:
        def __init__(self, _session):
            pass

        def has_video(self, **kwargs):
            return kwargs["video_id"] == "vid-1"

        def get_active_by_id(self, expansion_id):
            return expansion if expansion_id == 30 else None

    class DetailRepo:
        def __init__(self, _session):
            pass

        def list_by_vids(self, _video_ids):
            return {"vid-1": detail}

    class FeedbackRepo:
        def __init__(self, _session):
            pass

        def get_by_client_request_id(self, _request_id):
            return None

        def add(self, row):
            saved.append(row)
            return row

        def list_for_target(self, **_kwargs):
            return saved, len(saved)

    monkeypatch.setattr(feedback_service, "get_session", lambda: SessionContext())
    monkeypatch.setattr(feedback_service, "DemandGradeRepository", GradeRepo)
    monkeypatch.setattr(feedback_service, "DemandVideoExpansionRepository", ExpansionRepo)
    monkeypatch.setattr(feedback_service, "MultiDemandVideoDetailRepository", DetailRepo)
    monkeypatch.setattr(feedback_service, "DemandFeedbackRepository", FeedbackRepo)

    result = feedback_service.create_demand_feedback(
        _body(
            target_type="hit_content",
            video_id="vid-1",
            demand_video_expansion_id=30,
            feedback_action="oppose",
            reason_code="not_hit",
            content="没有覆盖老人学习使用的需求。",
        ),
        {
            "id": 7,
            "username": "zhangsan",
            "display_name": "张三",
        },
    )

    assert result["feedback_summary"] == {"count": 1}
    assert result["item"]["feedback_user"] == {
        "id": 7,
        "username": "zhangsan",
        "display_name": "张三",
    }
    assert result["item"]["target_snapshot"]["expanded_text"] == "降低老人使用智能设备的门槛"
    assert saved[0].feedback_user_id == 7
    assert saved[0].feedback_user_name_snapshot == "张三"
    assert saved[0].feedback_username_snapshot == "zhangsan"