"""GET /me/onboarding self-service endpoint (PR-4 2026-06-01).""" from __future__ import annotations import uuid from datetime import datetime, timedelta from zoneinfo import ZoneInfo import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.domains.onboarding.me_router import _build_self_payload 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: AsyncSession, *, hire_offset_days: int, email: str): pid = uuid.uuid4() p = IdentityPerson( id=pid, source_system="successfactors", hris_person_id=f"sf-{pid.hex[:6]}", work_email=email, display_name=email.split("@")[0], employment_status="active", hire_date=_today() + timedelta(days=hire_offset_days), company_name="PingCAP", job_title="Engineer", ) db.add(p) await db.flush() return p @pytest.mark.asyncio async def test_self_payload_shape(db_session): p = await _mk_person(db_session, hire_offset_days=2, email="newjoe@x.com") await OnboardingService(db_session).start_process( p.id, p.hire_date, trace_id="t", ) await db_session.commit() payload = await _build_self_payload(p.id, db_session) assert payload["process"]["hire_date"] == p.hire_date.isoformat() assert payload["self"]["work_email"] == "newjoe@x.com" assert payload["self"]["display_name"] == "newjoe" assert "dimensions" in payload assert len(payload["dimensions"]) == 10 # 入职须知 checklist ships in the same payload (empty when no items # are configured). assert payload["acknowledgements"] == [] # work_email is the LIVE-mailbox gate now (2026-07-06): a bare hire # has only the reserved SF address and no login identity, so the # mailbox isn't live → not satisfied. (Before the fix any `x@y` # string greened it.) assert payload["dimensions"]["work_email"] is False # SSO / seat / computer / vpn all start un-provisioned → false assert payload["dimensions"]["sso"] is False assert isinstance(payload["blockers"], list) @pytest.mark.asyncio async def test_no_open_process_raises(db_session): p = await _mk_person( db_session, hire_offset_days=-365, email="ancient@x.com", ) await db_session.commit() # No start_process call → no open row → raises. from app.core.exceptions import AppException with pytest.raises(AppException) as ex: await _build_self_payload(p.id, db_session) assert ex.value.code == "NO_OPEN_ONBOARDING_PROCESS" @pytest.mark.asyncio async def test_self_payload_includes_manager(db_session): mgr = await _mk_person( db_session, hire_offset_days=-365, email="boss@x.com", ) p = await _mk_person(db_session, hire_offset_days=3, email="newhire@x.com") p.manager_person_id = mgr.id await db_session.flush() svc = OnboardingService(db_session) await svc.start_process(p.id, p.hire_date, trace_id="t") await db_session.commit() payload = await _build_self_payload(p.id, db_session) assert payload["manager"] is not None assert payload["manager"]["work_email"] == "boss@x.com" # buddy concept removed 2026-06-15 — payload no longer carries it. assert "buddy" not in payload