from datetime import date

import pytest
from sqlalchemy import select

from app.domains.hr.models import HrLifecycleEvent, HrSyncBatch, HrSyncStagedEvent
from app.identity.person_models import IdentityExternalIdentity, IdentityPerson


class _FakeLifecycleClient:
    async def list_hire_employments(self, start_date: date, end_date: date) -> list[dict]:
        return [
            {
                "personIdExternal": "H001",
                "userId": "H001",
                "startDate": "/Date(1772841600000)/",
                "endDate": None,
                "lastModifiedDateTime": "/Date(1772937600000+0000)/",
                "originalStartDate": "/Date(1772841600000)/",
            }
        ]

    async def list_termination_employments(self, start_date: date, end_date: date) -> list[dict]:
        return [
            {
                "personIdExternal": "L001",
                "userId": "L001",
                "startDate": "/Date(1704067200000)/",
                "endDate": "/Date(1773532800000)/",
                "lastModifiedDateTime": "/Date(1773446400000+0000)/",
                "originalStartDate": "/Date(1704067200000)/",
            }
        ]

    async def get_users(self, user_ids: list[str]) -> list[dict]:
        users = {
            "H001": {
                "userId": "H001",
                "displayName": "Hire User",
                "email": "hire@example.com",
                "empId": "EMP-H001",
                "status": "active",
                "hireDate": "/Date(1772841600000)/",
                "lastModifiedDateTime": "/Date(1772937600000+0000)/",
            },
            "L001": {
                "userId": "L001",
                "displayName": "Leave User",
                "email": "leave@example.com",
                "empId": "EMP-L001",
                "status": "inactive",
                "hireDate": "/Date(1704067200000)/",
                "lastModifiedDateTime": "/Date(1773446400000+0000)/",
            },
        }
        return [users[user_id] for user_id in user_ids]

    async def get_jobs_for_users(self, user_ids: list[str]) -> list[dict]:
        jobs = {
            "H001": [
                {
                    "userId": "H001",
                    "startDate": "/Date(1772841600000)/",
                    "endDate": "/Date(253402214400000)/",
                    "event": "1722",
                    "eventReason": "A01",
                    "emplStatus": "2187",
                    "lastModifiedDateTime": "/Date(1772937600000+0000)/",
                    "seqNumber": "1",
                }
            ],
            "L001": [
                {
                    "userId": "L001",
                    "startDate": "/Date(1704067200000)/",
                    "endDate": "/Date(1773532800000)/",
                    "event": "1727",
                    "eventReason": "D01",
                    "emplStatus": "2197",
                    "lastModifiedDateTime": "/Date(1773446400000+0000)/",
                    "seqNumber": "2",
                }
            ],
        }
        results: list[dict] = []
        for user_id in user_ids:
            results.extend(jobs.get(user_id, []))
        return results


# NOTE: tests for the deprecated `POST /api/v1/hr/lifecycle-events/import`
# endpoint and its underlying `SuccessFactorsLifecycleService.import_events`
# method were removed in the 11th pre-launch audit. The endpoint was tagged
# `deprecated=True` but still bypassed staged preview/confirm + the
# mass-leaver brake; the only path to apply HRIS events is now staged
# preview → human confirm via /api/v1/hr/sync/*.
#
# Removed:
#   - test_successfactors_lifecycle_imports_hire_and_termination_events
#   - test_successfactors_lifecycle_import_is_idempotent
#   - test_admin_can_import_successfactors_lifecycle_events_and_audits
#   - test_employee_cannot_import_successfactors_lifecycle_events


def test_is_placeholder_email_detects_dummy_stub_and_synthetic():
    from app.domains.hr.service import SuccessFactorsLifecycleService as S

    # Known SAP stubs
    assert S.is_placeholder_email("dummy@sap.com")
    assert S.is_placeholder_email("Noreply@SAP.com")
    # PingCAP's numbered dummy pattern
    assert S.is_placeholder_email("dummy153@pingcap.cn")
    assert S.is_placeholder_email("dummy007@gaiaworks.cn")
    # Plain ``dummy@...`` without digits — seen in prod as service-
    # account placeholder (``dummy@qq.cn``, ``dummy@sap123.com``)
    assert S.is_placeholder_email("dummy@qq.cn")
    assert S.is_placeholder_email("dummy@sap123.com")
    # Our own synthetic fallback
    assert S.is_placeholder_email("U12345@successfactors.local")
    # Empty / None
    assert S.is_placeholder_email(None)
    assert S.is_placeholder_email("")
    assert S.is_placeholder_email("   ")
    # Real corporate emails must NOT be flagged
    assert not S.is_placeholder_email("xiaopeng.zhu@pingcap.com")
    assert not S.is_placeholder_email("alice@example.com")
    # User whose first-name happens to start with "dummy" is a false
    # positive, but that's vanishingly rare compared to the stubs we
    # want to catch; documented here so it isn't treated as a bug.
    assert S.is_placeholder_email("dummyuser@pingcap.com")


