"""Leave-data sync: SF EC Time-Off → hr_leave_accounts / hr_leave_postings.

Guards the data-driven mirror that supersedes the four-column carryover
(see ``app/domains/hr/leave_service.py::sync_leave_data``):

  1. ALL account types are mirrored (not just legal+welfare) — each SF
     TimeAccount becomes its own row, so a comp-leave type with several
     SF accounts lands as several rows.
  2. Per-account balance/entitled/used are aggregated from the detail
     lines; per-line running ``balance_after`` is computed in SF
     chronological order.
  3. The four legacy ``identity_persons.annual_leave_*`` columns are
     derived from the synced data (so the 年假计算器 keeps working).
  4. Idempotent re-run keeps row values stable; an account SF stops
     returning is soft-deleted (tombstoned), not duplicated.
  5. Authoritative empty snapshot tombstones stale rows while skipping
     unnecessary detail/type/request calls.
"""
from __future__ import annotations

from datetime import UTC, date, datetime
from decimal import Decimal
from uuid import uuid4

import pytest
import pytest_asyncio
from sqlalchemy import func, select

from app.domains.hr.leave_models import HrLeaveAccount, HrLeavePosting
from app.domains.hr.leave_service import HrLeaveService, leave_fiscal_year_label
from app.domains.hr.service import SuccessFactorsLifecycleService
from app.identity.person_models import IdentityPerson
from app.identity.successfactors_auth import BasicAuth
from app.identity.successfactors_client import SuccessFactorsClient


def _sf_date(d: date) -> str:
    ms = int(datetime(d.year, d.month, d.day, tzinfo=UTC).timestamp() * 1000)
    return f"/Date({ms})/"


def _fake_client() -> SuccessFactorsClient:
    return SuccessFactorsClient(
        base_url="https://fake.sapsf.cn/odata/v2",
        auth=BasicAuth("apiuser@MYCO", "pw"),
    )


def _acct(user_id, atype, start, end, uuid, closed=False) -> dict:
    return {
        "externalCode": uuid,
        "userId": user_id,
        "accountType": atype,
        "startDate": _sf_date(start),
        "endDate": _sf_date(end),
        "accountClosed": closed,
    }


def _detail(uuid, ext, btype, amount, when: date, by="Lifecycle Admin") -> dict:
    return {
        "externalCode": ext,
        "TimeAccount_externalCode": uuid,
        "bookingType": btype,
        "bookingAmount": amount,
        "bookingUnit": "DAYS",
        "bookingDate": _sf_date(when),
        "createdBy": by,
        "createdDateTime": _sf_date(when),
        "comment": None,
    }


@pytest_asyncio.fixture
async def _person(db_session):
    pid = uuid4()
    person = IdentityPerson(
        id=pid,
        work_email=f"leave-{pid}@test.com",
        display_name="Leave Subject",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="P0001",
    )
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


def _patch_sf(monkeypatch, accounts, details, types=None, requests=None):
    async def fake_accounts(self, *, user_ids, account_types=None):
        return list(accounts)

    async def fake_details(self, *, time_account_uuids):
        return list(details)

    async def fake_types(self):
        return dict(types or {})

    async def fake_requests(self, *, user_ids):
        return list(requests or [])

    monkeypatch.setattr(SuccessFactorsClient, "list_time_accounts", fake_accounts)
    monkeypatch.setattr(
        SuccessFactorsClient, "list_time_account_details", fake_details,
    )
    monkeypatch.setattr(
        SuccessFactorsClient, "list_time_account_types", fake_types,
    )
    monkeypatch.setattr(
        SuccessFactorsClient, "list_employee_time", fake_requests,
    )


def test_leave_fiscal_year_label_uses_period_end_not_raw_start_year():
    """A 2025-issued leave account that expires in 2026 is HR's FY26."""
    assert leave_fiscal_year_label(
        period_end=date(2026, 9, 30), fiscal_year=2025,
    ) == "FY26"
    # A January joiner can have a 2026 raw start but still belongs to FY26.
    assert leave_fiscal_year_label(
        period_end=date(2026, 9, 30), fiscal_year=2026,
    ) == "FY26"


