"""Public-holiday calendar: service invariants, HR API, employee 公示, exports.

The behaviours worth locking down here are the ones that were non-obvious while
building the feature, and the ones where a regression would be quiet rather than
loud:

* two different holidays on one date must both survive (the natural key is
  country+day+*name*, not country+day),
* ``residence_country`` wins over ``country``, and an alpha-3 ``country`` must
  not be truncated to a different real country (``CHN`` → ``CH``),
* an unmaintained country-year reports ``no_data`` instead of "no holidays",
* re-importing a spreadsheet never reverts an HR edit,
* the generated 放假通知 matches the wording HR actually publishes.
"""
from __future__ import annotations

import io
import uuid
from datetime import date

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.hr import holiday_service as svc
from app.domains.hr.holiday_models import (
    MAKEUP_WORKDAY,
    PUBLIC_HOLIDAY,
    HrPublicHoliday,
    HrPublicHolidayCalendar,
    name_key_for,
)
from app.domains.hr.holiday_notice import BILINGUAL, EN, ZH, NoticeRequest, render_notice
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/hr/holidays"


# ──────────────────────────── helpers ────────────────────────────


async def _seed_group(
    db: AsyncSession,
    *,
    country: str,
    name_en: str,
    days: list[date],
    makeup: list[date] | None = None,
    name_local: str | None = None,
    year: int | None = None,
    stated_days: int | None = None,
    gcal_url: str | None = None,
    label: str | None = None,
) -> str:
    resolved_year = year or days[0].year
    existing = await svc.get_calendar(db, country=country, year=resolved_year)
    if existing is None:
        db.add(HrPublicHolidayCalendar(
            country=country, year=resolved_year, label=label or country,
            stated_days=stated_days, gcal_url=gcal_url, source="seed",
        ))
        await db.flush()
    group_key, _rows = await svc.create_group(
        db,
        country=country,
        name_en=name_en,
        name_local=name_local,
        days=days,
        makeup_days=makeup or [],
        source="seed",
    )
    await db.commit()
    return group_key


async def _person(
    db: AsyncSession,
    *,
    residence: str | None = None,
    country: str | None = None,
) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"h-{uuid.uuid4().hex[:8]}@pingcap.com",
        display_name="Holiday Reader",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-hol-{uuid.uuid4().hex[:10]}",
        residence_country=residence,
        country=country,
    )
    db.add(person)
    await db.flush()
    await db.commit()
    return person


def _labor_day_group() -> svc.HolidayGroup:
    return svc.HolidayGroup(
        country="CN",
        group_key="cn-2026-labor-day-holiday",
        name_en="Labor Day Holiday",
        name_local="劳动节",
        note=None,
        days=[date(2026, 5, d) for d in (1, 2, 3, 4, 5)],
        makeup_days=[date(2026, 5, 9)],
    )


# ──────────────────────── aggregation / counting ────────────────────────


def test_contiguous_segments_splits_on_a_gap():
    days = [date(2026, 9, 25), date(2026, 9, 26), date(2026, 9, 28)]
    assert svc.contiguous_segments(days) == [
        (date(2026, 9, 25), date(2026, 9, 26)),
        (date(2026, 9, 28), date(2026, 9, 28)),
    ]


def test_contiguous_segments_ignores_duplicate_dates():
    days = [date(2026, 5, 1), date(2026, 5, 1), date(2026, 5, 2)]
    assert svc.contiguous_segments(days) == [(date(2026, 5, 1), date(2026, 5, 2))]


def test_observed_day_count_is_weekdays_net_of_makeup():
    # 2026-05-01 Fri, 02 Sat, 03 Sun, 04 Mon, 05 Tue -> 3 weekdays, 1 makeup.
    group = _labor_day_group()
    assert group.total_days == 5
    assert group.observed_days == 2


def test_observed_day_count_never_goes_negative():
    # Nonsense data (more makeup days than weekday holidays) must report 0
    # rather than a negative number of days off.
    assert svc.observed_day_count(
        [date(2026, 5, 2)],  # a Saturday: zero weekday holidays
        [date(2026, 5, 9), date(2026, 5, 16)],
    ) == 0


def test_build_groups_keeps_two_holidays_on_the_same_date_apart():
    """Hong Kong observes Ching Ming and Holy Saturday both on 2026-04-04."""
    rows = [
        HrPublicHoliday(
            id=1, country="HK", day=date(2026, 4, 4),
            group_key="hk-2026-ching-ming-festival", name_en="Ching Ming Festival",
            name_key="ching-ming-festival", day_type=PUBLIC_HOLIDAY, source="seed",
        ),
        HrPublicHoliday(
            id=2, country="HK", day=date(2026, 4, 4),
            group_key="hk-2026-holy-saturday", name_en="Holy Saturday",
            name_key="holy-saturday", day_type=PUBLIC_HOLIDAY, source="seed",
        ),
    ]
    groups = svc.build_groups(rows)
    assert len(groups) == 2
    assert {g.name_en for g in groups} == {"Ching Ming Festival", "Holy Saturday"}


def test_build_groups_takes_display_name_from_a_holiday_not_a_makeup_row():
    rows = [
        HrPublicHoliday(
            id=1, country="CN", day=date(2026, 5, 1), group_key="g",
            name_en="Labor Day Holiday", name_local="劳动节",
            name_key="labor-day-holiday", day_type=PUBLIC_HOLIDAY, source="seed",
        ),
        HrPublicHoliday(
            id=2, country="CN", day=date(2026, 5, 9), group_key="g",
            name_en="Makeup Workday (Labor Day Holiday)", name_local="劳动节调休上班",
            name_key="makeup-workday-labor-day-holiday", day_type=MAKEUP_WORKDAY,
            source="seed",
        ),
    ]
    group = svc.build_groups(rows)[0]
    assert group.name_en == "Labor Day Holiday"
    assert group.name_local == "劳动节"
    assert group.makeup_days == [date(2026, 5, 9)]


def test_name_key_survives_punctuation_and_case_drift():
    assert name_key_for("New Year's Day") == name_key_for("new years day")
    assert name_key_for("Easter Monday（observed）") == name_key_for(
        "Easter Monday (observed)"
    )
    # A distinct holiday must not collide with it.
    assert name_key_for("Easter Monday") != name_key_for("Easter Monday (observed)")


def test_name_key_keeps_non_ascii_names_distinct():
    """A CJK-only name must not slug to the same placeholder as another."""
    assert name_key_for("春节") != name_key_for("中秋节")
    assert name_key_for("春节") != "holiday"


# ──────────────────────── country resolution ────────────────────────


def test_person_country_prefers_residence_over_legal_entity():
    person = IdentityPerson(residence_country="SG", country="CHN")
    assert svc.person_country(person) == "SG"


def test_person_country_normalises_alpha3_instead_of_truncating():
    """``CHN``[:2] is ``CH`` — Switzerland. That bug would serve Swiss holidays."""
    person = IdentityPerson(residence_country=None, country="CHN")
    assert svc.person_country(person) == "CN"


def test_person_country_is_none_when_nothing_resolves():
    assert svc.person_country(IdentityPerson(residence_country="", country="ZZ")) is None
    assert svc.person_country(None) is None


def test_resolve_country_rejects_unrecognised_input():
    from app.core.exceptions import ValidationError

    assert svc.resolve_country("cn") == "CN"
    assert svc.resolve_country("中国") == "CN"
    assert svc.resolve_country("JPN") == "JP"
    with pytest.raises(ValidationError):
        svc.resolve_country("Freedonia")


# ──────────────────────── service writes ────────────────────────


async def test_create_group_stores_days_and_makeup_and_creates_calendar(
    db_session: AsyncSession,
):
    group_key, rows = await svc.create_group(
        db_session,
        country="cn",
        name_en="Labor Day Holiday",
        name_local="劳动节",
        days=[date(2026, 5, d) for d in (1, 2, 3, 4, 5)],
        makeup_days=[date(2026, 5, 9)],
    )
    await db_session.commit()
    assert len(rows) == 6
    stored = await svc.list_days(db_session, countries=["CN"], group_key=group_key)
    assert len([r for r in stored if r.day_type == PUBLIC_HOLIDAY]) == 5
    assert len([r for r in stored if r.day_type == MAKEUP_WORKDAY]) == 1
    # The coverage row must exist, or the External API would say no_data while
    # visibly serving holidays.
    assert await svc.get_calendar(db_session, country="CN", year=2026) is not None