@pytest.mark.asyncio
async def test_upsert_person_cross_tenant_shared_empid_does_not_crash(db_session, caplog):
    """Regression: two SF tenants can carry the same global empId
    (e.g., CN and EU both report P0876). When one tenant already owns
    ``employee_no='P0876'`` and a second-tenant confirm finds its own
    row by ``hris_person_id``, the empId write must NOT crash on
    ``uq_identity_persons_employee_no`` — leave the stale row's
    employee_no untouched and warn so admin can merge manually."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Row B — EU tenant, already confirmed P0876.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="164159077",
        employee_no="P0876",
        work_email="p0876-eu@successfactors.local",
        display_name="Zhu Xiaopeng",
        employment_status="active",
    ))
    # Row A — CN tenant, created by prior sync, no empId yet.
    row_a = IdentityPerson(
        source_system="successfactors",
        hris_person_id="P0876",
        employee_no=None,
        work_email="p0876-cn@successfactors.local",
        display_name="Zhu Xiaopeng",
        employment_status="active",
    )
    db_session.add(row_a)
    await db_session.flush()
    row_a_id = row_a.id

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 4, 21),
    )

    import logging
    with caplog.at_level(logging.WARNING, logger="app.domains.hr.service"):
        person, created = await service._upsert_person(
            user_id="P0876",
            user={"userId": "P0876", "empId": "P0876", "displayName": "Zhu Xiaopeng"},
            employment={"userId": "P0876", "startDate": "/Date(1704067200000)/"},
            event_type="onboard",
        )
        await db_session.flush()

    assert created is False
    assert person.id == row_a_id
    assert person.employee_no is None  # kept — can't conflict with Row B
    assert any("HRIS_EMPLOYEE_NUMBER_CONFLICT" in rec.message for rec in caplog.records)
    assert "P0876" not in caplog.text


# ── Welcome-email / provisioning gate ─────────────────────────────────


async def _stage_onboard_event(
    db_session,
    *,
    action: str,
    user_id: str,
    work_email: str,
) -> tuple[HrSyncBatch, HrSyncStagedEvent]:
    """Create a pending_review incremental batch carrying one onboard
    event with the requested ``action``. Returns (batch, staged_event)."""
    batch = HrSyncBatch(
        triggered_by="test",
        trigger_source="scheduled",
        sync_mode="incremental",
        start_date=date(2026, 5, 1),
        end_date=date(2026, 5, 2),
        status="pending_review",
    )
    db_session.add(batch)
    await db_session.flush()

    staged = HrSyncStagedEvent(
        batch_id=batch.id,
        event_type="onboard",
        user_id=user_id,
        action=action,
        status="pending",
        person_data={
            "user_id": user_id,
            "work_email": work_email,
            "display_name": "Test Person",
            "employee_no": user_id,
            "hire_date": "/Date(1746057600000)/",
        },
        employment_data={
            "userId": user_id,
            "personIdExternal": user_id,
            "startDate": "/Date(1746057600000)/",
            "endDate": None,
            "_jobs_snapshot": [],
        },
    )
    db_session.add(staged)
    await db_session.flush()
    await db_session.commit()
    return batch, staged


@pytest.mark.asyncio
async def test_confirm_batch_action_update_does_not_fire_welcome_or_provision(
    db_session, monkeypatch,
):
    """Regression (prod 2026-05-02): every hourly HRIS incremental sync
    re-fired welcome email + provisioning for active employees whose
    SF lastModifiedDateTime touched the lookback window but whose stored
    onboard lifecycle row's effective_date didn't match the SF startDate.
    Those events get classified ``action='update'`` (existing active
    person, no matching lifecycle row), and the original allow-list
    {create, reactivate, update} sent welcome mail to long-tenured staff
    (udi@pingcap.com, kevin@pingcap.com) on every tick. Only ``create``
    and ``reactivate`` represent genuinely new arrivals; ``update`` must
    be silent on these post-confirm hooks."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Existing active person — same SF userId as the staged event, so
    # the sync confirm path treats this as an "already in system" case.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="UDI001",
        employee_no="EMP-UDI001",
        work_email="udi@pingcap.com",
        display_name="Udi Test",
        employment_status="active",
    ))
    await db_session.flush()
    await db_session.commit()

    batch, _staged = await _stage_onboard_event(
        db_session, action="update", user_id="UDI001",
        work_email="udi@pingcap.com",
    )

    provision_calls: list[str] = []
    welcome_calls: list[str] = []

    async def _spy_provision(db, hris_user_id):
        provision_calls.append(hris_user_id)

    async def _spy_welcome(db, hris_user_id):
        welcome_calls.append(hris_user_id)

    async def _noop_apply_event(self, **kwargs):
        # Skip the heavy master-data lookups — the dispatch decision
        # under test happens AFTER ``_apply_event`` returns.
        return False, True, False, None  # added prior_trigger_source (P0-B)

    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.provision_for_new_hire",
        _spy_provision,
    )
    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.send_onboarding_it_info",
        _spy_welcome,
    )
    monkeypatch.setattr(
        "app.domains.hr.service.SuccessFactorsLifecycleService._apply_event",
        _noop_apply_event,
    )

    service = SuccessFactorsLifecycleService(db_session, today=date(2026, 5, 2))
    await service.confirm_batch(batch.id, confirmed_by="test")

    assert provision_calls == [], (
        "action=update must NOT trigger provisioning — these are "
        "already-active employees, not new hires"
    )
    assert welcome_calls == [], (
        "action=update must NOT trigger welcome email — these are "
        "already-active employees, not new hires"
    )


@pytest.mark.asyncio
async def test_confirm_batch_action_create_fires_welcome_and_provision(
    db_session, monkeypatch,
):
    """Positive lock-in: ``action='create'`` (genuinely new hire) must
    still fire both provisioning and welcome email after the trigger
    allow-list was tightened in the 2026-05-02 fix."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    batch, _staged = await _stage_onboard_event(
        db_session, action="create", user_id="NEW001",
        work_email="new001@pingcap.com",
    )

    provision_calls: list[str] = []
    welcome_calls: list[str] = []

    async def _spy_provision(db, hris_user_id):
        provision_calls.append(hris_user_id)

    async def _spy_welcome(db, hris_user_id):
        welcome_calls.append(hris_user_id)

    async def _noop_apply_event(self, **kwargs):
        return True, True, False, None  # added prior_trigger_source (P0-B)

    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.provision_for_new_hire",
        _spy_provision,
    )
    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.send_onboarding_it_info",
        _spy_welcome,
    )
    monkeypatch.setattr(
        "app.domains.hr.service.SuccessFactorsLifecycleService._apply_event",
        _noop_apply_event,
    )

    service = SuccessFactorsLifecycleService(db_session, today=date(2026, 5, 2))
    await service.confirm_batch(batch.id, confirmed_by="test")

    assert provision_calls == ["NEW001"]
    assert welcome_calls == ["NEW001"]


@pytest.mark.asyncio
async def test_confirm_batch_action_reactivate_fires_welcome_and_provision(
    db_session, monkeypatch,
):
    """Lock-in: a person being reactivated (was ``left`` /
    ``leaver_pending``, now back) IS a meaningful re-arrival and must
    re-fire provisioning + welcome email."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="REH001",
        employee_no="EMP-REH001",
        work_email="reh001@pingcap.com",
        display_name="Rehire Person",
        employment_status="left",
    ))
    await db_session.flush()
    await db_session.commit()

    batch, _staged = await _stage_onboard_event(
        db_session, action="reactivate", user_id="REH001",
        work_email="reh001@pingcap.com",
    )

    welcome_calls: list[str] = []
    provision_calls: list[str] = []

    async def _spy_provision(db, hris_user_id):
        provision_calls.append(hris_user_id)

    async def _spy_welcome(db, hris_user_id):
        welcome_calls.append(hris_user_id)

    async def _noop_apply_event(self, **kwargs):
        return True, True, False, None  # added prior_trigger_source (P0-B)

    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.provision_for_new_hire",
        _spy_provision,
    )
    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.send_onboarding_it_info",
        _spy_welcome,
    )
    monkeypatch.setattr(
        "app.domains.hr.service.SuccessFactorsLifecycleService._apply_event",
        _noop_apply_event,
    )

    service = SuccessFactorsLifecycleService(db_session, today=date(2026, 5, 2))
    await service.confirm_batch(batch.id, confirmed_by="test")

    assert provision_calls == ["REH001"]
    assert welcome_calls == ["REH001"]


