"""Manager team-joiners surface (PR-3 2026-06-01).

Covers:
  * direct-report manager sees their hire's process
  * skip-level manager sees subordinate's hire
  * unrelated user sees nothing
  * _viewer_can_manage_process gate: 403 on non-subordinate
"""
from __future__ import annotations

import uuid
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import pytest

from app.domains.onboarding.service import OnboardingService
from app.identity.person_models import IdentityPerson


def _today():
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


async def _mk_person(
    db,
    *,
    work_email: str,
    hire_offset_days: int | None = None,
    manager_person_id: uuid.UUID | None = None,
    employment_status: str = "active",
):
    pid = uuid.uuid4()
    p = IdentityPerson(
        id=pid,
        source_system="successfactors",
        hris_person_id=f"sf-{pid.hex[:6]}",
        work_email=work_email,
        display_name=work_email.split("@")[0],
        employment_status=employment_status,
        hire_date=(
            _today() + timedelta(days=hire_offset_days)
            if hire_offset_days is not None
            else None
        ),
        manager_person_id=manager_person_id,
    )
    db.add(p)
    await db.flush()
    return p


@pytest.mark.asyncio
async def test_direct_manager_sees_team_joiner(db_session):
    manager = await _mk_person(db_session, work_email="mgr@x.com")
    hire = await _mk_person(
        db_session,
        work_email="newhire@x.com",
        hire_offset_days=5,
        manager_person_id=manager.id,
    )
    svc = OnboardingService(db_session)
    await svc.start_process(hire.id, hire.hire_date, trace_id="t")
    await db_session.commit()
    items, total = await svc.list_my_team_joiners(manager.id)
    emails = {r["hire_work_email"] for r in items}
    assert "newhire@x.com" in emails
    assert total >= 1


@pytest.mark.asyncio
async def test_skip_level_manager_sees_subordinate_hire(db_session):
    grand = await _mk_person(db_session, work_email="grand@x.com")
    direct = await _mk_person(
        db_session, work_email="direct@x.com", manager_person_id=grand.id,
    )
    hire = await _mk_person(
        db_session,
        work_email="newhire2@x.com",
        hire_offset_days=8,
        manager_person_id=direct.id,
    )
    svc = OnboardingService(db_session)
    await svc.start_process(hire.id, hire.hire_date, trace_id="t")
    await db_session.commit()
    items, _ = await svc.list_my_team_joiners(grand.id)
    emails = {r["hire_work_email"] for r in items}
    assert "newhire2@x.com" in emails


@pytest.mark.asyncio
async def test_unrelated_user_sees_nothing(db_session):
    manager = await _mk_person(db_session, work_email="mgr2@x.com")
    other = await _mk_person(db_session, work_email="other@x.com")
    hire = await _mk_person(
        db_session,
        work_email="newhire3@x.com",
        hire_offset_days=4,
        manager_person_id=manager.id,
    )
    svc = OnboardingService(db_session)
    await svc.start_process(hire.id, hire.hire_date, trace_id="t")
    await db_session.commit()
    items, total = await svc.list_my_team_joiners(other.id)
    emails = {r["hire_work_email"] for r in items}
    assert "newhire3@x.com" not in emails


@pytest.mark.asyncio
async def test_left_hire_not_shown_as_joiner(db_session):
    """Regression (#1664): a hire marked 'left' (rescinded / early departure)
    with a still-open onboarding process must not appear as an incoming joiner."""
    manager = await _mk_person(db_session, work_email="mgr5@x.com")
    hire = await _mk_person(
        db_session,
        work_email="rescinded@x.com",
        hire_offset_days=5,
        manager_person_id=manager.id,
    )
    svc = OnboardingService(db_session)
    await svc.start_process(hire.id, hire.hire_date, trace_id="t")
    # Hire departs before starting; process stays open, employment flips to
    # left. Use a Core UPDATE so we exercise ONLY the joiners query filter —
    # an ORM attribute set would fire the Leaver-Blocked before_flush listener.
    from sqlalchemy import update

    await db_session.execute(
        update(IdentityPerson)
        .where(IdentityPerson.id == hire.id)
        .values(employment_status="left")
    )
    await db_session.commit()
    items, total = await svc.list_my_team_joiners(manager.id)
    emails = {r["hire_work_email"] for r in items}
    assert "rescinded@x.com" not in emails


@pytest.mark.asyncio
async def test_viewer_can_manage_process_gate(db_session):
    manager = await _mk_person(db_session, work_email="mgr3@x.com")
    other = await _mk_person(db_session, work_email="other2@x.com")
    hire = await _mk_person(
        db_session,
        work_email="newhire4@x.com",
        hire_offset_days=6,
        manager_person_id=manager.id,
    )
    svc = OnboardingService(db_session)
    proc = await svc.start_process(hire.id, hire.hire_date, trace_id="t")
    await db_session.commit()
    assert await svc._viewer_can_manage_process(manager.id, proc) is True
    assert await svc._viewer_can_manage_process(other.id, proc) is False