async def test_create_group_rejects_a_date_that_is_both_off_and_a_workday(
    db_session: AsyncSession,
):
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError):
        await svc.create_group(
            db_session,
            country="CN",
            name_en="Confused Festival",
            name_local=None,
            days=[date(2026, 5, 1)],
            makeup_days=[date(2026, 5, 1)],
        )


async def test_allocate_group_key_suffixes_a_repeated_name(db_session: AsyncSession):
    """Japan observes two separate "New Year Holidays" — they must not merge."""
    first = await _seed_group(
        db_session, country="JP", name_en="New Year Holidays",
        days=[date(2026, 1, 1), date(2026, 1, 2)],
    )
    second, _ = await svc.create_group(
        db_session,
        country="JP",
        name_en="New Year Holidays",
        name_local=None,
        days=[date(2026, 12, 29), date(2026, 12, 30)],
    )
    await db_session.commit()
    assert first != second
    assert second.startswith(first)
    groups = svc.build_groups(await svc.list_days(db_session, countries=["JP"]))
    assert len(groups) == 2


async def test_replace_group_is_diff_based_and_preserves_created_at(
    db_session: AsyncSession,
):
    group_key = await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1), date(2026, 5, 2)],
    )
    before = {
        r.day: r.created_at
        for r in await svc.list_days(db_session, countries=["CN"], group_key=group_key)
    }
    await svc.replace_group(
        db_session,
        country="CN",
        group_key=group_key,
        name_en="Labor Day Holiday",
        name_local="劳动节",
        days=[date(2026, 5, 1), date(2026, 5, 2), date(2026, 5, 3)],
        makeup_days=[date(2026, 5, 9)],
    )
    await db_session.commit()
    after = await svc.list_days(db_session, countries=["CN"], group_key=group_key)
    assert {r.day for r in after if r.day_type == PUBLIC_HOLIDAY} == {
        date(2026, 5, 1), date(2026, 5, 2), date(2026, 5, 3),
    }
    kept = [r for r in after if r.day == date(2026, 5, 1)][0]
    assert kept.created_at == before[date(2026, 5, 1)]


async def test_replace_group_removes_days_no_longer_listed(db_session: AsyncSession):
    group_key = await _seed_group(
        db_session, country="CN", name_en="Spring Festival",
        days=[date(2026, 2, d) for d in (15, 16, 17)],
    )
    await svc.replace_group(
        db_session, country="CN", group_key=group_key,
        name_en="Spring Festival", name_local=None,
        days=[date(2026, 2, 16), date(2026, 2, 17)],
    )
    await db_session.commit()
    remaining = await svc.list_days(db_session, countries=["CN"], group_key=group_key)
    assert {r.day for r in remaining} == {date(2026, 2, 16), date(2026, 2, 17)}


async def test_replace_group_refuses_to_collide_with_another_group(
    db_session: AsyncSession,
):
    from app.core.exceptions import ConflictError

    await _seed_group(
        db_session, country="HK", name_en="Ching Ming Festival",
        days=[date(2026, 4, 4)],
    )
    other = await _seed_group(
        db_session, country="HK", name_en="Holy Saturday", days=[date(2026, 4, 5)],
    )
    with pytest.raises(ConflictError):
        await svc.replace_group(
            db_session, country="HK", group_key=other,
            name_en="Ching Ming Festival", name_local=None,
            days=[date(2026, 4, 4)],
        )


async def test_bulk_upsert_skips_existing_rows_by_default(db_session: AsyncSession):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1)],
    )
    row = (await svc.list_days(db_session, countries=["CN"]))[0]
    row.note = "HR edited this"
    await db_session.commit()

    result = await svc.bulk_upsert(
        db_session,
        rows=[svc.BulkRow(
            country="CN", day=date(2026, 5, 1), name_en="Labor Day Holiday",
        )],
        source="import",
    )
    await db_session.commit()
    assert (result.created, result.skipped) == (0, 1)
    # The HR edit must survive a re-import.
    again = (await svc.list_days(db_session, countries=["CN"]))[0]
    assert again.note == "HR edited this"


async def test_bulk_upsert_overwrite_refreshes_matched_rows(db_session: AsyncSession):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", days=[date(2026, 5, 1)],
    )
    result = await svc.bulk_upsert(
        db_session,
        rows=[svc.BulkRow(
            country="CN", day=date(2026, 5, 1), name_en="Labor Day Holiday",
            name_local="劳动节", note="from vendor sheet",
        )],
        source="import",
        overwrite=True,
    )
    await db_session.commit()
    assert (result.created, result.updated) == (0, 1)
    row = (await svc.list_days(db_session, countries=["CN"]))[0]
    assert (row.name_local, row.note) == ("劳动节", "from vendor sheet")


async def test_bulk_upsert_groups_same_named_days_together(db_session: AsyncSession):
    """An imported multi-day holiday spread over several rows becomes one group."""
    result = await svc.bulk_upsert(
        db_session,
        rows=[
            svc.BulkRow(country="CN", day=date(2026, 10, d), name_en="National Day")
            for d in (1, 2, 3)
        ],
        source="import",
    )
    await db_session.commit()
    assert result.created == 3
    groups = svc.build_groups(await svc.list_days(db_session, countries=["CN"]))
    assert len(groups) == 1
    assert groups[0].total_days == 3


async def test_bulk_upsert_attaches_a_makeup_row_to_its_holiday_group(
    db_session: AsyncSession,
):
    await svc.bulk_upsert(
        db_session,
        rows=[
            svc.BulkRow(country="CN", day=date(2026, 5, 1), name_en="Labor Day"),
            svc.BulkRow(
                country="CN", day=date(2026, 5, 9),
                name_en="Makeup Workday (Labor Day)", day_type=MAKEUP_WORKDAY,
            ),
        ],
        source="import",
    )
    await db_session.commit()
    groups = svc.build_groups(await svc.list_days(db_session, countries=["CN"]))
    assert len(groups) == 1
    assert groups[0].makeup_days == [date(2026, 5, 9)]


async def test_bulk_upsert_validates_every_row_before_writing_any(
    db_session: AsyncSession,
):
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError):
        await svc.bulk_upsert(
            db_session,
            rows=[
                svc.BulkRow(country="CN", day=date(2026, 5, 1), name_en="Good"),
                svc.BulkRow(country="Freedonia", day=date(2026, 5, 2), name_en="Bad"),
            ],
            source="import",
        )
    # Nothing partially applied.
    assert await svc.list_days(db_session, countries=["CN"]) == []


async def test_delete_calendar_refuses_to_orphan_days(db_session: AsyncSession):
    from app.core.exceptions import ConflictError

    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    with pytest.raises(ConflictError):
        await svc.delete_calendar(db_session, country="CN", year=2026)
    result = await svc.delete_calendar(
        db_session, country="CN", year=2026, cascade=True,
    )
    await db_session.commit()
    assert result["days_deleted"] == 1
    assert await svc.list_days(db_session, countries=["CN"]) == []


async def test_coverage_distinguishes_unmaintained_from_empty(db_session: AsyncSession):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    result = await svc.coverage(db_session, countries=["CN", "PT"], year=2026)
    assert result == {"CN": "covered", "PT": "no_data"}


async def test_upcoming_includes_a_holiday_already_under_way(db_session: AsyncSession):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday",
        days=[date(2026, 5, d) for d in (1, 2, 3, 4, 5)],
    )
    await _seed_group(
        db_session, country="CN", name_en="Dragon Boat Festival",
        days=[date(2026, 6, 19)],
    )
    # Day 3 of the 5-day holiday: the current one must come first, not be skipped.
    groups = await svc.upcoming_groups(
        db_session, countries=["CN"], today=date(2026, 5, 3), limit=2,
    )
    assert [g.name_en for g in groups] == [
        "Labor Day Holiday", "Dragon Boat Festival",
    ]