@pytest.mark.asyncio
async def test_confirm_batch_action_skip_does_not_fire_welcome_or_provision(
    db_session, monkeypatch,
):
    """Lock-in: ``action='skip'`` (onboard event already recorded for
    this effective_date) must not trigger provisioning or welcome —
    the comment block in ``confirm_batch`` documents this contract."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="SKP001",
        employee_no="EMP-SKP001",
        work_email="skp001@pingcap.com",
        display_name="Skip Person",
        employment_status="active",
    ))
    await db_session.flush()
    await db_session.commit()

    batch, _staged = await _stage_onboard_event(
        db_session, action="skip", user_id="SKP001",
        work_email="skp001@pingcap.com",
    )

    provision_calls: list[str] = []
    welcome_calls: list[str] = []

    async def _spy_provision(db, hris_user_id):
        provision_calls.append(hris_user_id)

    async def _spy_welcome(db, hris_user_id):
        welcome_calls.append(hris_user_id)

    async def _noop_apply_event(self, **kwargs):
        return False, True, False, None  # added prior_trigger_source (P0-B)

    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.provision_for_new_hire",
        _spy_provision,
    )
    monkeypatch.setattr(
        "app.core.coordinators.hr_sync_coordinator.send_onboarding_it_info",
        _spy_welcome,
    )
    monkeypatch.setattr(
        "app.domains.hr.service.SuccessFactorsLifecycleService._apply_event",
        _noop_apply_event,
    )

    service = SuccessFactorsLifecycleService(db_session, today=date(2026, 5, 2))
    await service.confirm_batch(batch.id, confirmed_by="test")

    assert provision_calls == []
    assert welcome_calls == []


# ── Cross-instance ownership guard ────────────────────────────────────


@pytest.mark.asyncio
async def test_classify_skips_onboard_when_person_owned_by_other_instance(
    db_session,
):
    """Regression (prod 2026-05-02 — udi/kevin): the sf-eu-prod hourly
    incremental sync was ``_classify``-ing P01201 / P01436 as
    ``action='create'`` because the existing-person lookup is scoped to
    ``self.hris_instance_id`` (or NULL). The sf-cn-prod row was thus
    invisible to the EU sync, every tick produced a phantom 'new hire',
    and welcome emails fired hourly.

    The fix adds a cross-instance ownership guard: if a person with the
    same ``(source_system, hris_person_id)`` exists under a DIFFERENT
    instance, the running sync skips the event entirely (returns None
    from ``_stage_event``). The owning instance remains the single
    point of truth.
    """
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    cn_instance_id = uuid.uuid4()
    eu_instance_id = uuid.uuid4()

    # Person owned by CN instance.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="P01201",
        employee_no="EMP-P01201",
        work_email="udi@pingcap.com",
        display_name="Udi (CN-owned)",
        employment_status="active",
        hris_instance_id=cn_instance_id,
    ))
    await db_session.flush()
    await db_session.commit()

    # Run the EU-instance classify path. Without the guard, this would
    # return a staged event with action='create'.
    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=eu_instance_id, today=date(2026, 5, 2),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "P01201",
            "personIdExternal": "P01201",
            "startDate": "/Date(1746057600000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1746057600000+0000)/",
        },
        user={
            "userId": "P01201",
            "displayName": "Udi (EU view)",
            "email": "udi@pingcap.com",
            "empId": "P01201",
            "status": "active",
        },
        jobs=[],
    )
    assert staged is None, (
        "EU-instance sync must skip onboard events for users already "
        "owned by another HRIS instance (CN); otherwise welcome emails "
        "re-fire hourly for long-tenured staff"
    )


@pytest.mark.asyncio
async def test_classify_creates_when_no_cross_instance_owner_exists(
    db_session,
):
    """Positive lock-in: a genuinely new user (no row in any instance)
    still gets classified as ``create`` after the cross-instance guard
    is added. The guard must not over-fire and silence real new hires.
    """
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    instance_id = uuid.uuid4()
    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=instance_id, today=date(2026, 5, 2),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "BRAND_NEW",
            "personIdExternal": "BRAND_NEW",
            "startDate": "/Date(1746057600000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1746057600000+0000)/",
        },
        user={
            "userId": "BRAND_NEW",
            "displayName": "Genuine New Hire",
            "email": "new@pingcap.com",
            "empId": "BRAND_NEW",
            "status": "active",
        },
        jobs=[],
    )
    assert staged is not None
    assert staged.action == "create"


# ── HR test-record band (P99XX) ───────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize("event_type", ["onboard", "update", "offboard"])
async def test_stage_event_skips_p99xx_test_record(db_session, event_type):
    """HR reserves P9900-P9999 for their own SF test records
    (confirmed 2026-07-13). ``_stage_event`` must drop them for EVERY
    event type so they never enter / update / churn identity_persons —
    the single ingest chokepoint that starves every downstream path."""
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 13),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type=event_type,
        employment={
            "userId": "P9999",
            "personIdExternal": "P9999",
            "startDate": "/Date(1752364800000)/",
            "endDate": (
                "/Date(1752364800000)/" if event_type == "offboard" else None
            ),
            "lastModifiedDateTime": "/Date(1752364800000+0000)/",
        },
        user={
            "userId": "P9999",
            "displayName": "Lei Ting Lim (HR test)",
            "email": "nina.lim168@gmail.com",
            "empId": "P9999",
            "status": "active",
        },
        jobs=[],
    )
    assert staged is None


@pytest.mark.asyncio
async def test_stage_event_allows_real_five_digit_hire(db_session):
    """Lock-in: a real five-digit P-number (P10021) is NOT mistaken for a
    P99XX test record — the band must not over-fire on genuine hires."""
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 13),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "P10021",
            "personIdExternal": "P10021",
            "startDate": "/Date(1752364800000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1752364800000+0000)/",
        },
        user={
            "userId": "P10021",
            "displayName": "Genuine New Hire",
            "email": "new@pingcap.com",
            "empId": "P10021",
            "status": "active",
        },
        jobs=[],
    )
    assert staged is not None
    assert staged.action == "create"


@pytest.mark.asyncio
async def test_stage_event_preserves_first_and_last_name(db_session):
    """Preview→confirm must retain SF's structured name (P10020)."""
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 23),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(), event_type="onboard",
        employment={
            "userId": "P10020", "personIdExternal": "P10020",
            "startDate": "/Date(1786924800000)/", "endDate": None,
            "lastModifiedDateTime": "/Date(1784840065000+0000)/",
        },
        user={
            "userId": "P10020", "displayName": "Unknown",
            "firstName": "Otso", "lastName": "Juntunen",
            "email": "otso.juntunen@gmail.com", "empId": "P10020",
            "status": "active_external_suite",
        },
        jobs=[],
    )
    assert staged is not None
    assert staged.person_data["first_name"] == "Otso"
    assert staged.person_data["last_name"] == "Juntunen"