@pytest.mark.asyncio
async def test_mirrors_all_types_postings_and_derives_legacy(
    db_session, _person, monkeypatch,
):
    today = date(2026, 5, 14)
    accounts = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "legal-cur"),
        _acct("P0001", "CHN_AnnualLeave", date(2025, 1, 1), date(2025, 12, 31), "legal-prior", closed=True),
        _acct("P0001", "CHN_PL", date(2026, 1, 1), date(2026, 12, 31), "pl-cur"),
        _acct("P0001", "CHN_SickLeave", date(2026, 1, 1), date(2026, 12, 31), "sick-cur"),
    ]
    details = [
        # current legal: +10 accrual, then -1 and -4 taken → balance 5, used 5
        _detail("legal-cur", "d1", "ACCRUAL", "10", date(2026, 1, 1)),
        _detail("legal-cur", "d2", "EMPLOYEE_TIME", "-1", date(2026, 5, 8)),
        _detail("legal-cur", "d3", "EMPLOYEE_TIME", "-4", date(2026, 5, 20)),
        # prior legal: +10, -10 → balance 0
        _detail("legal-prior", "d4", "ACCRUAL", "10", date(2025, 1, 1)),
        _detail("legal-prior", "d5", "EMPLOYEE_TIME", "-10", date(2025, 6, 1)),
        # welfare current: +5.5
        _detail("pl-cur", "d6", "ACCRUAL", "5.5", date(2026, 1, 1)),
        # sick current: +6
        _detail("sick-cur", "d7", "ACCRUAL", "6", date(2026, 1, 1)),
    ]
    _patch_sf(monkeypatch, accounts, details, types={
        "CHN_AnnualLeave": "法定年假", "CHN_PL": "福利年假",
        "CHN_SickLeave": "带薪病假",
    })

    svc = SuccessFactorsLifecycleService(
        db_session, sf_client=_fake_client(), today=today,
    )
    summary = await svc.sync_leave_data(user_ids=["P0001"])
    assert summary["accounts_upserted"] == 4
    assert summary["postings_upserted"] == 7
    assert summary["persons_touched"] == 1

    read = HrLeaveService(db_session)
    rows = await read.list_accounts(str(_person.id))
    by_ext = {r.external_code: r for r in rows}

    # current legal: balance = 10-1-4 = 5, used = 5, entitled = 10, current
    legal = by_ext["legal-cur"]
    assert legal.account_type_label == "法定年假"
    assert legal.balance_days == Decimal("5.00")
    assert legal.used_days == Decimal("5.00")
    assert legal.entitled_days == Decimal("10.00")
    assert legal.is_current is True
    assert legal.fiscal_year == 2026

    # prior legal: not current, balance 0
    assert by_ext["legal-prior"].is_current is False
    assert by_ext["legal-prior"].balance_days == Decimal("0.00")

    # running balance_after on current legal, in chronological order
    postings = await read.list_postings(str(legal.id))  # newest first
    chrono = list(reversed(postings))
    assert [str(p.balance_after) for p in chrono] == ["10.00", "9.00", "5.00"]

    # legacy columns derived: carry_legal = prior-FY balance (0),
    # used_legal = current-FY used (5)
    await db_session.refresh(_person)
    assert _person.annual_leave_carry_legal_days == Decimal("0.0")
    assert _person.annual_leave_used_legal_days == Decimal("5.0")


@pytest.mark.asyncio
async def test_each_comp_account_is_its_own_row(db_session, _person, monkeypatch):
    """调休假 with several SF accounts → several Helix rows (user decision)."""
    today = date(2026, 5, 14)
    accounts = [
        _acct("P0001", "CHN_CompLeave", date(2026, 1, 1), date(2026, 12, 31), f"comp-{i}")
        for i in range(4)
    ]
    details = [
        _detail(f"comp-{i}", f"c{i}", "ACCRUAL", str(round(0.25 * (i + 1), 2)), date(2026, 1, 1))
        for i in range(4)
    ]
    _patch_sf(monkeypatch, accounts, details, types={"CHN_CompLeave": "调休假"})

    svc = SuccessFactorsLifecycleService(db_session, sf_client=_fake_client(), today=today)
    await svc.sync_leave_data(user_ids=["P0001"])

    rows = await HrLeaveService(db_session).list_accounts(str(_person.id))
    comp = [r for r in rows if r.account_type == "CHN_CompLeave"]
    assert len(comp) == 4, "each SF comp account must be its own row"


