SupplyAgent/tests/supply_infra/scheduler/test_sync_global_tree.py

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from supply_infra.odps.client import ODPSClient
from supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql import (
    _allocate_local_ids,
    _to_category_source_rows,
    sync_global_tree_odps_to_mysql,
)


def test_global_category_sql_uses_stable_tree_and_no_element_table() -> None:
    client = ODPSClient("id", "key", "project", "endpoint")
    client.execute_sql = MagicMock(return_value=[])

    client.fetch_global_categories("20260731")

    sql = client.execute_sql.call_args.args[0]
    assert "loghubods.global_category" in sql
    assert "stable_id,name,description,level,parent_stable_id" in sql
    assert "dt = '20260731'" in sql
    assert "retired_at_execution_id IS NULL" in sql
    assert "source_type = '实质'" in sql
    assert "ORDER BY level" in sql
    assert "global_tree_element" not in sql
    assert "pattern_mining_element" not in sql


def test_stable_ids_are_mapped_to_existing_and_new_local_ids() -> None:
    categories = _to_category_source_rows(
        [
            {
                "stable_id": 1,
                "name": "理念",
                "description": "",
                "level": 1,
                "parent_stable_id": None,
            },
            {
                "stable_id": 7392,
                "name": "壮烈牺牲",
                "description": "描述",
                "level": 6,
                "parent_stable_id": 1,
            },
        ]
    )

    mapping, failed = _allocate_local_ids(categories, {1: 10}, 11)

    assert mapping == {1: 10, 7392: 11}
    assert failed == set()


@patch(
    "supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql."
    "GlobalTreeCategoryRepository"
)
@patch("supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql.get_session")
@patch("supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql.get_odps_client")
def test_tree_sync_does_not_read_or_write_global_tree_element(
    mock_get_odps,
    mock_get_session,
    mock_repo_cls,
) -> None:
    mock_get_odps.return_value.fetch_global_categories.return_value = [
        {
            "stable_id": 1,
            "name": "理念",
            "description": "root",
            "level": 1,
            "parent_stable_id": None,
        },
        {
            "stable_id": 7392,
            "name": "壮烈牺牲",
            "description": "leaf",
            "level": 2,
            "parent_stable_id": 1,
        },
    ]
    mock_get_session.return_value.__enter__.return_value = MagicMock()
    repository = mock_repo_cls.return_value
    repository.count_active.return_value = 1
    repository.get_source_id_map.return_value = {1: 10}
    repository.get_max_id.return_value = 10
    repository.bulk_upsert.return_value = 2
    repository.mark_missing_deleted.return_value = 0

    result = sync_global_tree_odps_to_mysql("20260731")

    assert result["categories_inserted"] == 1
    assert result["global_tree_element_used"] is False
    rows = repository.bulk_upsert.call_args.args[0]
    assert rows[1]["source_id"] == 7392
    assert rows[1]["parent_id"] == 10


@patch("supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql.get_odps_client")
def test_empty_global_category_partition_fails_closed(mock_get_odps) -> None:
    mock_get_odps.return_value.fetch_global_categories.return_value = []

    with pytest.raises(RuntimeError, match="returned no active categories"):
        sync_global_tree_odps_to_mysql("20260731")


@patch(
    "supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql."
    "GlobalTreeCategoryRepository"
)
@patch("supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql.get_session")
@patch("supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql.get_odps_client")
def test_partial_tree_partition_does_not_retire_existing_tree(
    mock_get_odps,
    mock_get_session,
    mock_repo_cls,
) -> None:
    mock_get_odps.return_value.fetch_global_categories.return_value = [
        {
            "stable_id": 1,
            "name": "理念",
            "description": "",
            "level": 1,
            "parent_stable_id": None,
        }
    ]
    mock_get_session.return_value.__enter__.return_value = MagicMock()
    repository = mock_repo_cls.return_value
    repository.count_active.return_value = 100

    with pytest.raises(RuntimeError, match="dropped below safety threshold"):
        sync_global_tree_odps_to_mysql("20260731")

    repository.bulk_upsert.assert_not_called()
    repository.mark_missing_deleted.assert_not_called()