@pytest.mark.asyncio
async def test_stage_event_captures_custom07_as_email_name(db_session):
    """``User.custom07`` is the LEGACY fallback for the HR-assigned 邮箱名:
    when no ``PerPersonal.customString4`` is loaded for the person (empty
    ``_email_names_by_pid``), ``_stage_event`` still stages custom07 into
    ``person_data['email_name']`` so a blank customString4 doesn't regress
    the hire to the given.family pinyin.

    (The AUTHORITATIVE source is now ``PerPersonal.customString4`` — see
    ``test_stage_event_custom_string4_beats_custom07`` below.)
    """
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 20),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "P10019",
            "personIdExternal": "P10019",
            "startDate": "/Date(1753228800000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1753228800000+0000)/",
        },
        user={
            "userId": "P10019",
            "displayName": "John Chen",
            "email": "jianhua.chen@pingcap.com",
            "empId": "P10019",
            "status": "active",
            "custom07": "john.chen",
        },
        jobs=[],
    )
    assert staged is not None
    assert (staged.person_data or {}).get("email_name") == "john.chen", (
        "custom07 must be staged as email_name so HR's chosen mailbox wins "
        "over the given.family pinyin derivation."
    )


@pytest.mark.asyncio
async def test_stage_event_email_name_absent_when_no_custom07(db_session):
    """No ``custom07`` ⇒ ``email_name`` stays falsy, so reserve falls back to
    corporate-email / given.family derivation exactly as before (no regression
    for the common case where HR hasn't assigned a 邮箱名)."""
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 20),
    )
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "P10022",
            "personIdExternal": "P10022",
            "startDate": "/Date(1753228800000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1753228800000+0000)/",
        },
        user={
            "userId": "P10022",
            "displayName": "No EmailName Hire",
            "email": "someone@pingcap.com",
            "empId": "P10022",
            "status": "active",
        },
        jobs=[],
    )
    assert staged is not None
    assert not (staged.person_data or {}).get("email_name")


@pytest.mark.asyncio
async def test_stage_event_custom_string4_beats_custom07(db_session):
    """SF's AUTHORITATIVE 邮箱名 source is ``PerPersonal.customString4``
    (per SF's own outbound ``emailName <<-- customString4`` mapping). When the
    per-batch cache carries it, ``_stage_event`` stages customString4 as
    ``email_name`` EVEN IF the legacy ``User.custom07`` holds a different value.

    Pins the 2026-07-21 alignment: the pull path previously read only custom07
    (a different SF entity+field); if HR fills customString4 and NOT custom07,
    the sync path would have derived the wrong mailbox.
    """
    import uuid

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, hris_instance_id=uuid.uuid4(), today=date(2026, 7, 20),
    )
    # Simulate the per-batch load from PerPersonal.customString4.
    service._email_names_by_pid = {"P10019": "san.zhang"}
    staged = await service._stage_event(
        batch_id=uuid.uuid4(),
        event_type="onboard",
        employment={
            "userId": "P10019",
            "personIdExternal": "P10019",
            "startDate": "/Date(1753228800000)/",
            "endDate": None,
            "lastModifiedDateTime": "/Date(1753228800000+0000)/",
        },
        user={
            "userId": "P10019",
            "displayName": "John Chen",
            "email": "jianhua.chen@pingcap.com",
            "empId": "P10019",
            "status": "active",
            # Legacy field holds a STALE/different value — must lose.
            "custom07": "john.chen",
        },
        jobs=[],
    )
    assert staged is not None
    assert (staged.person_data or {}).get("email_name") == "san.zhang", (
        "PerPersonal.customString4 must win over the legacy User.custom07."
    )


# ── HR_SKIP_EMAIL_DOMAINS ─────────────────────────────────────────────


def test_email_domain_in_skip_list_matches_exact_and_subdomain():
    """``HR_SKIP_EMAIL_DOMAINS`` configures vendor / synthetic domains
    we never want to land in identity_persons. Subdomain match makes
    one entry cover the whole tree (``gaiaworks.cn`` covers
    ``partner.gaiaworks.cn`` without a separate entry)."""
    from app.domains.hr.service import SuccessFactorsLifecycleService as S

    # Defaults — gaiaworks.cn + successfactors.local
    assert S.email_domain_in_skip_list("sap@gaiaworks.cn")
    assert S.email_domain_in_skip_list("ECadmin@gaiaworks.cn")
    # subdomain
    assert S.email_domain_in_skip_list("claire.wu@partner.gaiaworks.cn")
    # synthetic fallback
    assert S.email_domain_in_skip_list("100113@successfactors.local")
    # case-insensitive
    assert S.email_domain_in_skip_list("Foo@Gaiaworks.CN")
    # real corporate email passes
    assert not S.email_domain_in_skip_list("alice@pingcap.com")
    assert not S.email_domain_in_skip_list("bob@pingcap-c.com")
    # not a tail match — ``notgaiaworks.cn`` should NOT match
    assert not S.email_domain_in_skip_list("foo@notgaiaworks.cn")
    # empty / malformed
    assert not S.email_domain_in_skip_list("")
    assert not S.email_domain_in_skip_list(None)
    assert not S.email_domain_in_skip_list("no-at-sign-here")