async def test_restore_seed_is_idempotent_and_keeps_edits(db_session: AsyncSession):
    first = await svc.restore_seed(db_session)
    await db_session.commit()
    assert first["days_added"] == 328
    assert first["calendars_added"] == 22

    # An HR edit that the restore must not stomp.
    cal = await svc.get_calendar(db_session, country="AU", year=2026)
    assert cal is not None
    cal.stated_days = 99
    await db_session.commit()

    second = await svc.restore_seed(db_session)
    await db_session.commit()
    assert second["days_added"] == 0
    assert second["calendars_added"] == 0
    cal = await svc.get_calendar(db_session, country="AU", year=2026)
    assert cal is not None and cal.stated_days == 99


async def test_seeded_stated_days_match_the_derived_count_except_romania(
    db_session: AsyncSession,
):
    """The document's headline numbers are observed weekday days off.

    Romania is the one genuine divergence (2026-06-01 carries two holiday names,
    so HR counted the date twice). Locking this in means a future seed edit that
    breaks the relationship elsewhere fails loudly.
    """
    await svc.restore_seed(db_session)
    await db_session.commit()
    summary = await svc.country_year_summary(db_session, year=2026)
    mismatched = {
        row["country"]: (row["stated_days"], row["observed_days"])
        for row in summary
        if not row["stated_days_matches_derived"]
    }
    assert mismatched == {"RO": (12, 11)}


async def test_gcal_url_must_be_an_https_google_calendar_link(
    db_session: AsyncSession,
):
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError):
        await svc.upsert_calendar(
            db_session, country="CN", year=2026,
            gcal_url="https://evil.example.com/steal",
        )
    with pytest.raises(ValidationError):
        await svc.upsert_calendar(
            db_session, country="CN", year=2026,
            gcal_url="javascript:alert(1)",
        )
    row = await svc.upsert_calendar(
        db_session, country="CN", year=2026,
        gcal_url="https://calendar.google.com/calendar/u/0?cid=abc",
    )
    await db_session.commit()
    assert row.gcal_url is not None


async def test_cross_year_holiday_gets_a_calendar_row_for_both_years(
    db_session: AsyncSession,
):
    """A 2026-12-31 → 2027-01-02 holiday must be visible in BOTH years.

    Ensuring only the anchor year left the January days stored but invisible on
    the External API: with no 2027 calendar row that year reports
    ``coverage="no_data"``, and a ``no_data`` year returns an empty holiday list
    by contract — so a scheduler would never see them.
    """
    await svc.create_group(
        db_session,
        country="VN",
        name_en="New Year Holidays",
        name_local=None,
        days=[date(2026, 12, 31), date(2027, 1, 1), date(2027, 1, 2)],
    )
    await db_session.commit()

    assert await svc.get_calendar(db_session, country="VN", year=2026) is not None
    assert await svc.get_calendar(db_session, country="VN", year=2027) is not None
    coverage = await svc.coverage(db_session, countries=["VN"], year=2027)
    assert coverage == {"VN": "covered"}
    # And the January days really are returned for 2027.
    rows = await svc.list_days(db_session, countries=["VN"], year=2027)
    assert {r.day for r in rows} == {date(2027, 1, 1), date(2027, 1, 2)}


async def test_cross_year_replace_also_covers_the_new_year(
    db_session: AsyncSession,
):
    group_key = await _seed_group(
        db_session, country="VN", name_en="Year End", days=[date(2026, 12, 30)],
    )
    await svc.replace_group(
        db_session,
        country="VN",
        group_key=group_key,
        name_en="Year End",
        name_local=None,
        days=[date(2026, 12, 30), date(2026, 12, 31), date(2027, 1, 1)],
    )
    await db_session.commit()
    assert await svc.get_calendar(db_session, country="VN", year=2027) is not None


async def test_cross_year_group_rejects_an_out_of_range_second_half(
    db_session: AsyncSession,
):
    """Validating only the first date let a bad year through on the other half."""
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError):
        await svc.create_group(
            db_session,
            country="VN",
            name_en="Impossible",
            name_local=None,
            days=[date(2026, 12, 31), date(2101, 1, 1)],
        )


async def test_group_search_treats_like_wildcards_literally(
    db_session: AsyncSession,
):
    """A search for "%" matched everything before the needle was escaped."""
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1)],
    )
    await _seed_group(
        db_session, country="CN", name_en="100% Bonus Day", days=[date(2026, 6, 1)],
    )

    # "%" must match only the holiday whose name literally contains it.
    hits = await svc.search_groups(db_session, year=2026, query="%")
    assert [g.name_en for g in hits] == ["100% Bonus Day"]

    # "_" likewise — it must not behave as "any single character".
    assert await svc.search_groups(db_session, year=2026, query="_") == []

    # A normal search still works, including on the local name.
    assert [g.name_en for g in await svc.search_groups(
        db_session, year=2026, query="劳动",
    )] == ["Labor Day Holiday"]


async def test_gcal_url_rejects_the_host_spoofing_shapes(db_session: AsyncSession):
    """The allow-list must not be fooled by userinfo or a suffix domain."""
    from app.core.exceptions import ValidationError

    for hostile in (
        "https://calendar.google.com@evil.com/x",
        "https://calendar.google.com.evil.com/x",
        "https://evil.com/?x=calendar.google.com",
        "http://calendar.google.com/x",
        "//calendar.google.com/x",
        "javascript:alert(1)",
        "https://xn--clendar-4va.google.com/x",
    ):
        with pytest.raises(ValidationError):
            await svc.upsert_calendar(
                db_session, country="CN", year=2026, gcal_url=hostile,
            )


def test_group_base_key_normalises_observed_and_makeup_wrappers():
    """The bug this guards was inside a readability fix, not the original code.

    The dataset pull named a makeup row after the *stripped* base ("Labor Day")
    while the rest days it offsets kept their marker ("Labor Day (observed)"), so
    the two landed in different groups: Labor Day split in half and each half
    reported ``observed_days = 0`` — hidden by the ``max(0, …)`` floor.
    """
    from app.domains.hr.holiday_keys import group_base_key

    family = {
        group_base_key("Labor Day"),
        group_base_key("Labor Day (observed)"),
        group_base_key("Makeup Workday (Labor Day)", is_makeup=True),
        group_base_key("Makeup Workday (Labor Day (observed))", is_makeup=True),
    }
    assert len(family) == 1, family
    # Two genuinely different holidays must still separate.
    assert group_base_key("Easter Monday") != group_base_key("Good Friday")


async def test_dataset_suggestion_groups_a_holiday_with_its_makeup_days(
    db_session: AsyncSession,
):
    """End-to-end on the real dataset: pull CN 2026, save, check the groups.

    Before the grouping fix this produced four groups for Labor Day and National
    Day (each split by the "(observed)" marker) with observed_days = 0 on the
    halves that carried the makeup workdays.
    """
    from app.domains.hr import holiday_source

    outcome = holiday_source.suggest("CN", 2026)
    assert outcome.supported and outcome.rows
    await svc.bulk_upsert(
        db_session,
        rows=[
            svc.BulkRow(
                country="CN", day=row.day, name_en=row.name_en,
                name_local=row.name_local, day_type=row.day_type,
            )
            for row in outcome.rows
        ],
        source="import",
    )
    await db_session.commit()

    by_name: dict[str, svc.HolidayGroup] = {}
    for group in svc.build_groups(await svc.list_days(db_session, countries=["CN"])):
        by_name[group.name_en] = group

    labor = next(g for k, g in by_name.items() if k.startswith("Labor Day"))
    assert labor.makeup_days == [date(2026, 5, 9)]
    assert labor.observed_days > 0, "the makeup day was attached to a split group"

    national = next(g for k, g in by_name.items() if k.startswith("National Day"))
    assert national.makeup_days == [date(2026, 9, 20), date(2026, 10, 10)]
    assert national.observed_days > 0


