hot_comment_radar/tests/unit/test_ai_schema.py

import pytest
import httpx

from app.services.ai_service import (
    AIAnalysisResult,
    OpenAICompatibleAIClient,
    analyze_comments_with_retry,
    build_report_summary_prompt,
    build_comment_prompt,
    calculate_analysis_status,
    parse_ai_comment_response,
)
from app.services.task_service import build_report_summary_provider


def test_build_comment_prompt_contains_ids_and_truncates_content():
    prompt = build_comment_prompt([{"comment_id": "c1", "content": "你" * 200}])

    assert "c1" in prompt
    assert "请原样回填输入中的 comment_id" in prompt
    assert "你" * 150 in prompt
    assert "你" * 151 not in prompt


def test_build_report_summary_prompt_contains_stats_labels_and_truncated_typical_comments():
    prompt = build_report_summary_prompt(
        {
            "sample_count": 2,
            "sentiment": {
                "positive": {"count": 1, "pct": 50.0},
                "neutral": {"count": 0, "pct": 0.0},
                "negative": {"count": 1, "pct": 50.0},
                "unknown": {"count": 0, "pct": 0.0},
            },
            "top_labels": [{"name": "价格争议", "count": 2}],
        },
        {"positive": [{"content": "好" * 200}], "negative": [{"content": "太贵"}]},
        word_limit=200,
    )

    assert "只返回一段中文总结" in prompt
    assert "样本评论数量：2" in prompt
    assert "positive: 1 (50.0%)" in prompt
    assert "价格争议: 2" in prompt
    assert "好" * 150 in prompt
    assert "好" * 151 not in prompt
    assert "JSON Array" not in prompt


def test_report_summary_provider_is_disabled_without_real_ai_client():
    assert build_report_summary_provider(client=None) is None


def test_parse_ai_comment_response_validates_array_sentiment_labels_and_ids():
    result = parse_ai_comment_response(
        '[{"comment_id":"c1","sentiment":"positive","labels":["质量好"],"reason":"认可"}]',
        expected_comment_ids={"c1"},
    )

    assert result == [AIAnalysisResult(comment_id="c1", sentiment="positive", labels=["质量好"], reason="认可")]


@pytest.mark.parametrize(
    "payload",
    [
        "not-json",
        '{"comment_id":"c1"}',
        '[{"comment_id":"missing","sentiment":"positive","labels":[],"reason":""}]',
        '[{"comment_id":"c1","sentiment":"happy","labels":[],"reason":""}]',
        '[{"comment_id":"c1","sentiment":"positive","labels":["a","b","c","d"],"reason":""}]',
    ],
)
def test_parse_ai_comment_response_rejects_invalid_payloads(payload):
    with pytest.raises(ValueError):
        parse_ai_comment_response(payload, expected_comment_ids={"c1"})


def test_calculate_analysis_status_uses_eighty_percent_threshold():
    assert calculate_analysis_status(success_count=8, total_count=10) == (0.8, "normal")
    assert calculate_analysis_status(success_count=7, total_count=10) == (0.7, "insufficient")
    assert calculate_analysis_status(success_count=0, total_count=0) == (0.0, "insufficient")


def test_analyze_comments_with_retry_marks_batch_failed_after_three_parse_failures(monkeypatch):
    sleeps = []
    monkeypatch.setattr("app.services.ai_service.time.sleep", sleeps.append)

    results = analyze_comments_with_retry(
        [{"comment_id": "c1", "content": "内容"}],
        requester=lambda _prompt: "not-json",
        max_retries=3,
    )

    assert results[0].comment_id == "c1"
    assert results[0].sentiment == "unknown"
    assert results[0].ai_analysis_status == "failed"
    assert results[0].reason == "ai_parse_failed"
    assert sleeps == [1, 2]


def test_openai_compatible_client_posts_chat_completion_and_returns_message_content():
    captured = {}

    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        captured["authorization"] = request.headers.get("authorization")
        captured["body"] = request.read().decode("utf-8")
        return httpx.Response(
            200,
            json={"choices": [{"message": {"content": '[{"comment_id":"c1","sentiment":"positive","labels":["认可"],"reason":"喜欢"}]'}}]},
        )

    http_client = httpx.Client(transport=httpx.MockTransport(handler))
    client = OpenAICompatibleAIClient(
        base_url="https://ai.example.com",
        api_key="test-ai-key",
        model="test-model",
        http_client=http_client,
    )

    result = client.request("prompt text")

    assert result == '[{"comment_id":"c1","sentiment":"positive","labels":["认可"],"reason":"喜欢"}]'
    assert captured["url"] == "https://ai.example.com/v1/chat/completions"
    assert captured["authorization"] == "Bearer test-ai-key"
    assert '"model":"test-model"' in captured["body"]
    assert "prompt text" in captured["body"]
    assert "请严格遵循用户指令输出。" in captured["body"]


def test_openai_compatible_client_accepts_custom_system_prompt():
    captured = {}

    def handler(request: httpx.Request) -> httpx.Response:
        captured["body"] = request.read().decode("utf-8")
        return httpx.Response(200, json={"choices": [{"message": {"content": "报告摘要"}}]})

    http_client = httpx.Client(transport=httpx.MockTransport(handler))
    client = OpenAICompatibleAIClient(
        base_url="https://ai.example.com",
        api_key="test-ai-key",
        model="test-model",
        http_client=http_client,
    )

    assert client.request("总结 prompt", system_prompt="只返回一段中文总结，不使用 Markdown。") == "报告摘要"
    assert "只返回一段中文总结" in captured["body"]
    assert "JSON Array" not in captured["body"]


def test_openai_compatible_client_accepts_base_url_that_already_includes_v1():
    captured = {}

    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        return httpx.Response(
            200,
            json={"choices": [{"message": {"content": "[]"}}]},
        )

    http_client = httpx.Client(transport=httpx.MockTransport(handler))
    client = OpenAICompatibleAIClient(
        base_url="https://ai.example.com/v1",
        api_key="test-ai-key",
        model="test-model",
        http_client=http_client,
    )

    client.request("prompt text")

    assert captured["url"] == "https://ai.example.com/v1/chat/completions"