def test_should_skip_user_drops_gaiaworks_partner_at_preview():
    """SAP gaiaworks implementation-partner accounts (``ECadmin@gaiaworks.cn``,
    ``COMPadmin@partner.gaiaworks.cn``) must be dropped at preview time —
    they're real strings SF returns, not placeholders, so the existing
    ``is_placeholder_email`` doesn't catch them."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(db=None, today=date(2026, 5, 7))

    # ECadmin + sap@gaiaworks.cn — dropped
    assert service.should_skip_user(
        "ECadmin",
        {"userId": "ECadmin", "email": "sap@gaiaworks.cn"},
    )
    # partner subdomain — dropped via subdomain match
    assert service.should_skip_user(
        "COMPadmin",
        {"userId": "COMPadmin", "email": "claire.wu@partner.gaiaworks.cn"},
    )
    # Real PingCAP employee passes
    assert not service.should_skip_user(
        "P01505",
        {"userId": "P01505", "email": "selva.ganesh@pingcap-c.com"},
    )


def test_should_skip_user_keeps_new_hire_with_absent_email():
    """A brand-new hire with NO corporate email in SF yet (email=null or
    empty) must NOT be skipped — Helix mints the mailbox from their name.

    Regression for EU hire P10013 Robert Wood (2026-07-12): a real
    future hire (valid EmpJob + future hireDate) with ``email=null`` was
    silently dropped from onboarding because ``is_placeholder_email(None)``
    is True and the preview skip treated absent-email as a placeholder.
    Positive placeholder stubs (``dummy*`` / ``*@successfactors.local``)
    must STILL be skipped; genuine service accounts are caught by the
    user_id skip list."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(db=None, today=date(2026, 7, 12))

    # Absent / empty email on a real userId — kept (Helix will derive it).
    assert not service.should_skip_user(
        "P10013", {"userId": "P10013", "email": None},
    )
    assert not service.should_skip_user(
        "P10013", {"userId": "P10013", "email": "   "},
    )
    # Positive placeholder stubs are still dropped.
    assert service.should_skip_user(
        "U999", {"userId": "U999", "email": "dummy153@pingcap.cn"},
    )
    assert service.should_skip_user(
        "U998", {"userId": "U998", "email": "U998@successfactors.local"},
    )


@pytest.mark.asyncio
async def test_upsert_person_skips_when_resolved_email_collides_with_canonical(
    db_session, caplog,
):
    """SF-side duplicate userIds for the same human: the canonical row
    already owns the real email (``yun.zhu@pingcap.com``); the duplicate
    SF userId would otherwise create a ghost row whose
    ``_resolve_unique_work_email`` falls back to
    ``<userId>@successfactors.local``. Confirmed in prod 2026-05-07
    (P01115 canonical / 100113 duplicate, both for 朱赟 Claire Zhu)."""
    import logging

    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Canonical row already in helix.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="P01115",
        employee_no="P01115",
        work_email="yun.zhu@pingcap.com",
        display_name="朱赟 Claire Zhu",
        employment_status="active",
    ))
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    with caplog.at_level(logging.INFO, logger="app.domains.hr.service"):
        # SF returns the duplicate userId 100113 with the SAME real email.
        # ``_resolve_unique_work_email`` collides, falls back to
        # ``100113@successfactors.local``. The new gate must catch this
        # and refuse to create the ghost row.
        person, created = await service._upsert_person(
            user_id="100113",
            user={
                "userId": "100113",
                "empId": "100113",
                "displayName": "朱赟 Claire Zhu",
                "email": "yun.zhu@pingcap.com",
            },
            employment={"userId": "100113", "startDate": "/Date(1692057600000)/"},
            event_type="onboard",
        )

    assert person is None
    assert created is False
    assert any(
        "HRIS_CANONICAL_EMAIL_COLLISION_SKIPPED" in rec.message
        for rec in caplog.records
    )
    assert "100113" not in caplog.text
    assert "yun.zhu@pingcap.com" not in caplog.text

    # And no ghost row got persisted.
    ghost = await db_session.scalar(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "100113")
    )
    assert ghost is None


@pytest.mark.asyncio
async def test_upsert_person_creates_new_hire_with_no_sf_email(db_session):
    """Regression (#191 → 2026-07-01): a brand-new hire whose SF ``User``
    carries NO email at all must be CREATED with the synthetic placeholder,
    NOT refused.

    HRIS never issues corporate mailboxes — Helix mints them post-create via
    ``reserve_and_writeback_email`` and writes the address back to SF. #191's
    ``HR_SKIP_EMAIL_DOMAINS`` gate refused ANY ``.local`` fallback, so it ate
    real new hires whose ``.local`` was just "no mailbox yet" (not a
    duplicate collision) — P10007 (Victor Schmidt) / P10010 (Yuang Zhou) on
    2026-07-01 never landed and so never appeared on the onboarding page for
    an admin to action. The person MUST land."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    person, created = await service._upsert_person(
        user_id="P10007",
        user={
            "userId": "P10007",
            "empId": None,
            "displayName": "Victor Schmidt",
            "firstName": "Victor",
            "lastName": "Schmidt",
            "email": None,  # HRIS gives no corporate email — Helix mints it
        },
        employment={"userId": "P10007", "startDate": "/Date(1782864000000)/"},
        event_type="onboard",
    )

    assert person is not None, "email-less new hire must be created, not dropped"
    assert created is True
    # Placeholder is expected here — the coordinator derives & writes back the
    # real corporate address after the person exists.
    assert person.work_email == "p10007@successfactors.local"
    # Actually persisted → visible on the onboarding page.
    row = await db_session.scalar(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P10007")
    )
    assert row is not None


@pytest.mark.asyncio
async def test_apply_event_short_circuits_when_upsert_returns_none(
    db_session,
):
    """When ``_upsert_person`` returns None (filtered by
    ``HR_SKIP_EMAIL_DOMAINS``), the rest of ``_apply_event`` must NOT
    fire — no manager assignment, no dept membership, no external
    identity, no lifecycle event. Otherwise we'd write half-built rows
    that point at a missing person."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Canonical row owns the real email — duplicate must collide on it.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="P01134",
        employee_no="P01134",
        work_email="terry.purcell@pingcap.com",
        display_name="Terry Purcell",
        employment_status="active",
    ))
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    (
        event_created,
        person_upserted,
        employee_upserted,
        prior_trigger_source,
    ) = await service._apply_event(
        event_type="onboard",
        employment={
            "userId": "80100057",
            "startDate": "/Date(1743465600000)/",
            "_jobs_snapshot": [],
        },
        user={
            "userId": "80100057",
            "empId": "80100057",
            "displayName": "Terry Purcell",
            "email": "terry.purcell@pingcap.com",
        },
        jobs=[],
    )

    assert event_created is False
    assert person_upserted is False
    assert employee_upserted is False
    assert prior_trigger_source is None

    # No lifecycle event got written for the ghost row either.
    ev = await db_session.scalar(
        select(HrLifecycleEvent).where(
            HrLifecycleEvent.event_type == "onboard",
        ).where(
            HrLifecycleEvent.person_id.in_(
                select(IdentityPerson.id).where(
                    IdentityPerson.hris_person_id == "80100057",
                )
            )
        )
    )
    assert ev is None
    # And no external identity row either.
    ext = await db_session.scalar(
        select(IdentityExternalIdentity).where(
            IdentityExternalIdentity.provider_subject == "80100057",
        )
    )
    assert ext is None