@pytest.mark.asyncio
async def test_idempotent_then_soft_deletes_vanished_account(
    db_session, _person, monkeypatch,
):
    today = date(2026, 5, 14)
    two = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "a1"),
        _acct("P0001", "CHN_PL", date(2026, 1, 1), date(2026, 12, 31), "a2"),
    ]
    det = [
        _detail("a1", "p1", "ACCRUAL", "5", date(2026, 1, 1)),
        _detail("a2", "p2", "ACCRUAL", "3", date(2026, 1, 1)),
    ]
    _patch_sf(monkeypatch, two, det)
    svc = SuccessFactorsLifecycleService(db_session, sf_client=_fake_client(), today=today)
    await svc.sync_leave_data(user_ids=["P0001"])

    # Re-run with the SAME data — no duplicate rows.
    await svc.sync_leave_data(user_ids=["P0001"])
    all_rows = (await db_session.execute(
        select(HrLeaveAccount).where(HrLeaveAccount.person_id == str(_person.id))
    )).scalars().all()
    assert len(all_rows) == 2, "idempotent upsert must not duplicate accounts"

    # Now SF stops returning a2 → it must be tombstoned, not deleted.
    _patch_sf(monkeypatch, [two[0]], [det[0]])
    summary = await svc.sync_leave_data(user_ids=["P0001"])
    assert summary["accounts_tombstoned"] == 1

    live = await HrLeaveService(db_session).list_accounts(str(_person.id))
    assert {r.external_code for r in live} == {"a1"}
    # the row still physically exists (soft delete)
    a2 = (await db_session.execute(
        select(HrLeaveAccount).where(HrLeaveAccount.external_code == "a2")
    )).scalar_one()
    assert a2.deleted_at is not None


@pytest.mark.asyncio
async def test_authoritative_empty_snapshot_tombstones_last_account_and_posting(
    db_session,
    _person,
    monkeypatch,
):
    """#2270: 1 → 0 is an authoritative snapshot, not a no-op."""
    instance_id = uuid4()
    _person.hris_instance_id = instance_id
    await db_session.commit()
    account = _acct(
        "P0001",
        "CHN_AnnualLeave",
        date(2026, 1, 1),
        date(2026, 12, 31),
        "last-account",
    )
    posting = _detail(
        "last-account",
        "last-posting",
        "ACCRUAL",
        "10",
        date(2026, 1, 1),
    )
    _patch_sf(monkeypatch, [account], [posting])
    svc = SuccessFactorsLifecycleService(
        db_session,
        sf_client=_fake_client(),
        today=date(2026, 5, 14),
        hris_instance_id=instance_id,
    )
    await svc.sync_leave_data(user_ids=["P0001"])

    async def empty_accounts(self, *, user_ids, account_types=None):
        return []

    async def unexpected_followup(*args, **kwargs):
        raise AssertionError("empty account snapshot must skip SF detail/type/request calls")

    monkeypatch.setattr(SuccessFactorsClient, "list_time_accounts", empty_accounts)
    monkeypatch.setattr(SuccessFactorsClient, "list_time_account_details", unexpected_followup)
    monkeypatch.setattr(SuccessFactorsClient, "list_time_account_types", unexpected_followup)
    monkeypatch.setattr(SuccessFactorsClient, "list_employee_time", unexpected_followup)

    summary = await svc.sync_leave_data(user_ids=["P0001"])

    assert summary["accounts_tombstoned"] == 1
    assert summary["persons_touched"] == 1
    assert await HrLeaveService(db_session).list_accounts(str(_person.id)) == []
    stored_account = await db_session.scalar(
        select(HrLeaveAccount).where(HrLeaveAccount.external_code == "last-account")
    )
    stored_posting = await db_session.scalar(
        select(HrLeavePosting).where(HrLeavePosting.external_code == "last-posting")
    )
    assert stored_account is not None and stored_account.deleted_at is not None
    assert stored_posting is not None and stored_posting.deleted_at is not None


