content-find-agent-new/tests/test_platform_observable_performance.py

from __future__ import annotations

from content_agent.business_modules.content_discovery.platform_observable_performance import (
    performance_score,
)


def test_douyin_play_count_is_natural_missing_and_score_is_bounded():
    result = performance_score(
        {
            "digg_count": 100000,
            "comment_count": 1000,
            "share_count": 500,
            "collect_count": 1000,
        },
        "douyin",
    )

    assert 0 <= result["platform_performance_score"] <= 100
    assert {row["field"] for row in result["platform_performance_components"]} == {
        "statistics.digg_count",
        "statistics.comment_count",
        "statistics.share_count",
        "statistics.collect_count",
    }
    assert result["missing_observable_fields"] == [
        {
            "field": "statistics.play_count",
            "missing_type": "natural_platform_missing",
            "platform": "douyin",
            "evidence": "跨平台字段映射.json",
        }
    ]
    assert "platform_heat" not in result


def test_kuaishou_all_five_fields_supported():
    result = performance_score(
        {
            "play_count": 10000,
            "digg_count": 2000,
            "comment_count": 200,
            "share_count": 100,
            "collect_count": 100,
        },
        "kuaishou",
    )

    assert len(result["platform_performance_components"]) == 5
    assert result["missing_observable_fields"] == []
    assert result["platform_performance_score"] is not None


def test_shipinhao_only_digg_supported_and_other_fields_natural_missing():
    result = performance_score({"digg_count": 500}, "shipinhao")

    assert [row["field"] for row in result["platform_performance_components"]] == [
        "statistics.digg_count"
    ]
    assert {row["field"] for row in result["missing_observable_fields"]} == {
        "statistics.comment_count",
        "statistics.share_count",
        "statistics.collect_count",
        "statistics.play_count",
    }


def test_supported_field_absent_is_runtime_missing():
    result = performance_score({"digg_count": 10}, "douyin")

    runtime_missing = [
        row for row in result["missing_observable_fields"]
        if row.get("missing_type") == "runtime_missing"
    ]
    assert {row["field"] for row in runtime_missing} == {
        "statistics.comment_count",
        "statistics.share_count",
        "statistics.collect_count",
    }