# ── work_email parity with SF provider_email (no-downgrade + backfill) ─────


@pytest.mark.asyncio
async def test_resolve_unique_work_email_promotes_placeholder_to_real_on_update(
    db_session,
):
    """Backfill-on-every-sync: a row sitting on a stale
    ``<user_id>@successfactors.local`` placeholder MUST be upgraded to
    the SF-returned real email on the next sync, provided no other row
    owns it. This is the explicit fix for the prod 100113 / 80100057 case
    — the SF dup-userId condition that originally created the placeholder
    is gone, but without backfill the row would stay broken forever."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="100113",
        employee_no="100113",
        work_email="100113@successfactors.local",
        display_name="朱赟 Claire Zhu",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    resolved = await service._resolve_unique_work_email(
        user_id="100113",
        desired="yun.zhu@pingcap.com",
        current_person_id=person.id,
        current_email=person.work_email,
    )

    assert resolved == "yun.zhu@pingcap.com"


@pytest.mark.asyncio
async def test_resolve_unique_work_email_does_not_downgrade_real_to_placeholder_on_collision(
    db_session, caplog,
):
    """No-downgrade guard: when SF returns a real email that collides
    with another live row's ``work_email`` AND the existing person
    already has its OWN real (non-placeholder) ``work_email``, the
    resolver MUST keep the existing real value rather than fall back to
    ``<user_id>@successfactors.local``. Without this guard, a transient
    SF dup-userId state observed once would persistently flip a real
    work_email to the placeholder — exactly the failure mode that
    poisoned 100113 / 80100057 in prod."""
    import logging

    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Some other person already owns the email SF will return.
    db_session.add(IdentityPerson(
        source_system="successfactors",
        hris_person_id="OTHER",
        employee_no="OTHER",
        work_email="claire.zhu@pingcap.com",
        display_name="Some Other Claire",
        employment_status="active",
    ))
    # The person being updated currently has its own real (non-placeholder)
    # work_email — must NOT be downgraded.
    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="100113",
        employee_no="100113",
        work_email="yun.zhu@pingcap.com",
        display_name="朱赟 Claire Zhu",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    with caplog.at_level(logging.WARNING, logger="app.domains.hr.service"):
        resolved = await service._resolve_unique_work_email(
            user_id="100113",
            # SF returns a colliding real email this tick.
            desired="claire.zhu@pingcap.com",
            current_person_id=person.id,
            current_email=person.work_email,
        )

    # Existing real value preserved — no downgrade.
    assert resolved == "yun.zhu@pingcap.com"
    # And we left a paper trail so SRE can see the dup-userId state.
    assert any(
        "HRIS_EMAIL_COLLISION_EXISTING_PRESERVED" in rec.message
        for rec in caplog.records
    )
    assert "100113" not in caplog.text
    assert "yun.zhu@pingcap.com" not in caplog.text
    assert "claire.zhu@pingcap.com" not in caplog.text


@pytest.mark.asyncio
async def test_resolve_unique_work_email_logs_warning_when_falling_back_to_placeholder(
    db_session, caplog,
):
    """Every fallback to ``<user_id>@successfactors.local`` must emit
    a ``WARNING`` log so SRE / on-call sees a signal whenever a
    placeholder lands. The placeholder rotting silently in prod is what
    delayed detection of the 100113 / 80100057 case for weeks."""
    import logging

    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    with caplog.at_level(logging.WARNING, logger="app.domains.hr.service"):
        # SF returned no email at all and there's no current person to
        # protect (create-time path) — the function MUST fall back to
        # the placeholder AND log a warning.
        resolved = await service._resolve_unique_work_email(
            user_id="P99999",
            desired=None,
            current_person_id=None,
            current_email=None,
        )

    assert resolved == "p99999@successfactors.local"
    assert any(
        "HRIS_PLACEHOLDER_EMAIL_ASSIGNED" in rec.message
        for rec in caplog.records
    )
    assert "P99999" not in caplog.text
    assert "p99999@successfactors.local" not in caplog.text


# ── corporate-domain guard (personal email must never become work_email) ──


@pytest.mark.asyncio
async def test_resolve_unique_work_email_allows_personal_when_no_corporate_to_protect(
    db_session,
):
    """A fresh hire whose SF primary email is personal (个人邮箱) and who
    has NO existing corporate work_email keeps the historical behaviour:
    the address flows through (no person-creation is blocked). The Google
    provisioner's own corporate-domain guard
    (``_local_part_from_hris_email``) is what derives the real mailbox from
    the name rather than the personal local part — the sync layer only
    protects an ALREADY-corporate value (see the sticky test below)."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 6, 29),
    )

    resolved = await service._resolve_unique_work_email(
        user_id="P10008",
        desired="1030263320@qq.com",
        current_person_id=None,
        current_email=None,
    )

    # Not forced to a placeholder — that would trip the dup-userId
    # person-creation guard and block legitimate new hires.
    assert resolved == "1030263320@qq.com"


@pytest.mark.asyncio
async def test_resolve_unique_work_email_keeps_real_corporate_when_sf_sends_personal(
    db_session,
):
    """Once a real corporate mailbox has been minted and written to
    ``work_email``, a later sync where SF *still* returns the personal
    primary email must KEEP the corporate value (sticky) — not downgrade
    it. The corporate-domain guard makes the real address survive."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="P10008",
        employee_no="P10008",
        work_email="junyuan.huang@pingcap.com",   # real corporate mailbox
        display_name="黄浚源 Junyuan Huang",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 6, 29),
    )

    resolved = await service._resolve_unique_work_email(
        user_id="P10008",
        desired="1030263320@qq.com",          # SF primary email is still personal
        current_person_id=person.id,
        current_email=person.work_email,
    )

    assert resolved == "junyuan.huang@pingcap.com"


@pytest.mark.asyncio
async def test_upsert_person_backfills_placeholder_on_subsequent_sync(db_session):
    """End-to-end: existing row with placeholder ``work_email`` gets
    upgraded to the SF-returned real email after one ``_upsert_person``
    call, mimicking what happens on the next HRIS sync tick once the
    SF-side dup that originally caused the placeholder is cleaned up."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    # Pre-existing poisoned row — same shape as prod 100113.
    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="100113",
        employee_no="100113",
        work_email="100113@successfactors.local",
        display_name="朱赟 Claire Zhu",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 7),
    )

    refreshed, created = await service._upsert_person(
        user_id="100113",
        user={
            "userId": "100113",
            "empId": "100113",
            "displayName": "朱赟 Claire Zhu",
            "email": "yun.zhu@pingcap.com",
        },
        employment={"userId": "100113", "startDate": "/Date(1692057600000)/"},
        event_type="onboard",
    )
    await db_session.flush()

    assert created is False
    assert refreshed is not None
    assert refreshed.id == person.id
    assert refreshed.work_email == "yun.zhu@pingcap.com"


