SupplyAgent/tests/api/test_growth_category_tree.py

from collections.abc import Generator
from contextlib import contextmanager
from types import SimpleNamespace

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

from api.auth_middleware import normal_user_can_access
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")


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,
        ),
        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["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