zkxw/backend/tests/test_cn_holiday_sync.py

"""cn_holiday sync / read unit tests"""
from datetime import date, datetime
from unittest.mock import MagicMock, patch

from app.services.cn_holiday import is_rest_day, list_day_flags
from app.services.cn_holiday_sync import (
    _is_rest_from_type,
    is_year_calendar_complete,
    map_ailcc_day_row,
    seconds_until_next_local_midnight,
    sync_cn_holidays,
)


def test_type_rest_flags():
    assert _is_rest_from_type(1) == 1
    assert _is_rest_from_type(2) == 1
    assert _is_rest_from_type(3) == 1
    assert _is_rest_from_type(0) == 0
    assert _is_rest_from_type(4) == 0


def test_map_row():
    row = map_ailcc_day_row(
        "2026-10-01",
        holiday={"holiday": True, "name": "国庆节（休）", "date": "2026-10-01"},
        type_info={"type": 2, "name": "国庆节（休）", "week": 4},
    )
    assert row["day_date"] == date(2026, 10, 1)
    assert row["day_type"] == 2
    assert row["is_rest_day"] == 1
    assert row["day_name"] == "国庆节（休）"
    assert row["week"] == 4
    assert row["raw_json"]


def test_map_row_workday_makeup():
    row = map_ailcc_day_row(
        "2026-10-10",
        holiday=None,
        type_info={"type": 4, "name": "国庆节（班）", "week": 6},
    )
    assert row["day_type"] == 4
    assert row["is_rest_day"] == 0


def test_map_row_infer_weekend_when_type_missing():
    # 2026-10-03 is Saturday
    row = map_ailcc_day_row("2026-10-03", holiday=None, type_info=None)
    assert row["day_type"] == 1
    assert row["is_rest_day"] == 1
    assert row["week"] == 6


def test_is_rest_day_from_row():
    db = MagicMock()
    row = MagicMock()
    row.is_rest_day = 1
    row.day_type = 2
    db.query.return_value.filter.return_value.first.return_value = row
    assert is_rest_day(db, date(2026, 10, 1)) is True


def test_is_rest_day_work_type():
    db = MagicMock()
    row = MagicMock()
    row.is_rest_day = 0
    row.day_type = 4
    db.query.return_value.filter.return_value.first.return_value = row
    assert is_rest_day(db, date(2026, 10, 10)) is False


def test_is_rest_day_fallback_weekend():
    db = MagicMock()
    db.query.return_value.filter.return_value.first.return_value = None
    # Saturday
    assert is_rest_day(db, date(2026, 8, 8)) is True
    # Friday
    assert is_rest_day(db, date(2026, 8, 7)) is False


def test_list_day_flags_fills_gaps():
    db = MagicMock()
    row = MagicMock()
    row.day_date = date(2026, 10, 1)
    row.day_type = 2
    row.is_rest_day = 1
    row.day_name = "国庆节（休）"
    db.query.return_value.filter.return_value.all.return_value = [row]

    flags = list_day_flags(db, date(2026, 10, 1), date(2026, 10, 3))
    assert len(flags) == 3
    assert flags[0] == {
        "dayDate": "2026-10-01",
        "dayType": 2,
        "isRestDay": True,
        "dayName": "国庆节（休）",
    }
    # 2026-10-02 Friday → work fallback
    assert flags[1]["dayDate"] == "2026-10-02"
    assert flags[1]["dayType"] == 0
    assert flags[1]["isRestDay"] is False
    # 2026-10-03 Saturday → rest fallback
    assert flags[2]["dayDate"] == "2026-10-03"
    assert flags[2]["dayType"] == 1
    assert flags[2]["isRestDay"] is True


def test_seconds_until_next_local_midnight():
    now = datetime(2026, 8, 10, 23, 59, 0)
    assert seconds_until_next_local_midnight(now) == 60
    noon = datetime(2026, 8, 10, 12, 0, 0)
    assert seconds_until_next_local_midnight(noon) == 12 * 3600


def test_is_year_calendar_complete_true():
    session = MagicMock()
    # 2026 is not leap → 365
    session.query.return_value.filter.return_value.count.return_value = 365
    assert is_year_calendar_complete(session, 2026) is True


def test_is_year_calendar_complete_false():
    session = MagicMock()
    session.query.return_value.filter.return_value.count.return_value = 100
    assert is_year_calendar_complete(session, 2026) is False


def test_is_year_calendar_complete_leap():
    session = MagicMock()
    session.query.return_value.filter.return_value.count.return_value = 366
    assert is_year_calendar_complete(session, 2024) is True
    session.query.return_value.filter.return_value.count.return_value = 365
    assert is_year_calendar_complete(session, 2024) is False


def test_sync_cn_holidays_upsert_mocked():
    holiday_payload = {
        "code": 0,
        "holiday": {
            "2026-10-01": {"holiday": True, "name": "国庆节（休）", "date": "2026-10-01"},
        },
        "type": {
            "2026-10-01": {"type": 2, "name": "国庆节（休）", "week": 4},
            "2026-10-02": {"type": 2, "name": "国庆节（休）", "week": 5},
        },
    }

    session = MagicMock()
    session.query.return_value.filter.return_value.first.return_value = None
    # year incomplete → still fetch
    session.query.return_value.filter.return_value.count.return_value = 0

    class Factory:
        def __call__(self):
            return session

    with patch("app.services.cn_holiday_sync._fetch_batch", return_value=holiday_payload), patch(
        "app.services.cn_holiday_sync._years_to_sync", return_value=[2026]
    ), patch(
        "app.services.cn_holiday_sync._months_for_year", return_value=[(2026, 10)]
    ):
        n = sync_cn_holidays(Factory())

    assert n >= 1
    assert session.add.call_count >= 1 or session.merge.call_count >= 1
    session.commit.assert_called()
    session.close.assert_called()


def test_sync_skips_complete_year():
    session = MagicMock()
    session.query.return_value.filter.return_value.count.return_value = 365

    class Factory:
        def __call__(self):
            return session

    with patch("app.services.cn_holiday_sync._fetch_batch") as fetch, patch(
        "app.services.cn_holiday_sync._years_to_sync", return_value=[2026]
    ):
        n = sync_cn_holidays(Factory())

    assert n == 0
    fetch.assert_not_called()
    session.commit.assert_called()
    session.close.assert_called()