async def test_cross_year_group_is_whole_in_a_year_scoped_view(
    db_session: AsyncSession,
):
    """A year query must not report a truncated day count for a cross-year holiday.

    Folding only the visible half gave the employee page "1 day" for 2026 while
    the notice HR sends — fetched by group_key, therefore unclipped — said 3.
    """
    await svc.create_group(
        db_session,
        country="VN",
        name_en="New Year Holidays",
        name_local=None,
        days=[date(2026, 12, 31), date(2027, 1, 1), date(2027, 1, 2)],
    )
    await db_session.commit()

    for year in (2026, 2027):
        rows = await svc.list_days(db_session, countries=["VN"], year=year)
        groups = await svc.complete_groups(db_session, rows)
        assert len(groups) == 1, year
        assert groups[0].total_days == 3, f"{year}: {groups[0].days}"
        assert groups[0].start == date(2026, 12, 31)
        assert groups[0].end == date(2027, 1, 2)

    # And a group nowhere near the window edge is not needlessly refetched.
    await _seed_group(
        db_session, country="CN", name_en="Mid Year", days=[date(2026, 6, 15)],
    )
    mid = await svc.complete_groups(
        db_session,
        await svc.list_days(db_session, countries=["CN"], year=2026),
    )
    assert [g.total_days for g in mid] == [1]


async def test_public_view_reports_the_whole_cross_year_holiday(
    client: AsyncClient, employee_headers, db_session: AsyncSession,
):
    await svc.create_group(
        db_session, country="VN", name_en="New Year Holidays", name_local=None,
        days=[date(2026, 12, 31), date(2027, 1, 1)],
    )
    await db_session.commit()
    resp = await client.get(
        f"{BASE}/public?year=2026&scope=all", headers=employee_headers,
    )
    groups = resp.json()["data"]["groups"]
    assert len(groups) == 1
    assert groups[0]["total_days"] == 2


# ──────────────────────── notice generator ────────────────────────


def test_zh_notice_matches_the_wording_hr_publishes():
    # Layout matches, line for line, a notice HR actually sent to the company
    # (2026 劳动节). Two things were wrong until 2026-08-12 and both were
    # reported from a real paste into a Feishu group: the bullets were Markdown
    # ``* `` (Feishu renders nothing on paste, so an asterisk showed up
    # literally) and there were spurious blank lines after the intro and after
    # the tips heading. Golden text, so a future "tidy-up" has to be deliberate.
    expected = "\n".join([
        "2026 年劳动节放假通知",
        "",
        "@All 劳动节将至，为方便伙伴们提前做好假期安排，"
        "2026 年劳动节放假安排通知如下：",
        "• 5 月 1 日（星期五）至 5 月 5 日（星期二）放假，共 5 天；",
        "• 5 月 9 日（周六）补班，正常上班。",
        "",
        "「温馨提示」",
        "• 保管财物：请妥善保管个人贵重物品，办公笔记本请务必随身携带；",
        "• 检查电源：离岗前请确认关闭显示器、主机、插线板等所有设备电源，杜绝安全隐患；",
        "• 锁闭门窗 & 关闭电器：放假期间如需进出办公室，请随手关门，离开时确认灯、门窗、"
        "空调等设施设备处于关闭状态；",
        "• 出行安全：假期出行，请注意人身与财产安全",
        "",
        "五月初始，夏意未满，趁微风正好，让身心小憩。致敬每一份辛勤付出，"
        "愿大家假期愉快，节后见！ ❤️",
    ])
    got = render_notice(_labor_day_group(), NoticeRequest(language=ZH))
    assert got["text"] == expected


def test_notice_never_emits_markdown_that_pastes_as_punctuation():
    """The notice is pasted, never rendered — so no Markdown may appear.

    Feishu group messages and email clients paste plain text: a ``* `` bullet
    arrives as a visible asterisk and ``**bold**`` as four of them. That is the
    exact "复制出来不能直接用…很丑" report. Checked across both languages and
    with tips on, because the tips block is where the temptation to bold a
    label lives.
    """
    for lang in (ZH, EN):
        text = render_notice(
            _labor_day_group(), NoticeRequest(language=lang, include_tips=True),
        )["text"]
        for line in text.split("\n"):
            assert not line.startswith("* "), line
            assert not line.startswith("- "), line
            assert "**" not in line, line
        # And the bullets that ARE there are real bullet characters.
        assert "\u2022 " in text


def test_notice_closing_line_is_replaceable_and_can_be_dropped():
    custom = render_notice(
        _labor_day_group(),
        NoticeRequest(language=ZH, closing_zh="祝各位假期愉快", include_emoji=False),
    )
    assert custom["text"].endswith("祝各位假期愉快")
    none = render_notice(_labor_day_group(), NoticeRequest(language=ZH, closing_zh=""))
    assert none["text"].rstrip().endswith("出行安全：假期出行，请注意人身与财产安全")


def test_notice_omits_the_makeup_line_when_there_is_no_makeup_day():
    group = svc.HolidayGroup(
        country="JP", group_key="jp-2026-marine-day", name_en="Marine Day",
        name_local=None, note=None, days=[date(2026, 7, 20)], makeup_days=[],
    )
    text = render_notice(group, NoticeRequest(language=ZH))["text"]
    assert "补班" not in text
    assert "\u2022 7 月 20 日（星期一）放假。" in text


def test_notice_spaces_a_latin_holiday_name_inside_chinese_text():
    group = svc.HolidayGroup(
        country="JP", group_key="k", name_en="Marine Day", name_local=None,
        note=None, days=[date(2026, 7, 20)], makeup_days=[],
    )
    text = render_notice(
        group, NoticeRequest(language=ZH, region_label="日本"),
    )["text"]
    assert text.startswith("2026 年日本 Marine Day 放假通知")
    # No double space after "@All" — padding both sides unconditionally caused one.
    assert "@All  " not in text


def test_notice_spaces_an_ascii_name_after_the_year_with_no_region_label():
    """``"2026 年Founders Day 放假通知"`` was missing the space after 年."""
    group = svc.HolidayGroup(
        country="US", group_key="k", name_en="Founders Day", name_local=None,
        note=None, days=[date(2026, 3, 2)], makeup_days=[],
    )
    text = render_notice(group, NoticeRequest(language=ZH))["text"]
    assert text.startswith("2026 年 Founders Day 放假通知")
    assert "@All  " not in text


def test_en_notice_does_not_repeat_the_word_holiday():
    text = render_notice(_labor_day_group(), NoticeRequest(language=EN))["text"]
    assert text.startswith("Holiday Notice — Labor Day Holiday, 2026")
    assert "Holiday Holiday Notice" not in text


def test_bilingual_notice_carries_both_bodies():
    payload = render_notice(_labor_day_group(), NoticeRequest(language=BILINGUAL))
    assert payload["text_zh"] and payload["text_en"]
    assert payload["text_zh"] in payload["text"]
    assert payload["text_en"] in payload["text"]


def test_notice_refuses_a_makeup_only_group():
    group = svc.HolidayGroup(
        country="CN", group_key="k", name_en="Makeup Workday (X)", name_local=None,
        note=None, days=[], makeup_days=[date(2026, 5, 9)],
    )
    with pytest.raises(ValueError):
        render_notice(group, NoticeRequest(language=ZH))


def test_notice_facts_expose_the_derived_numbers():
    facts = render_notice(_labor_day_group(), NoticeRequest(language=ZH))["facts"]
    assert facts["total_days"] == 5
    assert facts["observed_days"] == 2
    assert facts["segments"] == [
        {"start": "2026-05-01", "end": "2026-05-05", "days": 5}
    ]
    assert facts["makeup_days"] == ["2026-05-09"]


# ──────────────────────── exports ────────────────────────


