SupplyAgent/tests/api/test_growth_category_tree.py

from collections.abc import Generator
from contextlib import contextmanager
from types import SimpleNamespace
from unittest.mock import MagicMock

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from api.auth_middleware import normal_user_can_access
from api.services import aigc_article_html
from api.services import growth_category_tree as service
from api.services.growth_category_tree import _build_growth_tree
from supply_infra.db.base import Base
from supply_infra.db.models.global_category_content_weight import (
    GlobalCategoryContentWeight,
)
from supply_infra.db.models.global_v2 import GlobalCategoryV2


def test_growth_tree_api_is_available_to_normal_users() -> None:
    assert normal_user_can_access("GET", "/api/growth-category-tree")
    assert normal_user_can_access("GET", "/api/growth-category-tree/38/channel-contents")
    assert normal_user_can_access("GET", "/api/growth-channel-content/content-1/view")


def test_build_growth_tree_uses_stable_ids_and_daily_scores() -> None:
    categories = [
        SimpleNamespace(
            stable_id=10,
            parent_stable_id=None,
            level=1,
            name="根分类",
            description="根描述",
        ),
        SimpleNamespace(
            stable_id=11,
            parent_stable_id=10,
            level=2,
            name="子分类",
            description=None,
        ),
        SimpleNamespace(
            stable_id=12,
            parent_stable_id=999,
            level=2,
            name="孤立分类",
            description=None,
        ),
    ]
    weights = [
        SimpleNamespace(
            stable_id=10,
            source_element_count=8,
            read_rate_score=0.3,
            avg_read_rate_score=1.2,
            like_rate_score=0.04,
            account_uid_count=2,
            channel_content_id_count=5,
            cal_fans_num_sum=123.45,
        ),
        SimpleNamespace(
            stable_id=11,
            source_element_count=0,
            read_rate_score=0.0,
            avg_read_rate_score=0.0,
            like_rate_score=0.0,
        ),
    ]

    nodes = _build_growth_tree(categories, weights)

    assert [node["id"] for node in nodes] == [10, 12]
    root = nodes[0]
    assert root["weights"] == {
        "read_rate": 0.3,
        "avg_read_rate": 1.2,
        "like_rate": 0.04,
    }
    assert root["counts"] == {
        "read_rate": 8,
        "avg_read_rate": 8,
        "like_rate": 8,
    }
    assert root["account_uid_count"] == 2
    assert root["channel_content_id_count"] == 5
    assert root["cal_fans_num_sum"] == 123.45
    assert root["children"][0]["id"] == 11
    assert root["children"][0]["counts"]["read_rate"] == 0
    assert "hung_word_count" not in root


def test_build_growth_category_tree_serializes_before_session_closes(monkeypatch) -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            GlobalCategoryV2.__table__,
            GlobalCategoryContentWeight.__table__,
        ],
    )
    factory = sessionmaker(bind=engine, expire_on_commit=True)
    with factory() as session:
        session.add(
            GlobalCategoryV2(
                id=1,
                stable_id=10,
                name="根分类",
                description=None,
                source_type="实质",
                level=1,
                parent_stable_id=None,
            )
        )
        session.add(
            GlobalCategoryContentWeight(
                id=1,
                stable_id=10,
                parent_stable_id=None,
                level=1,
                biz_dt="20260819",
                status="completed",
                source_element_count=5,
                read_rate_score=0.2,
                avg_read_rate_score=1.5,
                like_rate_score=0.03,
            )
        )
        session.commit()

    @contextmanager
    def get_test_session() -> Generator[Session, None, None]:
        session = factory()
        try:
            yield session
            session.commit()
        finally:
            session.close()

    monkeypatch.setattr(service, "get_session", get_test_session)

    payload = service.build_growth_category_tree()

    assert payload["biz_dt"] == "20260819"
    assert payload["nodes"][0]["weights"]["avg_read_rate"] == 1.5


def test_list_growth_category_channel_contents_serializes_top_ten(monkeypatch) -> None:
    @contextmanager
    def get_test_session():
        yield MagicMock()

    weight_repository = MagicMock()
    weight_repository.get_latest_completed_biz_dt.return_value = "20260819"
    rank_repository = MagicMock()
    rank_repository.list_top.return_value = [
        SimpleNamespace(
            rank_no=1,
            channel_content_id="content-1",
            source_element_id=101,
            source_category_stable_id=12,
            contribution=0.8,
            metric_value=0.4,
            weighted_score=1.0,
            weighted_share=0.25,
        )
    ]
    monkeypatch.setattr(service, "get_session", get_test_session)
    monkeypatch.setattr(
        service,
        "GlobalCategoryContentWeightRepository",
        lambda _session: weight_repository,
    )
    monkeypatch.setattr(
        service,
        "GlobalCategoryChannelContentRankRepository",
        lambda _session: rank_repository,
    )

    payload = service.list_growth_category_channel_contents(
        stable_id=38,
        metric="read_rate",
    )

    assert payload["biz_dt"] == "20260819"
    assert payload["total"] == 1
    assert payload["items"][0]["weighted_score"] == 1.0
    rank_repository.list_top.assert_called_once_with(
        biz_dt="20260819",
        stable_id=38,
        metric_type="read_rate",
    )


def test_latest_article_html_uses_parameterized_content_id(monkeypatch) -> None:
    connection = MagicMock()
    connection.scalar.return_value = b"<html>article</html>"
    engine = MagicMock()
    engine.connect.return_value.__enter__.return_value = connection
    monkeypatch.setattr(aigc_article_html, "_get_aigc_readonly_engine", lambda: engine)

    html = aigc_article_html.get_latest_article_html("content-1")

    assert html == "<html>article</html>"
    parameters = connection.scalar.call_args.args[1]
    assert parameters == {"channel_content_id": "content-1"}