@pytest.mark.asyncio
async def test_empty_snapshot_only_tombstones_current_instance_rows(
    db_session,
    _person,
    monkeypatch,
):
    """An explicit CN owner must not let CN's empty snapshot delete EU rows."""
    cn_id = uuid4()
    eu_id = uuid4()
    _person.hris_instance_id = cn_id
    cn_account = HrLeaveAccount(
        external_code="cn-account",
        person_id=str(_person.id),
        hris_person_id="P0001",
        hris_instance_id=cn_id,
        account_type="CHN_AnnualLeave",
        is_current=True,
    )
    eu_account = HrLeaveAccount(
        external_code="eu-account",
        person_id=str(_person.id),
        hris_person_id="P0001",
        hris_instance_id=eu_id,
        account_type="EU_AnnualLeave",
        is_current=True,
    )
    db_session.add_all([cn_account, eu_account])
    await db_session.flush()
    db_session.add_all(
        [
            HrLeavePosting(
                external_code="cn-posting",
                leave_account_id=cn_account.id,
                account_external_code="cn-account",
                person_id=str(_person.id),
                booking_type="ACCRUAL",
                booking_amount=Decimal("10"),
            ),
            HrLeavePosting(
                external_code="eu-posting",
                leave_account_id=eu_account.id,
                account_external_code="eu-account",
                person_id=str(_person.id),
                booking_type="ACCRUAL",
                booking_amount=Decimal("20"),
            ),
        ]
    )
    await db_session.commit()
    _patch_sf(monkeypatch, [], [])

    summary = await SuccessFactorsLifecycleService(
        db_session,
        sf_client=_fake_client(),
        today=date(2026, 5, 14),
        hris_instance_id=cn_id,
    ).sync_leave_data(user_ids=["P0001"])

    assert summary["accounts_tombstoned"] == 1
    await db_session.refresh(cn_account)
    await db_session.refresh(eu_account)
    assert cn_account.deleted_at is not None
    assert eu_account.deleted_at is None
    postings = {
        row.external_code: row
        for row in (await db_session.execute(select(HrLeavePosting))).scalars()
    }
    assert postings["cn-posting"].deleted_at is not None
    assert postings["eu-posting"].deleted_at is None


@pytest.mark.asyncio
async def test_legacy_unowned_person_is_not_tombstoned_by_wrong_instance(
    db_session,
    _person,
    monkeypatch,
):
    """A NULL-owned person may be visible to every fan-out service.

    A row already stamped by CN is authoritative evidence for CN only; an
    empty EU snapshot must leave it intact instead of cross-tenant deleting.
    """
    cn_id = uuid4()
    eu_id = uuid4()
    account = HrLeaveAccount(
        external_code="legacy-cn-account",
        person_id=str(_person.id),
        hris_person_id="P0001",
        hris_instance_id=cn_id,
        account_type="CHN_AnnualLeave",
        is_current=True,
    )
    db_session.add(account)
    await db_session.flush()
    db_session.add(
        HrLeavePosting(
            external_code="legacy-cn-posting",
            leave_account_id=account.id,
            account_external_code=account.external_code,
            person_id=str(_person.id),
            booking_type="ACCRUAL",
            booking_amount=Decimal("10"),
        )
    )
    await db_session.commit()
    _patch_sf(monkeypatch, [], [])

    summary = await SuccessFactorsLifecycleService(
        db_session,
        sf_client=_fake_client(),
        today=date(2026, 5, 14),
        hris_instance_id=eu_id,
    ).sync_leave_data(user_ids=["P0001"])

    assert summary["accounts_tombstoned"] == 0
    await db_session.refresh(account)
    assert account.deleted_at is None
    posting = await db_session.scalar(
        select(HrLeavePosting).where(HrLeavePosting.external_code == "legacy-cn-posting")
    )
    assert posting is not None and posting.deleted_at is None