def test_adjustment_sentence_matches_the_source_document_wording():
    from app.domains.hr.holiday_export import adjustment_sentence

    single = svc.HolidayGroup(
        country="CN", group_key="k", name_en="New Year Holiday", name_local="元旦",
        note=None, days=[date(2026, 1, d) for d in (1, 2, 3)],
        makeup_days=[date(2026, 1, 4)],
    )
    assert adjustment_sentence(single) == (
        "3 days off with adjustment. January 4 (Sunday) is a working day."
    )
    plural = svc.HolidayGroup(
        country="CN", group_key="k2", name_en="National Day Holiday",
        name_local="国庆", note=None,
        days=[date(2026, 10, d) for d in range(1, 8)],
        makeup_days=[date(2026, 9, 20), date(2026, 10, 10)],
    )
    assert adjustment_sentence(plural) == (
        "7 days off with adjustment. September 20 (Sunday) and "
        "October 10 (Saturday) are working days."
    )
    assert adjustment_sentence(
        svc.HolidayGroup(
            country="JP", group_key="k3", name_en="Marine Day", name_local=None,
            note=None, days=[date(2026, 7, 20)], makeup_days=[],
        )
    ) is None


def test_docx_export_rebuilds_every_seeded_country():
    """The Word export must survive a re-parse with every country intact."""
    from docx import Document

    from app.domains.hr.holiday_export import build_docx
    from app.domains.hr.holiday_seed_2026 import SEED_2026, SEED_YEAR

    rows, calendars = [], []
    for idx, (code, spec) in enumerate(SEED_2026.items(), start=1):
        calendars.append(HrPublicHolidayCalendar(
            country=code, year=SEED_YEAR, label=spec["label"],
            stated_days=spec["stated_days"], gcal_url=spec["gcal_url"],
            region_note=spec["region_note"],
        ))
        for day, group_key, name_en, name_local, day_type in spec["days"]:
            rows.append(HrPublicHoliday(
                id=idx * 1000 + len(rows), country=code,
                day=date.fromisoformat(day), group_key=group_key,
                name_en=name_en, name_local=name_local,
                name_key=name_key_for(name_en), day_type=day_type, source="seed",
            ))

    buf = build_docx(year=SEED_YEAR, rows=rows, calendars=calendars)
    doc = Document(io.BytesIO(buf.getvalue()))
    text = "\n".join(p.text for p in doc.paragraphs)
    assert "PingCAP Public Holidays for Global Team Members 2026" in text
    # One numbered section per country, and the headline count is HR's own.
    assert "1. PingCAP observes the following Australia Holidays - 8 days" in text
    assert "PingCAP observes the following Italy Holidays - 8 days" in text
    # 22 country tables plus the 5 regional-advisory single-cell tables.
    assert len(doc.tables) == 27
    # Every subscription link made it through as a real hyperlink relationship.
    link_targets = {
        rel.target_ref for rel in doc.part.rels.values() if rel.is_external
    }
    assert len(link_targets) == 22


def test_excel_rows_use_the_import_parser_aliases():
    """An export must be re-importable without renaming a single column."""
    from app.domains.hr.holiday_export import EXPORT_COLUMNS
    from app.domains.hr.holiday_import import parse_rows

    rows = [
        HrPublicHoliday(
            id=1, country="CN", day=date(2026, 5, 1), group_key="g",
            name_en="Labor Day Holiday", name_local="劳动节",
            name_key="labor-day-holiday", day_type=PUBLIC_HOLIDAY, source="seed",
        ),
    ]
    from app.domains.hr.holiday_export import excel_rows

    flat = excel_rows(rows, {"CN": "China"})
    # Re-key by the human header labels, the way openpyxl hands them back.
    header_by_key = dict(EXPORT_COLUMNS)
    sheet_row = {header_by_key[k]: v for k, v in flat[0].items()}
    report = parse_rows([sheet_row])
    assert report.missing_required == []
    assert len(report.rows) == 1
    parsed = report.rows[0]
    assert (parsed.status, parsed.country, parsed.day) == (
        "new", "CN", date(2026, 5, 1),
    )
    assert parsed.name_local == "劳动节"
    assert parsed.day_type == PUBLIC_HOLIDAY


# ──────────────────────── import parser ────────────────────────


def test_parse_date_accepts_every_shape_hr_uses():
    from app.domains.hr.holiday_import import parse_date

    assert parse_date("2026-05-01") == date(2026, 5, 1)
    assert parse_date("Friday, May 1, 2026") == date(2026, 5, 1)
    assert parse_date("May 1, 2026") == date(2026, 5, 1)
    assert parse_date("2026年5月1日") == date(2026, 5, 1)
    assert parse_date("2026/05/01") == date(2026, 5, 1)
    assert parse_date(date(2026, 5, 1)) == date(2026, 5, 1)
    assert parse_date("not a date") is None
    assert parse_date("") is None


def test_date_regexes_stay_linear_on_hostile_input():
    """Guards against the ReDoS the first version of these patterns shipped with.

    The original long-date pattern opened with an optional ``[A-Za-z]+`` weekday
    prefix before a generic ``[A-Za-z]+`` month, so both groups could consume the
    same letters: 8,000 letters cost ~650 ms of backtracking, and HR pastes up to
    20,000 characters of untrusted text through this path. Month names are now a
    literal alternation and every whitespace run is bounded.
    """
    import time

    from app.domains.hr.holiday_ai_service import _LINE_CJK, _LINE_LONG
    from app.domains.hr.holiday_import import _CJK_DATE, _LONG_DATE

    payloads = ["A" * 20_000, " " * 20_000, "January " * 2_500, "2026" * 5_000]
    for pattern in (_LONG_DATE, _CJK_DATE, _LINE_LONG, _LINE_CJK):
        for payload in payloads:
            started = time.perf_counter()
            pattern.search(payload)
            elapsed_ms = (time.perf_counter() - started) * 1000
            assert elapsed_ms < 100, (
                f"{pattern.pattern[:40]!r} took {elapsed_ms:.0f}ms on a "
                f"{len(payload)}-char payload — the pattern is backtracking"
            )


def test_long_date_parsing_survived_the_redos_fix():
    """Every date shape the source document and HR spreadsheets actually use."""
    from app.domains.hr.holiday_import import parse_date

    assert parse_date("Thursday, January 1, 2026") == date(2026, 1, 1)
    assert parse_date("Monday January 19, 2026") == date(2026, 1, 19)
    assert parse_date("Friday, December 25 , 2026") == date(2026, 12, 25)
    assert parse_date("Jan 1, 2026") == date(2026, 1, 1)
    assert parse_date("SEPTEMBER 26, 2026") == date(2026, 9, 26)
    assert parse_date("june 8, 2026") == date(2026, 6, 8)
    assert parse_date("May 1 2026") == date(2026, 5, 1)
    # A word that merely looks like a month must not resolve.
    assert parse_date("Januaryish 1, 2026") is None


def test_parse_rows_reports_errors_per_line_without_dropping_good_rows():
    from app.domains.hr.holiday_import import parse_rows

    report = parse_rows([
        {"Country": "CN", "Date": "2026-05-01", "Holiday": "Labor Day"},
        {"Country": "Freedonia", "Date": "2026-05-02", "Holiday": "Bad Country"},
        {"Country": "JP", "Date": "gibberish", "Holiday": "Bad Date"},
        {"Country": "JP", "Date": "2026-07-20", "Holiday": ""},
    ])
    assert report.counts() == {"new": 1, "duplicate": 0, "error": 3}
    assert [r.line for r in report.rows] == [2, 3, 4, 5]
    assert "unrecognised country" in (report.rows[1].message or "")
    assert "unreadable date" in (report.rows[2].message or "")
    assert "name is empty" in (report.rows[3].message or "")


def test_parse_rows_flags_duplicates_against_the_database_and_the_file():
    from app.domains.hr.holiday_import import parse_rows

    existing = {("CN", "2026-05-01", name_key_for("Labor Day"))}
    report = parse_rows(
        [
            {"Country": "CN", "Date": "2026-05-01", "Holiday": "Labor Day"},
            {"Country": "CN", "Date": "2026-05-02", "Holiday": "Labor Day 2"},
            {"Country": "CN", "Date": "2026-05-02", "Holiday": "Labor Day 2"},
        ],
        existing_keys=existing,
    )
    assert report.counts() == {"new": 1, "duplicate": 2, "error": 0}


