"""Tests for the daily / local-working-hours inventory reminder.

Replaces the prior one-shot semantics. Verifies:

* working_hours_only gate skips delivery outside Mon–Fri 10:00–17:59
  in the recipient's country's IANA zone,
* per-recipient daily dedup: a person reminded inside their local day
  is skipped on a later call inside the same local day,
* per-recipient daily dedup releases the next local day,
* ``force=True`` bypasses both gates,
* ``cycle.reminder_sent_at`` is set on the first successful send and
  stable afterwards.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.working_hours import local_start_of_day, resolve_timezone
from app.domains.asset.inventory_service import AssetInventoryService
from app.domains.asset.models import (
    AssetInventoryCycle,
    AssetInventoryItem,
    AssetItem,
)
from app.identity.person_models import IdentityPerson


async def _make_person(
    db: AsyncSession, email: str, *, country: str | None = "CN",
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        country=country,
    )
    db.add(p)
    await db.commit()
    return p


async def _make_active_cycle_with_item(
    db: AsyncSession,
    holder: IdentityPerson,
    *,
    started_at: datetime,
) -> tuple[AssetInventoryCycle, AssetInventoryItem]:
    asset = AssetItem(
        asset_tag=f"REM-{uuid.uuid4().hex[:6]}",
        category="laptop",
        status="claimed",
        location="北京",
        assigned_to=holder.id,
    )
    db.add(asset)
    await db.flush()
    cycle = AssetInventoryCycle(
        id=uuid.uuid4(),
        name="reminder test",
        status="active",
        started_at=started_at,
        total_items=1,
        pending_count=1,
    )
    db.add(cycle)
    await db.flush()
    item = AssetInventoryItem(
        id=uuid.uuid4(),
        cycle_id=cycle.id,
        asset_id=asset.id,
        holder_kind="assignee",
        holder_person_id=holder.id,
        asset_tag_snapshot=asset.asset_tag,
        asset_category_snapshot=asset.category,
        asset_status_snapshot=asset.status,
        status="pending",
    )
    db.add(item)
    await db.commit()
    return cycle, item


@pytest.fixture
def stub_dispatch(monkeypatch):
    """No-op NotificationService.deliver — the tests assert on
    service return values + per-item ``reminded_at`` mutation, not on
    Feishu side effects. The reminder path enqueues a real outbox row
    (committed pre-send since the two-phase split), then calls
    ``deliver``; stubbing ``deliver`` keeps external channels out."""
    async def _noop(self, outbox, env, *, trace_id="", actor_user_id=None):
        return None

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.deliver",
        _noop,
    )


@pytest.mark.asyncio
async def test_force_bypasses_working_hours_and_daily_dedup(
    db_session: AsyncSession, stub_dispatch,
):
    holder = await _make_person(db_session, "force@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)

    sent = await svc.send_cycle_reminder(cycle.id, force=True)
    assert sent.recipients_notified == 1
    await db_session.refresh(item)
    assert item.reminded_at is not None

    # A second force call same-day still sends (force trumps the
    # per-day dedup) — once the up-front claim's concurrency grace has
    # passed. Backdate the stamp beyond _FORCE_RECLAIM_GRACE (still the
    # same local day for the daily-dedup's purposes) to model "admin
    # re-nudges half an hour later" rather than an overlapping call.
    item.reminded_at = datetime.now(UTC) - timedelta(minutes=30)
    await db_session.commit()
    sent2 = await svc.send_cycle_reminder(cycle.id, force=True)
    assert sent2.recipients_notified == 1


@pytest.mark.asyncio
async def test_working_hours_gate_skips_outside_business_hours(
    db_session: AsyncSession, stub_dispatch, monkeypatch,
):
    holder = await _make_person(db_session, "wh@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)

    # Pin "now" inside the service module to a Saturday 10:00 SHA.
    # 2026-05-23 is a Saturday; 02:00 UTC = 10:00 Asia/Shanghai.
    import app.domains.asset.inventory_service as svc_mod
    pinned = datetime(2026, 5, 23, 2, 0, tzinfo=UTC)

    class _FrozenDatetime(datetime):
        @classmethod
        def now(cls, tz=None):  # type: ignore[override]
            return pinned.astimezone(tz) if tz else pinned.replace(tzinfo=None)
    monkeypatch.setattr(svc_mod, "datetime", _FrozenDatetime)

    sent = await svc.send_cycle_reminder(
        cycle.id, working_hours_only=True,
    )
    assert sent.recipients_notified == 0
    await db_session.refresh(item)
    assert item.reminded_at is None


@pytest.mark.asyncio
async def test_working_hours_gate_delivers_during_business_hours(
    db_session: AsyncSession, stub_dispatch, monkeypatch,
):
    holder = await _make_person(db_session, "wh-ok@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)

    # 2026-05-21 Thursday 10:00 Asia/Shanghai = 02:00 UTC.
    import app.domains.asset.inventory_service as svc_mod
    pinned = datetime(2026, 5, 21, 2, 0, tzinfo=UTC)

    class _FrozenDatetime(datetime):
        @classmethod
        def now(cls, tz=None):  # type: ignore[override]
            return pinned.astimezone(tz) if tz else pinned.replace(tzinfo=None)
    monkeypatch.setattr(svc_mod, "datetime", _FrozenDatetime)

    sent = await svc.send_cycle_reminder(
        cycle.id, working_hours_only=True,
    )
    assert sent.recipients_notified == 1
    await db_session.refresh(item)
    assert item.reminded_at is not None


@pytest.mark.asyncio
async def test_daily_dedup_blocks_same_local_day(
    db_session: AsyncSession, stub_dispatch,
):
    holder = await _make_person(db_session, "dedup@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    # Anchor the prior reminder to *this morning, holder-local* rather
    # than "now − 5 min": a small delta straddles the local-midnight
    # boundary when CI runs in the first minutes after CN midnight
    # (16:00–16:05 UTC), pushing "5 min ago" onto the previous local
    # day so dedup wouldn't block. local_start_of_day + 6h is always
    # inside the current local day, so the dedup intent holds at any
    # wall-clock time.
    tz = resolve_timezone(holder)
    item.reminded_at = local_start_of_day(
        datetime.now(UTC), tz,
    ) + timedelta(hours=6)
    await db_session.commit()
    svc = AssetInventoryService(db_session)

    sent = await svc.send_cycle_reminder(cycle.id)
    assert sent.recipients_notified == 0


@pytest.mark.asyncio
async def test_daily_dedup_handles_naive_reminded_at_from_db(
    db_session: AsyncSession, stub_dispatch,
):
    """Regression for PR #440 → fix/inventory-reminder-naive-aware-cmp:
    MySQL/TiDB ``DATETIME`` columns drop tzinfo, so a freshly-loaded
    ``AssetInventoryItem.reminded_at`` is naive even though the
    handler wrote it from ``datetime.now(UTC)``. The original PR #440
    code compared this naive value against an aware ``local_start_of_day``
    return → ``TypeError: can't compare offset-naive and offset-aware
    datetimes`` on every active cycle with at least one previously
    reminded item. The local pytest fixture had silently passed
    because SQLAlchemy's identity map cached the aware datetime
    in-memory; a fresh worker process load-from-DB tripped it.

    Reproduce the worker boundary by ``expire_all()`` between the
    setup write and the service call, forcing SQLAlchemy to re-load
    ``reminded_at`` from the DB (where MySQL has stripped tzinfo).
    """
    holder = await _make_person(db_session, "naive-cmp@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=5),
    )
    # Last reminder 2 days ago — far outside any local day window —
    # but the value we care about is *naive-from-DB*, not the cutoff.
    item.reminded_at = datetime.now(UTC) - timedelta(days=2)
    await db_session.commit()
    # Capture the id BEFORE evicting so the test call doesn't trigger
    # lazy reload from a sync attribute access path.
    cycle_id = cycle.id
    # Evict cycle + item so the service's fresh ``select(...)`` returns
    # rows whose datetime columns come back naive (MySQL strips
    # tzinfo). Mirrors the worker-process state on prod.
    db_session.expunge(cycle)
    db_session.expunge(item)
    svc = AssetInventoryService(db_session)

    # Pre-fix: this raises TypeError inside send_cycle_reminder.
    # Post-fix: cleanly delivers (2-day-old reminder is stale).
    sent = await svc.send_cycle_reminder(cycle_id)
    assert sent.recipients_notified == 1


@pytest.mark.asyncio
async def test_daily_dedup_allows_next_local_day(
    db_session: AsyncSession, stub_dispatch,
):
    holder = await _make_person(db_session, "next-day@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=5),
    )
    # Last reminder 2 days ago — last local day's reminder is stale.
    item.reminded_at = datetime.now(UTC) - timedelta(days=2)
    await db_session.commit()
    svc = AssetInventoryService(db_session)

    sent = await svc.send_cycle_reminder(cycle.id)
    assert sent.recipients_notified == 1


@pytest.mark.asyncio
async def test_null_display_name_falls_back_to_email_localpart(
    db_session: AsyncSession,
):
    """Reminders to a Person row with NULL ``display_name`` should not
    render "你好 ,". Verify the envelope greeting falls back to the
    work_email local-part so the card stays readable."""
    from app.domains.asset.inventory_service import AssetInventoryService
    holder = IdentityPerson(
        id=uuid.uuid4(),
        work_email="anonymous.user@test.com",
        display_name=None,
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        country="CN",
    )
    db_session.add(holder)
    await db_session.commit()
    cycle, _item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)
    envelope = svc._build_envelope(
        holder, cycle, personal_n=1, office_n=0,
        template_key="asset_inventory_reminder",
        portal_url="https://example.com",
    )
    # No bare comma after a blank greeting.
    assert "你好 ," not in envelope.body_text
    assert "anonymous.user" in envelope.body_text
    assert envelope.context["person_name"] == "anonymous.user"


@pytest.mark.asyncio
async def test_service_rejects_oversized_evidence_photos(
    db_session: AsyncSession, stub_dispatch,
):
    """Service-layer defence-in-depth: even if a caller bypasses the
    Pydantic schema, oversized / malformed photo keys must be
    rejected at the service boundary."""
    from app.core.exceptions import ValidationError
    from app.domains.asset.inventory_service import AssetInventoryService
    holder = await _make_person(db_session, "ddx@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)

    # Too many keys.
    with pytest.raises(ValidationError):
        await svc.respond_to_item(
            item.id,
            actor_person_id=holder.id,
            is_admin=False,
            response_type="not_mine",
            evidence_photos=[f"k{i}.png" for i in range(11)],
        )

    # Empty-string key.
    with pytest.raises(ValidationError):
        await svc.respond_to_item(
            item.id,
            actor_person_id=holder.id,
            is_admin=False,
            response_type="not_mine",
            evidence_photos=["valid.png", "   "],
        )

    # Oversize per-key.
    with pytest.raises(ValidationError):
        await svc.respond_to_item(
            item.id,
            actor_person_id=holder.id,
            is_admin=False,
            response_type="not_mine",
            evidence_photos=["x" * 600],
        )


@pytest.mark.asyncio
async def test_service_rejects_duplicate_evidence_photo_keys(
    db_session: AsyncSession, stub_dispatch,
) -> None:
    from app.core.exceptions import ValidationError
    holder = await _make_person(db_session, "duplicate@test.com", country="CN")
    _cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    service = AssetInventoryService(db_session)
    with pytest.raises(ValidationError, match="duplicate object keys"):
        await service.respond_to_item(
            item.id,
            actor_person_id=holder.id,
            is_admin=False,
            response_type="not_mine",
            response_notes="duplicate evidence",
            evidence_photos=["same.png", " same.png "],
        )


@pytest.mark.asyncio
async def test_reminder_sent_at_set_on_first_send_only(
    db_session: AsyncSession, stub_dispatch,
):
    holder = await _make_person(db_session, "rsa@test.com", country="CN")
    cycle, item = await _make_active_cycle_with_item(
        db_session, holder,
        started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)

    sent = await svc.send_cycle_reminder(cycle.id, force=True)
    assert sent.recipients_notified == 1
    await db_session.refresh(cycle)
    first = cycle.reminder_sent_at
    assert first is not None

    # Second force call should NOT overwrite reminder_sent_at —
    # that field captures the first reminder (audit / UI). Backdate
    # the claim stamp past _FORCE_RECLAIM_GRACE so the second force
    # actually sends (an immediate re-force claims zero by design).
    item.reminded_at = datetime.now(UTC) - timedelta(minutes=30)
    await db_session.commit()
    sent2 = await svc.send_cycle_reminder(cycle.id, force=True)
    assert sent2.recipients_notified == 1
    await db_session.refresh(cycle)
    assert cycle.reminder_sent_at == first


# ── Lead / HRBP escalation (manual opt-in) ─────────────────────────


@pytest.fixture
def capture_dispatch(monkeypatch):
    """Capture every delivered envelope so escalation tests can assert
    on recipient + which pending people each digest names. Patches
    ``deliver`` (not ``dispatch``): both reminder legs enqueue + commit
    the outbox row first, then hand the envelope to ``deliver``."""
    sent: list = []

    async def _capture(self, outbox, envelope, *, trace_id="", actor_user_id=None):
        sent.append(envelope)
        return None

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.deliver",
        _capture,
    )
    return sent


async def _add_pending_assignee(
    db: AsyncSession,
    cycle: AssetInventoryCycle,
    holder: IdentityPerson,
) -> AssetInventoryItem:
    asset = AssetItem(
        asset_tag=f"ESC-{uuid.uuid4().hex[:6]}",
        category="laptop",
        status="claimed",
        location="北京",
        assigned_to=holder.id,
    )
    db.add(asset)
    await db.flush()
    item = AssetInventoryItem(
        id=uuid.uuid4(),
        cycle_id=cycle.id,
        asset_id=asset.id,
        holder_kind="assignee",
        holder_person_id=holder.id,
        asset_tag_snapshot=asset.asset_tag,
        asset_category_snapshot=asset.category,
        asset_status_snapshot=asset.status,
        status="pending",
    )
    db.add(item)
    await db.commit()
    return item


@pytest.mark.asyncio
async def test_no_escalation_by_default(
    db_session: AsyncSession, capture_dispatch,
):
    """Flags default off — the cron path (and a plain manual remind)
    never touches Leads / HRBPs."""
    lead = await _make_person(db_session, "lead0@test.com")
    holder = await _make_person(db_session, "rep0@test.com")
    holder.manager_person_id = lead.id
    await db_session.commit()
    cycle, _ = await _make_active_cycle_with_item(
        db_session, holder, started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)
    out = await svc.send_cycle_reminder(cycle.id, force=True)
    assert out.recipients_notified == 1
    assert out.managers_notified == 0
    assert out.hrbps_notified == 0
    assert all(
        e.template_key != "asset_inventory_escalation" for e in capture_dispatch
    )


@pytest.mark.asyncio
async def test_notify_manager_one_aggregated_card_per_lead(
    db_session: AsyncSession, capture_dispatch,
):
    """Two pending reports under the same Lead → ONE digest card that
    names both. A third report with no manager is silently skipped."""
    lead = await _make_person(db_session, "lead1@test.com")
    rep_a = await _make_person(db_session, "repa@test.com")
    rep_b = await _make_person(db_session, "repb@test.com")
    orphan = await _make_person(db_session, "orphan@test.com")
    rep_a.manager_person_id = lead.id
    rep_b.manager_person_id = lead.id
    # orphan has no manager_person_id
    await db_session.commit()

    cycle, _ = await _make_active_cycle_with_item(
        db_session, rep_a, started_at=datetime.now(UTC) - timedelta(days=2),
    )
    await _add_pending_assignee(db_session, cycle, rep_b)
    await _add_pending_assignee(db_session, cycle, orphan)

    svc = AssetInventoryService(db_session)
    out = await svc.send_cycle_reminder(
        cycle.id, force=True, notify_manager=True,
    )
    assert out.recipients_notified == 3
    # ONE card to the single shared Lead — not one per report.
    assert out.managers_notified == 1
    digests = [
        e for e in capture_dispatch
        if e.template_key == "asset_inventory_escalation"
    ]
    assert len(digests) == 1
    digest = digests[0]
    assert digest.recipient.id == lead.id
    # The card names BOTH unresponsive reports.
    assert "repa" in digest.body_text
    assert "repb" in digest.body_text
    assert digest.context["pending_count"] == 2


@pytest.mark.asyncio
async def test_notify_manager_skips_leaver_lead(
    db_session: AsyncSession, capture_dispatch,
):
    # Create the Lead already-left (NOT flip an active person to "left",
    # which would fire the LeaverBlocked listener and make this test
    # order-dependent in the full suite).
    lead = IdentityPerson(
        id=uuid.uuid4(),
        work_email="exlead@test.com",
        display_name="exlead",
        employment_status="left",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        country="CN",
    )
    db_session.add(lead)
    await db_session.flush()
    rep = await _make_person(db_session, "rep2@test.com")
    rep.manager_person_id = lead.id
    await db_session.commit()
    cycle, _ = await _make_active_cycle_with_item(
        db_session, rep, started_at=datetime.now(UTC) - timedelta(days=2),
    )
    svc = AssetInventoryService(db_session)
    out = await svc.send_cycle_reminder(
        cycle.id, force=True, notify_manager=True,
    )
    assert out.managers_notified == 0
    assert all(
        e.template_key != "asset_inventory_escalation" for e in capture_dispatch
    )


@pytest.mark.asyncio
async def test_notify_hrbp_resolves_and_aggregates(
    db_session: AsyncSession, capture_dispatch, monkeypatch,
):
    """Two pending people share one HRBP → ONE digest to that HRBP."""
    from app.integrations.feishu_hrbp import HrbpResolution

    hrbp = await _make_person(db_session, "thehrbp@test.com")
    p1 = await _make_person(db_session, "emp1@test.com")
    p2 = await _make_person(db_session, "emp2@test.com")
    await db_session.commit()

    async def _fake_resolve(db, person):
        return HrbpResolution(
            person_id=hrbp.id, reason=None, hrbp_email=hrbp.work_email,
        )

    monkeypatch.setattr(
        "app.integrations.feishu_hrbp.resolve_hrbp_for_person",
        _fake_resolve,
    )

    cycle, _ = await _make_active_cycle_with_item(
        db_session, p1, started_at=datetime.now(UTC) - timedelta(days=2),
    )
    await _add_pending_assignee(db_session, cycle, p2)

    svc = AssetInventoryService(db_session)
    out = await svc.send_cycle_reminder(
        cycle.id, force=True, notify_hrbp=True,
    )
    assert out.hrbps_notified == 1
    digests = [
        e for e in capture_dispatch
        if e.template_key == "asset_inventory_escalation"
    ]
    assert len(digests) == 1
    assert digests[0].recipient.id == hrbp.id
    assert digests[0].context["pending_count"] == 2