@pytest.mark.asyncio
async def test_no_op_when_sf_returns_nothing(db_session, _person, monkeypatch):
    _patch_sf(monkeypatch, [], [])
    svc = SuccessFactorsLifecycleService(
        db_session, sf_client=_fake_client(), today=date(2026, 5, 14),
    )
    summary = await svc.sync_leave_data(user_ids=["P0001"])
    assert summary["accounts_upserted"] == 0
    rows = await HrLeaveService(db_session).list_accounts(str(_person.id))
    assert rows == []


@pytest.mark.asyncio
async def test_reason_stamped_from_employee_time_onto_taken_leave(
    db_session, _person, monkeypatch,
):
    """A taken-leave posting whose date falls inside a time-off request
    gets that request's reason/type/status; accrual lines stay clean."""
    today = date(2026, 5, 30)
    accounts = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "a1"),
    ]
    details = [
        _detail("a1", "p1", "ACCRUAL", "10", date(2026, 1, 1)),
        _detail("a1", "p2", "EMPLOYEE_TIME", "-1", date(2026, 5, 22)),
    ]
    requests = [{
        "userId": "P0001",
        "startDate": _sf_date(date(2026, 5, 22)),
        "endDate": _sf_date(date(2026, 5, 22)),
        "timeType": "CN_ANNUAL",
        "comment": "回家探亲",
        "approvalStatus": "APPROVED",
    }]
    _patch_sf(monkeypatch, accounts, details, requests=requests)
    svc = SuccessFactorsLifecycleService(db_session, sf_client=_fake_client(), today=today)
    await svc.sync_leave_data(user_ids=["P0001"])

    postings = await HrLeaveService(db_session).list_postings(
        str((await HrLeaveService(db_session).list_accounts(str(_person.id)))[0].id)
    )
    taken = [p for p in postings if p.booking_type == "EMPLOYEE_TIME"]
    accrual = [p for p in postings if p.booking_type == "ACCRUAL"]
    assert taken[0].reason == "回家探亲"
    assert taken[0].leave_type == "CN_ANNUAL"
    assert taken[0].request_status == "APPROVED"
    # accrual line must NOT pick up a reason
    assert accrual[0].reason is None


@pytest.mark.asyncio
async def test_cancellation_reversal_nets_out_of_used_days(
    db_session, _person, monkeypatch,
):
    """Cancelling approved leave posts a POSITIVE EMPLOYEE_TIME reversal of
    the same booking type. It must net the original negative booking out of
    used_days — not inflate both used AND entitled. Regression: the old
    aggregation did ``used += abs(amt)`` (so a +3 reversal added 3 to used)
    and ``entitled += amt`` for any positive row (so it also added 3 to
    entitled), corrupting the annual-leave calculator's used figure."""
    today = date(2026, 5, 14)
    accounts = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "legal-cur"),
    ]
    details = [
        _detail("legal-cur", "d1", "ACCRUAL", "10", date(2026, 1, 1)),
        _detail("legal-cur", "d2", "EMPLOYEE_TIME", "-3", date(2026, 4, 1)),  # leave taken
        _detail("legal-cur", "d3", "EMPLOYEE_TIME", "3", date(2026, 4, 10)),  # cancelled → reversal
    ]
    _patch_sf(monkeypatch, accounts, details, types={"CHN_AnnualLeave": "法定年假"})

    svc = SuccessFactorsLifecycleService(
        db_session, sf_client=_fake_client(), today=today,
    )
    await svc.sync_leave_data(user_ids=["P0001"])

    legal = {r.external_code: r for r in
             await HrLeaveService(db_session).list_accounts(str(_person.id))}["legal-cur"]
    assert legal.balance_days == Decimal("10.00")
    assert legal.used_days == Decimal("0.00"), "reversal must net the taken leave out of used"
    assert legal.entitled_days == Decimal("10.00"), "a reversal is not an entitlement accrual"

    # the legacy column the annual-leave calculator reads must reflect 0 used
    await db_session.refresh(_person)
    assert _person.annual_leave_used_legal_days == Decimal("0.0")