def test_parse_rows_accepts_chinese_headers_and_type_words():
    from app.domains.hr.holiday_import import parse_rows

    report = parse_rows([
        {"国家": "中国", "日期": "2026年5月1日", "假期名称": "Labor Day",
         "本地名称": "劳动节", "类型": "假期"},
        {"国家": "中国", "日期": "2026年5月9日", "假期名称": "Makeup",
         "本地名称": "调休", "类型": "补班"},
    ])
    assert report.missing_required == []
    assert [r.status for r in report.rows] == ["new", "new"]
    assert [r.day_type for r in report.rows] == [PUBLIC_HOLIDAY, MAKEUP_WORKDAY]


def test_parse_rows_reports_missing_required_headers():
    from app.domains.hr.holiday_import import parse_rows

    report = parse_rows([{"Date": "2026-05-01", "Something": "x"}])
    assert set(report.missing_required) == {"country", "name_en"}
    assert report.rows == []


# ──────────────────────── HR API ────────────────────────


async def test_employee_cannot_reach_the_hr_surface(
    client: AsyncClient, employee_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    for method, url in (
        ("get", f"{BASE}/calendars?year=2026"),
        ("get", f"{BASE}/groups?year=2026"),
        ("get", f"{BASE}/days?year=2026"),
        ("get", f"{BASE}/export.xlsx?year=2026"),
        ("post", f"{BASE}/restore-defaults"),
    ):
        resp = await getattr(client, method)(url, headers=employee_headers)
        assert resp.status_code == 403, f"{method} {url} -> {resp.status_code}"


async def test_hr_can_create_read_and_delete_a_group(
    client: AsyncClient, hr_headers,
):
    create = await client.post(
        f"{BASE}/groups",
        headers=hr_headers,
        json={
            "country": "CN",
            "name_en": "Labor Day Holiday",
            "name_local": "劳动节",
            "days": ["2026-05-01", "2026-05-02", "2026-05-03",
                     "2026-05-04", "2026-05-05"],
            "makeup_days": ["2026-05-09"],
        },
    )
    assert create.status_code == 201, create.text
    group = create.json()["data"]
    assert group["total_days"] == 5
    assert group["observed_days"] == 2
    assert group["makeup_days"] == ["2026-05-09"]

    listing = await client.get(f"{BASE}/groups?year=2026", headers=hr_headers)
    assert listing.status_code == 200
    assert len(listing.json()["data"]["groups"]) == 1

    # The calendar row is created alongside, so coverage is never a lie.
    calendars = await client.get(f"{BASE}/calendars?year=2026", headers=hr_headers)
    payload = calendars.json()["data"]["countries"]
    assert [c["country"] for c in payload] == ["CN"]
    assert payload[0]["observed_days"] == 2

    removed = await client.delete(
        f"{BASE}/groups/CN/{group['group_key']}", headers=hr_headers,
    )
    assert removed.status_code == 200
    assert removed.json()["data"]["rows_deleted"] == 6


async def test_group_create_rejects_an_unknown_country(
    client: AsyncClient, hr_headers,
):
    resp = await client.post(
        f"{BASE}/groups",
        headers=hr_headers,
        json={"country": "Freedonia", "name_en": "X", "days": ["2026-05-01"]},
    )
    assert resp.status_code == 422


async def test_group_update_requires_the_path_and_body_country_to_agree(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    group_key = await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    resp = await client.put(
        f"{BASE}/groups/JP/{group_key}",
        headers=hr_headers,
        json={"country": "CN", "name_en": "Labor Day", "days": ["2026-05-01"]},
    )
    assert resp.status_code == 422


async def test_day_endpoints_round_trip(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1)],
    )
    added = await client.post(
        f"{BASE}/days",
        headers=hr_headers,
        json={"country": "CN", "day": "2026-05-02", "name_en": "Labor Day Holiday"},
    )
    assert added.status_code == 201, added.text
    day_id = added.json()["data"]["id"]

    # A second identical add is a conflict, not a duplicate row.
    again = await client.post(
        f"{BASE}/days",
        headers=hr_headers,
        json={"country": "CN", "day": "2026-05-02", "name_en": "Labor Day Holiday"},
    )
    assert again.status_code in (400, 409)

    patched = await client.put(
        f"{BASE}/days/{day_id}",
        headers=hr_headers,
        json={"note": "moved by government notice", "day": "2026-05-06"},
    )
    assert patched.status_code == 200
    assert patched.json()["data"]["day"] == "2026-05-06"

    deleted = await client.delete(f"{BASE}/days/{day_id}", headers=hr_headers)
    assert deleted.status_code == 200


async def test_days_endpoint_returns_a_month_matrix(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday",
        days=[date(2026, 5, 1), date(2026, 5, 2)], makeup=[date(2026, 5, 9)],
    )
    resp = await client.get(f"{BASE}/days?year=2026&month=5", headers=hr_headers)
    matrix = resp.json()["data"]["matrix"]
    assert set(matrix) == {"2026-05-01", "2026-05-02", "2026-05-09"}
    assert matrix["2026-05-09"][0]["day_type"] == MAKEUP_WORKDAY
    # A different month must be empty rather than absent-and-confusing.
    other = await client.get(f"{BASE}/days?year=2026&month=6", headers=hr_headers)
    assert other.json()["data"]["matrix"] == {}


async def test_bulk_endpoint_reports_created_and_skipped(
    client: AsyncClient, hr_headers,
):
    body = {
        "rows": [
            {"country": "CN", "day": "2026-10-01", "name_en": "National Day"},
            {"country": "CN", "day": "2026-10-02", "name_en": "National Day"},
        ],
        "source": "import",
    }
    first = await client.post(f"{BASE}/bulk", headers=hr_headers, json=body)
    assert first.json()["data"]["created"] == 2
    second = await client.post(f"{BASE}/bulk", headers=hr_headers, json=body)
    assert second.json()["data"]["skipped"] == 2