# ── _assign_manager: stale pointer clear (Round-20 audit, finding M1) ──


@pytest.mark.asyncio
async def test_assign_manager_clears_when_hris_drops_managerid(db_session):
    """Regression: an incremental sync event that omits ``managerId``
    (re-org → person now reports to no one, or HRIS data correction)
    must clear ``manager_person_id`` rather than preserving the prior
    pointer. Pre-fix the function ``return``ed on missing managerId,
    leaving the old manager linked forever — RBAC kept granting that
    manager view over the subordinate, and Leaver-Blocked dual-sign
    still accepted them as a valid secondary signer."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    manager = IdentityPerson(
        source_system="successfactors",
        hris_person_id="MGR-001",
        work_email="mgr-001@pingcap.com",
        display_name="Old Manager",
        employment_status="active",
    )
    db_session.add(manager)
    await db_session.flush()

    subordinate = IdentityPerson(
        source_system="successfactors",
        hris_person_id="EMP-001",
        work_email="emp-001@pingcap.com",
        display_name="Subordinate",
        employment_status="active",
        manager_person_id=manager.id,
    )
    db_session.add(subordinate)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 17),
    )

    # Event no longer reports a managerId — clear the stale pointer.
    await service._assign_manager(subordinate, job={})
    assert subordinate.manager_person_id is None

    # Re-attach to test the "lookup miss" branch.
    subordinate.manager_person_id = manager.id
    await db_session.flush()

    # Event references a managerId Helix has never seen → clear (not preserve).
    await service._assign_manager(
        subordinate, job={"managerId": "MGR-NEVER-SYNCED"},
    )
    assert subordinate.manager_person_id is None


@pytest.mark.asyncio
async def test_assign_manager_sets_when_hris_reports_known_manager(db_session):
    """Positive lock-in: when HRIS reports a managerId that resolves to
    an existing IdentityPerson, the pointer wires up correctly."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    manager = IdentityPerson(
        source_system="successfactors",
        hris_person_id="MGR-002",
        work_email="mgr-002@pingcap.com",
        display_name="Manager",
        employment_status="active",
    )
    db_session.add(manager)
    await db_session.flush()

    subordinate = IdentityPerson(
        source_system="successfactors",
        hris_person_id="EMP-002",
        work_email="emp-002@pingcap.com",
        display_name="Subordinate",
        employment_status="active",
    )
    db_session.add(subordinate)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 5, 17),
    )

    await service._assign_manager(subordinate, job={"managerId": "MGR-002"})
    assert subordinate.manager_person_id == manager.id


@pytest.mark.asyncio
async def test_lifecycle_event_natural_key_is_unique(db_session):
    """Regression: (person_id, event_type, effective_date) must be UNIQUE so a
    concurrent SuccessFactors webhook redelivery can't double-insert the same
    lifecycle event (which fired onboarding/offboarding twice). The service's
    begin_nested/IntegrityError guard relies on this constraint."""
    from sqlalchemy.exc import IntegrityError

    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="P-UNIQ-1",
        work_email="uniq1@successfactors.local",
        display_name="Uniq One",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    eff = date(2026, 6, 1)
    db_session.add(HrLifecycleEvent(
        person_id=person.id, event_type="offboard", effective_date=eff,
        detail={}, trigger_source="webhook",
    ))
    await db_session.flush()

    db_session.add(HrLifecycleEvent(
        person_id=person.id, event_type="offboard", effective_date=eff,
        detail={}, trigger_source="webhook",
    ))
    with pytest.raises(IntegrityError):
        await db_session.flush()


@pytest.mark.asyncio
async def test_apply_event_populates_feishu_attrs_gender_office_employee_type(
    db_session,
):
    """Regression (migration 344): ``_apply_event`` must land the three
    Feishu-bound HRIS attributes on identity_persons:
      * gender             ← PerPersonal.gender master cache
      * office_code         ← EmpJob.customString7 (job snapshot)
      * employee_type_code  ← EmpJob.customString25 (job snapshot)
    These feed the HRIS → Helix → Feishu sync (replacing 北森→Authing→飞书)."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 6, 26),
    )
    # Pre-seed master-data caches so _ensure_master_data_loaded short-circuits
    # (it early-returns when _companies is not None). Gender shares the
    # PerPersonal pull keyed by personIdExternal.
    service._companies = {}
    service._locations = {}
    service._departments = {}
    service._positions = {}
    service._personal_mobiles = {}
    service._personal_emails = {}
    service._personal_birthdays = {}
    service._personal_genders = {"FZ-PID": "M"}

    await service._apply_event(
        event_type="onboard",
        employment={
            "userId": "FZ001",
            "personIdExternal": "FZ-PID",
            "startDate": "/Date(1772841600000)/",
            "_jobs_snapshot": [
                {
                    "userId": "FZ001",
                    "company": "DP00020",
                    "location": "CHN0075",
                    "costCenter": "C0001",
                    # The two new EmpJob picklist codes (migration 344).
                    "customString7": "2659",   # sh-office → "SH Office"
                    "customString25": "1070",  # 正式 → Feishu enum
                }
            ],
        },
        user={
            "userId": "FZ001",
            "empId": "FZ001",
            "displayName": "Feishu Attr User",
            "email": "feishu.attr@pingcap.com",
            "status": "t",
        },
        jobs=[],
    )
    await db_session.flush()

    person = await db_session.scalar(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "FZ001")
    )
    assert person is not None
    assert person.gender == "M"
    assert person.office_code == "2659"
    assert person.employee_type_code == "1070"


@pytest.mark.asyncio
async def test_apply_event_populates_international_assignment_fields(db_session):
    """Migration 392: ``_apply_event`` lands the 外派 (international
    assignment) EmpJob custom fields on identity_persons — home/host country,
    status code, start/end dates — and the derived
    ``is_on_international_assignment`` (host set AND host != home) is True.
    Field IDs confirmed via live SF $metadata labels 2026-07-07."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 6, 26),
    )
    service._companies = {}
    service._locations = {}
    service._departments = {}
    service._positions = {}
    service._personal_mobiles = {}
    service._personal_emails = {}
    service._personal_birthdays = {}
    service._personal_genders = {}

    await service._apply_event(
        event_type="onboard",
        employment={
            "userId": "GA001",
            "personIdExternal": "GA-PID",
            "startDate": "/Date(1772841600000)/",
            "_jobs_snapshot": [
                {
                    "userId": "GA001",
                    "company": "DP00005",
                    "customString44": "CHN",           # 外派前国家 (home)
                    "customString38": "SGP",           # 外派后国家 (host)
                    "customString39": "53780",         # 外派状态 (raw code)
                    "customDate2": "/Date(1776470400000)/",  # 外派开始日期
                    "customDate4": "/Date(1808006400000)/",  # 外派结束日期
                }
            ],
        },
        user={
            "userId": "GA001",
            "empId": "GA001",
            "displayName": "Assignee One",
            "email": "assignee.one@pingcap.com",
            "status": "t",
        },
        jobs=[],
    )
    await db_session.flush()

    person = await db_session.scalar(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "GA001")
    )
    assert person is not None
    assert person.assignment_home_country == "CHN"
    assert person.assignment_host_country == "SGP"
    assert person.assignment_status_code == "53780"
    # Dates parsed from SF /Date(ms)/ into real dates, start before end.
    assert isinstance(person.assignment_start_date, date)
    assert isinstance(person.assignment_end_date, date)
    assert person.assignment_start_date < person.assignment_end_date
    # Derived signal: host set AND host != home → on assignment.
    assert person.is_on_international_assignment is True