@pytest.mark.asyncio
async def test_sync_one_person_leave_now_pulls_just_that_person(
    db_session, _person, monkeypatch,
):
    """The per-person 'refresh from SF now' helper re-pulls a single
    person's Time-Off synchronously (the HR manual-sync button / the
    refresh_my_leave_balances MCP tool), so a just-edited leave request
    shows immediately instead of waiting for the 06:30 cron."""
    from app.domains.hr.service import sync_one_person_leave_now

    accounts = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "legal-cur"),
    ]
    details = [
        _detail("legal-cur", "d1", "ACCRUAL", "10", date(2026, 1, 1)),
        _detail("legal-cur", "d2", "EMPLOYEE_TIME", "-2", date(2026, 5, 12)),
    ]
    _patch_sf(monkeypatch, accounts, details, types={"CHN_AnnualLeave": "法定年假"})

    # fan_out reads HRIS integration rows (none in the test DB) — pin it to
    # a single fake-client service so the helper exercises the real loop.
    async def fake_fan_out(cls, db, *, today=None):
        return [SuccessFactorsLifecycleService(
            db, sf_client=_fake_client(), today=date(2026, 5, 14),
        )]
    monkeypatch.setattr(
        SuccessFactorsLifecycleService, "fan_out", classmethod(fake_fan_out),
    )

    summary = await sync_one_person_leave_now(db_session, _person)
    assert summary.get("skipped") is None
    assert summary["accounts_upserted"] == 1
    assert summary["persons_touched"] == 1

    rows = await HrLeaveService(db_session).list_accounts(str(_person.id))
    legal = {r.external_code: r for r in rows}["legal-cur"]
    assert legal.balance_days == Decimal("8.00")
    assert legal.used_days == Decimal("2.00")
    # legacy column the calculator reads is re-derived in the same pass
    await db_session.refresh(_person)
    assert _person.annual_leave_used_legal_days == Decimal("2.0")


@pytest.mark.asyncio
async def test_sync_one_person_leave_audit_failure_rolls_back_leave_rows(
    db_session, _person, monkeypatch,
):
    """The caller audit and refreshed Time-Off rows are one commit (#2467)."""
    from app.domains.hr.service import sync_one_person_leave_now

    accounts = [
        _acct(
            "P0001", "CHN_AnnualLeave", date(2026, 1, 1),
            date(2026, 12, 31), "audit-failure-account",
        ),
    ]
    details = [
        _detail(
            "audit-failure-account", "audit-failure-posting",
            "ACCRUAL", "10", date(2026, 1, 1),
        ),
    ]
    _patch_sf(monkeypatch, accounts, details)

    async def fake_fan_out(cls, db, *, today=None):
        return [SuccessFactorsLifecycleService(
            db, sf_client=_fake_client(), today=date(2026, 5, 14),
        )]

    async def fail_audit(_summary):
        raise RuntimeError("simulated audit failure")

    monkeypatch.setattr(
        SuccessFactorsLifecycleService, "fan_out", classmethod(fake_fan_out),
    )

    with pytest.raises(RuntimeError, match="simulated audit failure"):
        await sync_one_person_leave_now(
            db_session, _person, before_commit=fail_audit,
        )

    assert await db_session.scalar(
        select(func.count()).select_from(HrLeaveAccount).where(
            HrLeaveAccount.external_code == "audit-failure-account",
        )
    ) == 0
    assert await db_session.scalar(
        select(func.count()).select_from(HrLeavePosting).where(
            HrLeavePosting.external_code == "audit-failure-posting",
        )
    ) == 0