async def test_restore_defaults_endpoint_seeds_the_baseline(
    client: AsyncClient, hr_headers,
):
    resp = await client.post(f"{BASE}/restore-defaults", headers=hr_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["days_added"] == 328
    assert data["calendars_added"] == 22
    assert data["year"] == 2026


async def test_notice_endpoint_renders_from_stored_data(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    group_key = await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, d) for d in (1, 2, 3, 4, 5)],
        makeup=[date(2026, 5, 9)],
    )
    resp = await client.post(
        f"{BASE}/notice",
        headers=hr_headers,
        json={"country": "CN", "group_key": group_key, "language": "zh"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["text"].startswith("2026 年劳动节放假通知")
    assert "\u2022 5 月 9 日（周六）补班，正常上班。" in data["text"]
    assert data["facts"]["observed_days"] == 2


async def test_notice_endpoint_labels_a_non_china_holiday(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    group_key = await _seed_group(
        db_session, country="JP", name_en="Marine Day", days=[date(2026, 7, 20)],
        label="Japan",
    )
    resp = await client.post(
        f"{BASE}/notice",
        headers=hr_headers,
        json={"country": "JP", "group_key": group_key, "language": "zh"},
    )
    # A space on BOTH sides of the Latin run: after 年 and before 放假通知.
    assert resp.json()["data"]["text"].startswith("2026 年 Japan Marine Day 放假通知")


async def test_notice_endpoint_404s_for_an_unknown_group(
    client: AsyncClient, hr_headers,
):
    resp = await client.post(
        f"{BASE}/notice",
        headers=hr_headers,
        json={"country": "CN", "group_key": "nope", "language": "zh"},
    )
    assert resp.status_code == 404


async def test_xlsx_and_docx_exports_download(
    client: AsyncClient, hr_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1)], stated_days=1,
        gcal_url="https://calendar.google.com/calendar/u/0?cid=abc",
    )
    xlsx = await client.get(f"{BASE}/export.xlsx?year=2026", headers=hr_headers)
    assert xlsx.status_code == 200
    assert xlsx.headers["X-Export-Returned"] == "1"
    assert xlsx.content[:2] == b"PK"

    docx = await client.get(f"{BASE}/export.docx?year=2026", headers=hr_headers)
    assert docx.status_code == 200
    assert docx.content[:2] == b"PK"
    assert "public_holidays" in docx.headers["content-disposition"].lower() or \
        "pingcap_public_holidays" in docx.headers["content-disposition"].lower()


async def test_docx_export_404s_when_there_is_nothing_to_export(
    client: AsyncClient, hr_headers,
):
    resp = await client.get(f"{BASE}/export.docx?year=2035", headers=hr_headers)
    assert resp.status_code == 404


async def test_every_write_endpoint_refuses_a_read_only_role(
    client: AsyncClient, auditor_headers, db_session: AsyncSession,
):
    """The gate matrix, not a sample of it.

    The first version of this test covered 5 of 26 endpoints and none of them
    with a read-only *governance* role — an auditor passes ``require_hr_read``,
    so every write had to be checked against exactly that, not against an
    ordinary employee who fails both gates.
    """
    group_key = await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    day_id = (await svc.list_days(db_session, countries=["CN"]))[0].id

    writes = [
        ("post", f"{BASE}/groups",
         {"country": "CN", "name_en": "X", "days": ["2026-07-01"]}),
        ("put", f"{BASE}/groups/CN/{group_key}",
         {"country": "CN", "name_en": "X", "days": ["2026-05-01"]}),
        ("delete", f"{BASE}/groups/CN/{group_key}", None),
        ("post", f"{BASE}/days",
         {"country": "CN", "day": "2026-07-02", "name_en": "X"}),
        ("put", f"{BASE}/days/{day_id}", {"note": "x"}),
        ("delete", f"{BASE}/days/{day_id}", None),
        ("post", f"{BASE}/bulk",
         {"rows": [{"country": "CN", "day": "2026-07-03", "name_en": "X"}]}),
        ("put", f"{BASE}/calendars/CN/2026", {"stated_days": 1}),
        ("delete", f"{BASE}/calendars/CN/2026", None),
        ("post", f"{BASE}/restore-defaults", None),
        ("post", f"{BASE}/calendars/roll-forward",
         {"from_year": 2026, "to_year": 2027}),
        ("post", f"{BASE}/suggest", {"country": "CN", "year": 2027}),
        ("post", f"{BASE}/ai/parse", {"text": "2026-05-01 X"}),
        # Composing an all-hands announcement is an HR act, so these two are
        # write-gated even though they only return text.
        ("post", f"{BASE}/notice", {"country": "CN", "group_key": group_key}),
        ("post", f"{BASE}/notice/closing",
         {"country": "CN", "group_key": group_key}),
    ]
    for method, url, body in writes:
        kwargs = {"headers": auditor_headers}
        if body is not None:
            kwargs["json"] = body
        resp = await getattr(client, method)(url, **kwargs)
        assert resp.status_code == 403, f"{method.upper()} {url} -> {resp.status_code}"


async def test_auditor_may_still_read_the_console(
    client: AsyncClient, auditor_headers, db_session: AsyncSession,
):
    """The read surface is intentionally wider than the write surface."""
    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    for url in (
        f"{BASE}/years",
        f"{BASE}/countries",
        f"{BASE}/calendars?year=2026",
        f"{BASE}/groups?year=2026",
        f"{BASE}/days?year=2026",
        f"{BASE}/export.xlsx?year=2026",
    ):
        resp = await client.get(url, headers=auditor_headers)
        assert resp.status_code == 200, f"{url} -> {resp.status_code}"


async def test_mcp_notice_tool_is_hr_only(db_session: AsyncSession):
    """The one security-relevant line in the MCP surface, previously untested.

    ``tool_runtime`` enforces active-employee status and nothing more, and any
    employee can mint their own PAT — so this explicit role check is the only
    thing standing between an ordinary employee and an all-hands announcement
    drafted in the company's voice.
    """
    from app.core.rbac import GLOBAL_ADMIN_ROLE, HR_ROLE, has_role

    assert has_role(["hr"], GLOBAL_ADMIN_ROLE, HR_ROLE)
    assert has_role(["global_admin"], GLOBAL_ADMIN_ROLE, HR_ROLE)
    # The legacy alias must pass, or an it_admin loses access on rename.
    assert has_role(["it_admin"], GLOBAL_ADMIN_ROLE, HR_ROLE)
    # And the roles that must NOT.
    for denied in (["employee"], ["auditor"], ["management"], ["finance"], []):
        assert not has_role(denied, GLOBAL_ADMIN_ROLE, HR_ROLE), denied

    # The tool really does perform that check — assert on the source rather than
    # trusting the comment, since this is the line that must not silently go.
    import inspect

    from app.mcp import tools

    source = inspect.getsource(tools.generate_holiday_notice)
    assert "has_role(" in source and "HR_ROLE" in source
    assert "ForbiddenError" in source


async def test_restore_defaults_does_not_resurrect_a_moved_seed_row(
    db_session: AsyncSession,
):
    """Moving a seeded holiday must not give HR two of it on the next restore."""
    await svc.restore_seed(db_session)
    await db_session.commit()
    ching_ming = next(
        r for r in await svc.list_days(db_session, countries=["HK"])
        if r.day == date(2026, 4, 4) and "Ching Ming" in r.name_en
    )
    before = len(await svc.list_days(db_session, countries=["HK"]))
    await svc.update_day(db_session, day_id=ching_ming.id, day=date(2026, 4, 6))
    await db_session.commit()

    await svc.restore_seed(db_session)
    await db_session.commit()
    after = await svc.list_days(db_session, countries=["HK"])
    assert len(after) == before, "the seed re-inserted the row HR had moved"
    assert date(2026, 4, 4) not in {r.day for r in after if "Ching Ming" in r.name_en}


async def test_restore_defaults_does_not_undo_a_deliberate_clear(
    db_session: AsyncSession,
):
    """``clear_*`` must survive a restore — a blank field is not "never set"."""
    await svc.restore_seed(db_session)
    await db_session.commit()
    await svc.upsert_calendar(
        db_session, country="AU", year=2026, clear_gcal=True, clear_stated=True,
    )
    await db_session.commit()

    await svc.restore_seed(db_session)
    await db_session.commit()
    cal = await svc.get_calendar(db_session, country="AU", year=2026)
    assert cal is not None
    assert cal.gcal_url is None, "restore re-filled a link HR deliberately cleared"
    assert cal.stated_days is None


# ──────────────────────── employee 公示 ────────────────────────


async def test_public_view_defaults_to_the_readers_own_country(
    client: AsyncClient, db_session: AsyncSession,
):
    from app.core.security import create_access_token

    await _seed_group(
        db_session, country="CN", name_en="Labor Day Holiday", name_local="劳动节",
        days=[date(2026, 5, 1)], label="China",
    )
    await _seed_group(
        db_session, country="JP", name_en="Marine Day", days=[date(2026, 7, 20)],
        label="Japan",
    )
    person = await _person(db_session, residence="JP", country="JPN")
    headers = {"Authorization": f"Bearer {create_access_token(str(person.id), ['employee'])}"}

    resp = await client.get(f"{BASE}/public?year=2026", headers=headers)
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["my_country"] == "JP"
    assert data["selected_countries"] == ["JP"]
    assert [g["name_en"] for g in data["groups"]] == ["Marine Day"]


async def test_public_view_widens_when_the_readers_country_is_unmaintained(
    client: AsyncClient, db_session: AsyncSession,
):
    from app.core.security import create_access_token

    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    # Portugal has no calendar: show the company-wide view, not a blank page.
    person = await _person(db_session, residence="PT")
    headers = {"Authorization": f"Bearer {create_access_token(str(person.id), ['employee'])}"}
    resp = await client.get(f"{BASE}/public?year=2026", headers=headers)
    data = resp.json()["data"]
    assert data["my_country"] == "PT"
    assert data["selected_countries"] == ["CN"]
    assert len(data["groups"]) == 1


async def test_public_view_supports_all_and_custom_scopes(
    client: AsyncClient, employee_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
    )
    await _seed_group(
        db_session, country="JP", name_en="Marine Day", days=[date(2026, 7, 20)],
    )
    all_scope = await client.get(
        f"{BASE}/public?year=2026&scope=all", headers=employee_headers,
    )
    assert len(all_scope.json()["data"]["groups"]) == 2

    custom = await client.get(
        f"{BASE}/public?year=2026&scope=custom&countries=JP",
        headers=employee_headers,
    )
    assert [g["country"] for g in custom.json()["data"]["groups"]] == ["JP"]

    bad = await client.get(
        f"{BASE}/public?year=2026&scope=custom", headers=employee_headers,
    )
    assert bad.status_code == 422


async def test_public_upcoming_marks_an_in_progress_holiday(
    client: AsyncClient, employee_headers, db_session: AsyncSession,
):
    from datetime import datetime, timedelta

    from app.core.working_hours import resolve_timezone

    # The endpoint computes "today" in the CALLER's timezone
    # (``resolve_timezone``, default Asia/Shanghai), so seeding around a
    # UTC ``date.today()`` puts the holiday a day off for the eight hours
    # between 16:00 UTC and midnight — the run at 2026-08-12 16:08 UTC
    # got ``days_until == -2``. Seed against the same clock the endpoint
    # reads.
    probe = IdentityPerson(
        id=uuid.uuid4(), work_email="tz-probe@x.io", display_name="tz",
        employment_status="active",
        source_system="successfactors", hris_person_id="sf-tz",
    )
    today = datetime.now(resolve_timezone(probe)).date()
    await _seed_group(
        db_session, country="CN", name_en="Ongoing Holiday",
        days=[today - timedelta(days=1), today, today + timedelta(days=1)],
        year=today.year,
    )
    resp = await client.get(
        f"{BASE}/public/upcoming?scope=all", headers=employee_headers,
    )
    assert resp.status_code == 200
    upcoming = resp.json()["data"]["upcoming"]
    assert upcoming and upcoming[0]["in_progress"] is True
    # Started yesterday, so days_until is negative — the card must still show it.
    assert upcoming[0]["days_until"] == -1
    assert upcoming[0]["total_days"] == 3


async def test_public_day_lookup_reports_coverage(
    client: AsyncClient, employee_headers, db_session: AsyncSession,
):
    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
        makeup=[date(2026, 5, 9)],
    )
    holiday = await client.get(
        f"{BASE}/public/day?day=2026-05-01&country=CN", headers=employee_headers,
    )
    assert holiday.json()["data"]["is_public_holiday"] is True

    makeup = await client.get(
        f"{BASE}/public/day?day=2026-05-09&country=CN", headers=employee_headers,
    )
    body = makeup.json()["data"]
    assert body["is_makeup_workday"] is True
    assert body["is_public_holiday"] is False

    workday = await client.get(
        f"{BASE}/public/day?day=2026-05-07&country=CN", headers=employee_headers,
    )
    assert workday.json()["data"]["is_public_holiday"] is False
    assert workday.json()["data"]["coverage"] == "covered"

    unmaintained = await client.get(
        f"{BASE}/public/day?day=2026-05-01&country=PT", headers=employee_headers,
    )
    assert unmaintained.json()["data"]["coverage"] == "no_data"


