"""需求池同步内部子阶段的异常隔离测试。""" 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()