@pytest.mark.asyncio
async def test_sync_one_person_leave_now_skips_non_sf_person(
    db_session, monkeypatch,
):
    """A person not sourced from SuccessFactors has nothing to pull — the
    helper short-circuits with skipped and never touches the SF client."""
    from app.domains.hr.service import sync_one_person_leave_now

    pid = uuid4()
    person = IdentityPerson(
        id=pid,
        work_email=f"manual-{pid}@test.com",
        display_name="Manual Person",
        employment_status="active",
        source_system="manual",
        hris_person_id=f"MANUAL-{pid}",
    )
    db_session.add(person)
    await db_session.flush()

    def _boom(*a, **k):  # fan_out must NOT be reached
        raise AssertionError("fan_out should not run for a non-SF person")
    monkeypatch.setattr(SuccessFactorsLifecycleService, "fan_out", classmethod(_boom))

    summary = await sync_one_person_leave_now(db_session, person)
    assert summary["skipped"] == "not_sf_sourced"
    assert summary["accounts_upserted"] == 0


@pytest.mark.asyncio
async def test_persists_absences_from_employee_time(
    db_session, _person, monkeypatch,
):
    """The EmployeeTime REQUESTS themselves (not just the reason stamp on
    postings) land in hr_leave_absences — with span + status + quantity —
    and re-sync is idempotent + tombstones a request SF stops returning."""
    from app.domains.hr.leave_models import HrLeaveAbsence

    today = date(2026, 8, 6)
    accounts = [
        _acct("P0001", "CHN_AnnualLeave", date(2026, 1, 1), date(2026, 12, 31), "a1"),
    ]
    details = [_detail("a1", "p1", "ACCRUAL", "10", date(2026, 1, 1))]
    requests = [
        {
            "externalCode": "et-1",
            "userId": "P0001",
            "startDate": _sf_date(date(2026, 8, 12)),
            "endDate": _sf_date(date(2026, 8, 14)),
            "timeType": "CHN_ANNL",
            "quantityInDays": "3",
            "approvalStatus": "APPROVED",
            "comment": "私事",  # must NOT be stored on the absence row
        },
        {
            "externalCode": "et-2",
            "userId": "P0001",
            "startDate": _sf_date(date(2026, 8, 20)),
            "endDate": _sf_date(date(2026, 8, 20)),
            "timeType": "CHN_PSL",
            "quantityInDays": "0.5",
            "approvalStatus": "PENDING",
        },
    ]
    _patch_sf(monkeypatch, accounts, details, requests=requests)
    svc = SuccessFactorsLifecycleService(db_session, sf_client=_fake_client(), today=today)
    summary = await svc.sync_leave_data(user_ids=["P0001"])
    assert summary["absences_upserted"] == 2
    assert summary["absences_tombstoned"] == 0

    rows = (await db_session.execute(
        select(HrLeaveAbsence).where(
            HrLeaveAbsence.person_id == str(_person.id),
            HrLeaveAbsence.deleted_at.is_(None),
        ).order_by(HrLeaveAbsence.start_date)
    )).scalars().all()
    assert [r.external_code for r in rows] == ["et-1", "et-2"]
    assert rows[0].start_date == date(2026, 8, 12)
    assert rows[0].end_date == date(2026, 8, 14)
    assert rows[0].approval_status == "APPROVED"
    assert rows[0].quantity_days == Decimal("3.00")
    assert rows[0].time_type == "CHN_ANNL"
    assert rows[1].quantity_days == Decimal("0.50")
    assert rows[1].approval_status == "PENDING"
    # The free-text comment is never persisted on the absence row.
    assert not hasattr(rows[0], "comment")

    # Re-sync dropping et-2 → tombstoned, et-1 stays live (idempotent).
    _patch_sf(monkeypatch, accounts, details, requests=[requests[0]])
    svc2 = SuccessFactorsLifecycleService(db_session, sf_client=_fake_client(), today=today)
    summary2 = await svc2.sync_leave_data(user_ids=["P0001"])
    assert summary2["absences_upserted"] == 1
    assert summary2["absences_tombstoned"] == 1

    live = (await db_session.execute(
        select(HrLeaveAbsence.external_code).where(
            HrLeaveAbsence.person_id == str(_person.id),
            HrLeaveAbsence.deleted_at.is_(None),
        )
    )).scalars().all()
    assert set(live) == {"et-1"}