async def test_public_countries_marks_the_readers_own(
    client: AsyncClient, db_session: AsyncSession,
):
    from app.core.security import create_access_token

    await _seed_group(
        db_session, country="CN", name_en="Labor Day", days=[date(2026, 5, 1)],
        label="China",
    )
    person = await _person(db_session, residence="CN")
    headers = {"Authorization": f"Bearer {create_access_token(str(person.id), ['employee'])}"}
    resp = await client.get(f"{BASE}/public/countries?year=2026", headers=headers)
    data = resp.json()["data"]
    assert data["my_country"] == "CN"
    assert data["my_country_covered"] is True
    assert data["countries"][0]["label"] == "China"

def test_only_the_ai_endpoints_carry_the_sensitive_rate_limiter():
    """The sensitive-action limiter belongs on LLM calls, nothing else here.

    Its budget is GLOBAL per user (10 calls/hour, shared with PII reveal and
    every export in the product), so putting it on ordinary holiday endpoints
    was doubly wrong: holiday data is company-published and non-personal, and
    the notice dialog re-renders on every language / tips / emoji toggle — so
    clicking through a few holidays locked the user out of PII reveal product-
    wide for an hour. Reported from prod as 「随便点点就 Too many requests」.

    Pinned as a route-table assertion rather than a behavioural one because the
    failure mode is someone adding the dependency to a new endpoint "for
    consistency"; that has to fail here, loudly, with this explanation.
    """
    from app.core.dependencies import check_sensitive_endpoint_rate_limit
    from app.domains.hr.holiday_router import router

    limited = set()
    for route in router.routes:
        for dep in getattr(route, "dependant", None).dependencies if getattr(
            route, "dependant", None,
        ) else []:
            if dep.call is check_sensitive_endpoint_rate_limit:
                limited.add(f"{sorted(route.methods)[0]} {route.path}")
    assert limited == {
        "POST /api/v1/hr/holidays/ai/parse",
        "POST /api/v1/hr/holidays/notice/closing",
    }, limited

def test_poster_follows_the_edited_notice_text():
    """Editing the notice must change the image, not just the text.

    The first version built the poster from the structured render only, so HR
    could rewrite a sentence, download the picture and get the old wording —
    「我改了文字以后也不会改图片吗 那有啥用」. The poster is now rebuilt from the
    same text the textarea holds, through the SAME parser the rich-text copy
    uses, so the two can never disagree.
    """
    from app.domains.hr.holiday_notice import poster_from_text, render_notice

    rendered = render_notice(
        _labor_day_group(), NoticeRequest(language=ZH), poster_region="中国大陆",
    )
    base = dict(rendered["poster"])  # type: ignore[arg-type]
    edited = (
        "2026 年劳动节放假通知\n"
        "\n"
        "@All 今年劳动节我们放长一点。\n"
        "\u2022 5 月 1 日（星期五）至 5 月 6 日（星期三）放假，共 6 天；\n"
        "\u2022 5 月 9 日（周六）补班，正常上班。\n"
        "\n"
        "「安全提醒」\n"
        "\u2022 锁好门窗：离开前请确认门窗关闭；\n"
        "\n"
        "祝大家假期愉快。"
    )
    spec = poster_from_text(edited, base)

    # Wording comes from the edit …
    assert spec["intro"] == "@All 今年劳动节我们放长一点。"
    assert spec["lines"] == ["5 月 1 日（星期五）至 5 月 6 日（星期三）放假，共 6 天；"]
    assert spec["makeup_lines"] == ["5 月 9 日（周六）补班，正常上班。"]
    assert spec["tips_heading"] == "「安全提醒」"
    assert [t["label"] for t in spec["tips"]] == ["锁好门窗"]
    assert spec["closing"] == "祝大家假期愉快。"
    assert spec["holiday_name"] == "劳动节" and spec["year"] == 2026
    # … while what the text cannot express is kept from the structured render.
    assert spec["country"] == base["country"]
    assert spec["month"] == base["month"] and spec["day"] == base["day"]
    assert spec["region_label"] == "中国大陆"


def test_poster_keeps_a_hand_rewritten_title_whole():
    """A title that does not match 「YYYY 年…放假通知」 must not be chopped up."""
    from app.domains.hr.holiday_notice import poster_from_text, render_notice

    base = dict(render_notice(  # type: ignore[arg-type]
        _labor_day_group(), NoticeRequest(language=ZH),
    )["poster"])
    spec = poster_from_text(
        "Labour Day 2026 — office closed\n\n@All heads up.\n\u2022 May 1 — closed\n"
        "\nSee you after.",
        base,
    )
    assert spec["holiday_name"] == "Labour Day 2026 — office closed"
    assert spec["heading"] == ""


def test_poster_title_parser_handles_hostile_long_input_without_regex_backtracking():
    from app.domains.hr.holiday_notice import poster_from_text, render_notice

    base = dict(render_notice(  # type: ignore[arg-type]
        _labor_day_group(), NoticeRequest(language=ZH),
    )["poster"])
    hostile_title = "2026 年" + ("节" * 100_000) + "并非通知"
    spec = poster_from_text(hostile_title, base)

    assert spec["holiday_name"] == hostile_title
    assert spec["heading"] == ""