SupplyAgent/tests/supply_infra/scheduler/test_sync_multi_demand_pool.py

"""需求池同步内部子阶段的异常隔离测试。"""
from __future__ import annotations

from unittest.mock import MagicMock, patch

from supply_infra.category_match import CategoryMatch
from supply_infra.scheduler.jobs.demand_pool.sync import (
    _classify_words,
    _sync_pool_rows,
)


@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.GlobalTreeCategoryRepository"
)
@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.DemandBelongCategoryRepository"
)
@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.MultiDemandPoolDiRepository"
)
@patch("supply_infra.scheduler.jobs.demand_pool.sync.get_session")
def test_classification_calls_match_api_once_per_term_and_keeps_partial_success(
    mock_get_session,
    mock_pool_repo_cls,
    mock_belong_repo_cls,
    mock_tree_repo_cls,
) -> None:
    mock_get_session.return_value.__enter__.return_value = MagicMock()
    mock_pool_repo_cls.return_value.list_demand_names_by_biz_dt.return_value = [
        "牺牲 未知词",
        "失败词",
    ]
    mock_belong_repo_cls.return_value.get_existing_names.return_value = set()
    mock_belong_repo_cls.return_value.upsert_category_matches.return_value = 1
    mock_tree_repo_cls.return_value.get_active_source_id_map.return_value = {
        7392: 88
    }
    client = MagicMock()

    def _match(term: str, *, description: str):
        assert description == ""
        if term == "失败词":
            raise RuntimeError("upstream unavailable")
        if term == "未知词":
            return None
        return CategoryMatch(
            term="牺牲",
            stable_id=7392,
            name="壮烈牺牲",
            path="/理念/事件/军事事件/军人事迹/英勇战功/壮烈牺牲",
            description="为国为民献出生命",
            level=6,
            score=1.0,
        )

    client.match_one.side_effect = _match

    result = _classify_words("20260721", client=client)

    assert client.match_one.call_count == 3
    client.match_one.assert_any_call("牺牲", description="")
    client.match_one.assert_any_call("未知词", description="")
    client.match_one.assert_any_call("失败词", description="")
    assert result["success"] is False
    assert result["api_calls"] == 3
    assert result["matched"] == 1
    assert result["persisted"] == 1
    assert result["unmatched_terms"] == ["未知词"]
    assert result["failed_items"] == [
        {"term": "失败词", "error": "upstream unavailable"}
    ]
    rows = mock_belong_repo_cls.return_value.upsert_category_matches.call_args.args[0]
    assert rows == [
        {
            "name": "牺牲",
            "category_id": 88,
            "reason": (
                "category_match_api_v2: "
                "path=/理念/事件/军事事件/军人事迹/英勇战功/壮烈牺牲; "
                "score=1.0000; stable_id=7392"
            ),
            "is_delete": 0,
        }
    ]


@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.DemandBelongPoolRelRepository"
)
@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.MultiDemandPoolDiRepository"
)
@patch("supply_infra.scheduler.jobs.demand_pool.sync.get_session")
@patch("supply_infra.scheduler.jobs.demand_pool.sync.get_odps_client")
def test_same_count_content_revision_is_updated(
    mock_get_odps,
    mock_get_session,
    mock_pool_repo_cls,
    mock_rel_repo_cls,
) -> None:
    mock_get_odps.return_value.fetch_multi_demand_pool.return_value = [
        {
            "strategy": "逐月",
            "demand_id": "d-1",
            "demand_name": "修订后的需求",
            "weight": 0.8,
            "type": "topic",
            "video_list": ["v-2"],
            "extend": "new",
        }
    ]
    mock_get_session.return_value.__enter__.return_value = MagicMock()
    repo = mock_pool_repo_cls.return_value
    repo.list_source_rows_by_biz_dt.return_value = [
        {
            "id": 11,
            "strategy": "逐月",
            "demand_id": "d-1",
            "demand_name": "修订前的需求",
            "weight": 0.5,
            "type": "topic",
            "video_count": 1,
            "video_list": '["v-1"]',
            "extend": "old",
            "biz_dt": "20260730",
        }
    ]
    repo.delete_by_keys.return_value = 0
    repo.bulk_insert.return_value = 0
    repo.update_source_fields.return_value = 1
    mock_rel_repo_cls.return_value.delete_by_pool_ids.return_value = 0

    result = _sync_pool_rows("20260730")

    assert result["fetched"] == 1
    assert result["inserted"] == 0
    assert result["deleted"] == 0
    assert result["updated"] == 1
    assert result["unchanged"] == 0
    assert len(result["source_snapshot_hash"]) == 64
    repo.update_source_fields.assert_called_once()
    updated_row = repo.update_source_fields.call_args.args[1][0]
    assert updated_row["demand_name"] == "修订后的需求"
    assert updated_row["video_list"] == '["v-2"]'


@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.DemandBelongPoolRelRepository"
)
@patch(
    "supply_infra.scheduler.jobs.demand_pool.sync.MultiDemandPoolDiRepository"
)
@patch("supply_infra.scheduler.jobs.demand_pool.sync.get_session")
@patch("supply_infra.scheduler.jobs.demand_pool.sync.get_odps_client")
def test_unchanged_content_is_not_rewritten(
    mock_get_odps,
    mock_get_session,
    mock_pool_repo_cls,
    mock_rel_repo_cls,
) -> None:
    upstream = {
        "strategy": "逐月",
        "demand_id": "d-1",
        "demand_name": "老年 健身",
        "weight": 0.8,
        "type": "topic",
        "video_list": ["v-1"],
        "extend": "same",
    }
    mock_get_odps.return_value.fetch_multi_demand_pool.return_value = [upstream]
    mock_get_session.return_value.__enter__.return_value = MagicMock()
    repo = mock_pool_repo_cls.return_value
    repo.list_source_rows_by_biz_dt.return_value = [
        {
            "id": 11,
            "strategy": "逐月",
            "demand_id": "d-1",
            "demand_name": "老年 健身",
            "weight": 0.8,
            "type": "topic",
            "video_count": 1,
            "video_list": '["v-1"]',
            "extend": "same",
            "biz_dt": "20260730",
        }
    ]
    repo.delete_by_keys.return_value = 0
    repo.bulk_insert.return_value = 0
    repo.update_source_fields.return_value = 0
    mock_rel_repo_cls.return_value.delete_by_pool_ids.return_value = 0

    result = _sync_pool_rows("20260730")

    assert result["updated"] == 0
    assert result["unchanged"] == 1
    repo.update_source_fields.assert_not_called()