@pytest.mark.parametrize(
    "raw, expected",
    [
        ("english", "en_US"),
        ("English", "en_US"),
        ("en_US", "en_US"),
        ("en-GB", "en_US"),
        ("chinese", "zh_CN"),
        ("中文", "zh_CN"),
        ("zh_CN", "zh_CN"),
        ("japanese", "ja_JP"),
        ("ja_JP", "ja_JP"),
        ("日本語", "ja_JP"),
        ("", None),
        (None, None),
        ("klingon", None),
    ],
)
def test_normalize_preferred_language(raw, expected):
    """SF ``defaultLocale`` is free-text, not BCP-47 — canonicalise it to
    the closed set the PATCH endpoint enforces, or None when unknown."""
    from app.domains.hr.service import _normalize_preferred_language

    assert _normalize_preferred_language(raw) == expected


@pytest.mark.asyncio
async def test_preferred_language_seeded_once_then_user_choice_sticky(db_session):
    """Regression (2026-06-29): ``preferred_language`` is a USER-OWNED UI
    preference. SF ``defaultLocale`` may SEED a brand-new person, but a
    later sync must NOT overwrite a value already on the row — else every
    sync/release re-stamps SF's locale and snaps the user back, undoing
    the in-app language switch.

    Also covers normalisation: SF returns free-text (``chinese``), which
    must be canonicalised to ``zh_CN`` (not written raw).
    """
    from app.domains.hr.service import SuccessFactorsLifecycleService

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 6, 26),
    )
    # Pre-seed master-data caches so _ensure_master_data_loaded short-circuits.
    service._companies = {}
    service._locations = {}
    service._departments = {}
    service._positions = {}
    service._personal_mobiles = {}
    service._personal_emails = {}
    service._personal_birthdays = {}
    service._personal_genders = {}

    employment = {
        "userId": "PL001",
        "personIdExternal": "PL-PID",
        "startDate": "/Date(1772841600000)/",
        "_jobs_snapshot": [{"userId": "PL001", "company": "DP00020"}],
    }
    user = {
        "userId": "PL001",
        "empId": "PL001",
        "displayName": "Lang User",
        "email": "lang.user@pingcap.com",
        "status": "t",
        "defaultLocale": "chinese",  # SF free-text
    }

    # 1) First upsert with SF free-text defaultLocale="chinese" → seeds the
    #    empty field, NORMALISED to the canonical "zh_CN" (not raw).
    person, _ = await service._upsert_person(
        "PL001", user, employment, "onboard",
    )
    await db_session.flush()
    assert person is not None
    assert person.preferred_language == "zh_CN"  # seeded + normalised

    # 2) User switches to en_US in-app (mirrors PATCH /auth/me/profile).
    person.preferred_language = "en_US"
    await db_session.flush()

    # 3) A later sync where SF still reports defaultLocale="chinese" must
    #    NOT clobber the user's en_US choice — this is the bug.
    person2, _ = await service._upsert_person(
        "PL001", user, employment, "onboard",
    )
    await db_session.flush()
    assert person2 is not None
    assert person2.preferred_language == "en_US"  # user choice preserved


@pytest.mark.asyncio
async def test_resolve_unique_work_email_keeps_google_backed_corporate_over_sf_corporate(
    db_session,
):
    """corporate-vs-corporate conflict (the Julia P01572 / roger.zhou03
    lesson, 2026-07-12): SF carries ANOTHER hire's cloned corporate email,
    while the person's real mailbox is live in Google and linked as their
    active login identity. The Google-verified address must win — every
    sync tick was re-poisoning the corrected work_email and breaking her
    SSO match. Google is authoritative for corporate mailboxes."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="P01572",
        employee_no="P01572",
        work_email="julia@pingcap.com",       # real, Google-verified mailbox
        display_name="杨懿梅 Julia",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    db_session.add(IdentityExternalIdentity(
        person_id=person.id,
        provider="google",
        provider_subject="julia-google-uid",
        provider_email="julia@pingcap.com",
        is_login_identity=True,
        identity_status="active",
    ))
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 7, 12),
    )

    resolved = await service._resolve_unique_work_email(
        user_id="P01572",
        desired="roger.zhou03@pingcap.com",   # SF data-entry error (corporate!)
        current_person_id=person.id,
        current_email=person.work_email,
    )

    assert resolved == "julia@pingcap.com"


@pytest.mark.asyncio
async def test_resolve_unique_work_email_corporate_update_flows_when_not_google_backed(
    db_session,
):
    """The Google-backed protection must NOT freeze every corporate
    address: with no active Google login identity matching the current
    work_email (mailbox not minted yet, or already renamed Google-side),
    an SF corporate update flows through as before."""
    from app.domains.hr.service import SuccessFactorsLifecycleService

    person = IdentityPerson(
        source_system="successfactors",
        hris_person_id="P99001",
        employee_no="P99001",
        work_email="old.name@pingcap.com",    # corporate but NOT google-backed
        display_name="Old Name",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=_FakeLifecycleClient(), today=date(2026, 7, 12),
    )

    resolved = await service._resolve_unique_work_email(
        user_id="P99001",
        desired="new.name@pingcap.com",
        current_person_id=person.id,
        current_email=person.work_email,
    )

    assert resolved == "new.name@pingcap.com"