"""Tests for the 工号系统 (employee-number allocation) feature."""

from __future__ import annotations

import uuid
from datetime import UTC, date, datetime

import pytest

from app.core.exceptions import AppException
from app.domains.hr import employee_number_service as svc
from app.domains.hr.models import HrEmployeeNumber
from app.identity.person_models import IdentityPerson


async def _make_person(db_session, *, email: str, name: str) -> uuid.UUID:
    """Minimal active person so a 工号's ``created_by`` FK resolves."""
    pid = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=pid, work_email=email, display_name=name,
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{pid.hex[:8]}",
    ))
    await db_session.flush()
    return pid

# ── Pure helpers ───────────────────────────────────────────────────────


def test_format_employee_no():
    assert svc.format_employee_no(1) == "P00001"
    assert svc.format_employee_no(2388) == "P02388"
    assert svc.format_employee_no(2389) == "P02389"
    # Grows past 99999 naturally.
    assert svc.format_employee_no(100000) == "P100000"


def test_parse_and_canonicalize():
    assert svc.parse_seq("P02388") == 2388
    assert svc.parse_seq("P2388") == 2388  # no leading zeros still parses
    assert svc.parse_seq("X9") is None
    assert svc.parse_seq("P12A") is None
    # Canonicalization collapses P2388 and P02388 to the same key.
    assert svc.canonicalize("P2388") == ("P02388", 2388)
    assert svc.canonicalize("garbage") is None


# ── Allocation (service) ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_allocate_increments_from_empty(db_session):
    n0, seq0, max0 = await svc.get_next(db_session)
    assert n0 == "P00001" and seq0 == 1 and max0 is None

    r1 = await svc.allocate(db_session, name="Alice", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert r1.employee_no == "P00001" and r1.seq == 1

    r2 = await svc.allocate(db_session, name="Bob", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert r2.employee_no == "P00002" and r2.seq == 2

    nxt, seq, mx = await svc.get_next(db_session)
    assert nxt == "P00003" and seq == 3 and mx == "P00002"


@pytest.mark.asyncio
async def test_next_continues_from_seeded_max(db_session):
    # Simulate the migration-249 seed: a P02388 row with no creator.
    db_session.add(HrEmployeeNumber(
        employee_no="P02388", seq=2388, name="Legacy", source="seed",
    ))
    # A legacy non-P value is kept but excluded from the max (seq NULL).
    db_session.add(HrEmployeeNumber(
        employee_no="LEGACY-7", seq=None, name="Contractor", source="seed",
    ))
    await db_session.commit()

    nxt, seq, mx = await svc.get_next(db_session)
    assert nxt == "P02389" and seq == 2389 and mx == "P02388"

    r = await svc.allocate(db_session, name="New", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert r.employee_no == "P02389"


# ── CSV import (service) ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_import_skip_bad_rows(db_session):
    rows = [
        {"工号": "", "姓名": "新人甲", "入职日期": "2026-01-02", "备注": "auto"},      # 1 ok (auto)
        {"工号": "P02388", "姓名": "指定乙", "入职日期": "", "备注": ""},               # 2 ok (explicit)
        {"工号": "P2388", "姓名": "重复丙", "入职日期": "", "备注": ""},                 # 3 dup (canon == P02388)
        {"工号": "X1", "姓名": "格式丁", "入职日期": "", "备注": ""},                    # 4 bad_format
        {"工号": "", "姓名": "", "入职日期": "", "备注": ""},                            # 5 missing_name
        {"工号": "", "姓名": "日期戊", "入职日期": "not-a-date", "备注": ""},            # 6 bad_hire_date
        {"工号": "", "姓名": "新人己", "入职日期": "", "备注": ""},                      # 7 ok (auto, continues)
    ]
    result = await svc.import_rows(db_session, rows, created_by=None)

    assert result["total"] == 7
    assert result["created"] == 3
    assert result["skipped"] == 4
    reasons = {e["row"]: e["reason"] for e in result["errors"]}
    assert reasons == {3: "duplicate", 4: "bad_format", 5: "missing_name", 6: "bad_hire_date"}

    # The first auto row got P00001; the explicit row raised the max to
    # 2388 so the second auto row (row 7) continues from there.
    everything, total = await svc.list_numbers(db_session, page=1, page_size=50)
    by_no = {r["employee_no"]: r for r in everything}
    assert total == 3
    assert "P00001" in by_no  # row 1 auto
    assert "P02388" in by_no  # row 2 explicit
    assert "P02389" in by_no  # row 7 auto, continues past the explicit max
    assert by_no["P00001"]["hire_date"] == date(2026, 1, 2)
    assert all(r["source"] == "import" for r in everything)


@pytest.mark.asyncio
async def test_import_duplicate_against_existing(db_session):
    await svc.allocate(db_session, name="Existing", hire_date=None, remark=None, created_by=None)
    await db_session.commit()  # P00001 exists

    rows = [{"工号": "P00001", "姓名": "撞号", "入职日期": "", "备注": ""}]
    result = await svc.import_rows(db_session, rows, created_by=None)
    assert result["created"] == 0
    assert result["errors"][0]["reason"] == "duplicate"


@pytest.mark.asyncio
async def test_allocate_skips_soft_deleted_seq(db_session):
    # A soft-deleted row still occupies its number on the GLOBAL unique
    # index, so allocation must never re-mint it.
    db_session.add(HrEmployeeNumber(
        employee_no="P00005", seq=5, name="Gone", source="manual",
        deleted_at=datetime(2026, 6, 1, tzinfo=UTC),
    ))
    await db_session.commit()

    nxt, seq, _mx = await svc.get_next(db_session)
    assert nxt == "P00006" and seq == 6
    r = await svc.allocate(db_session, name="New", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert r.employee_no == "P00006"


@pytest.mark.asyncio
async def test_import_duplicate_against_soft_deleted_is_graceful(db_session):
    # Importing a number that matches a soft-deleted row must be reported as
    # a per-row duplicate, NOT crash the whole batch on commit.
    db_session.add(HrEmployeeNumber(
        employee_no="P00009", seq=9, name="Gone", source="manual",
        deleted_at=datetime(2026, 6, 1, tzinfo=UTC),
    ))
    await db_session.commit()

    rows = [
        {"工号": "P9", "姓名": "撞软删", "入职日期": "", "备注": ""},   # canon P00009, dup vs soft-deleted
        {"工号": "", "姓名": "好人", "入职日期": "", "备注": ""},        # auto, must still succeed
    ]
    result = await svc.import_rows(db_session, rows, created_by=None)
    await db_session.commit()
    assert result["created"] == 1
    assert result["errors"] == [{"row": 1, "employee_no": "P00009", "reason": "duplicate"}]
    # Auto row continues past the (soft-deleted) max seq 9 → P00010.
    live, total = await svc.list_numbers(db_session, page=1, page_size=50)
    assert total == 1
    assert live[0]["employee_no"] == "P00010"


@pytest.mark.asyncio
async def test_list_filter_escapes_like_wildcards(db_session):
    await svc.allocate(db_session, name="Alice", hire_date=None, remark=None, created_by=None)
    await svc.allocate(db_session, name="Bob", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    # A literal '_' must match literally (no rows), not act as a single-char
    # wildcard that would match every name.
    rows, total = await svc.list_numbers(db_session, name="A_ice", page=1, page_size=50)
    assert total == 0 and rows == []
    rows2, total2 = await svc.list_numbers(db_session, name="Alice", page=1, page_size=50)
    assert total2 == 1


# ── Report (service) ───────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_report_excludes_seed_from_trend(db_session):
    db_session.add(HrEmployeeNumber(
        employee_no="P02000", seq=2000, name="Seeded", source="seed",
        hire_date=date(2020, 3, 1),
    ))
    await db_session.commit()
    await svc.allocate(
        db_session, name="Manual", hire_date=date(2026, 6, 1), remark=None, created_by=None,
    )
    await db_session.commit()

    report = await svc.build_report(db_session)
    assert report["kpi"]["total"] == 2
    assert report["kpi"]["max_employee_no"] == "P02001"
    assert report["kpi"]["next_employee_no"] == "P02002"

    # Source distribution counts seed + manual.
    src = {b["source"]: b["count"] for b in report["source_distribution"]}
    assert src == {"seed": 1, "manual": 1}

    # Monthly trend excludes the seed row → only the manual allocation.
    trend_total = sum(b["count"] for b in report["monthly_trend"])
    assert trend_total == 1

    # Hire-year distribution includes both (2020 + 2026).
    years = {b["year"]: b["count"] for b in report["hire_year_distribution"]}
    assert years.get("2020") == 1
    assert years.get("2026") == 1


# ── Maintain / delete + reclaim (service) ──────────────────────────────


@pytest.mark.asyncio
async def test_delete_tail_reclaims_number(db_session):
    """Deleting the current max number rolls the sequence back so the next
    allocation hands the SAME number out again (no unique-index collision —
    proves the delete is physical, not soft)."""
    a = await svc.allocate(db_session, name="A", hire_date=None, remark=None, created_by=None)
    b = await svc.allocate(db_session, name="B", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert (a.employee_no, b.employee_no) == ("P00001", "P00002")
    assert (await svc.get_next(db_session))[0] == "P00003"

    # Delete the tail (P00002, the current max).
    await svc.delete_number(db_session, number_id=b.id, actor_id=uuid.uuid4(), is_admin=True)
    await db_session.commit()

    # Next rolls back to the freed number, and re-allocation re-mints it.
    assert (await svc.get_next(db_session))[0] == "P00002"
    c = await svc.allocate(db_session, name="C", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert c.employee_no == "P00002"


@pytest.mark.asyncio
async def test_delete_middle_keeps_gap_not_reclaimed(db_session):
    """Deleting a number BELOW the current max leaves a gap; allocation
    continues from the surviving max (the gap is not back-filled)."""
    a = await svc.allocate(db_session, name="A", hire_date=None, remark=None, created_by=None)
    _b = await svc.allocate(db_session, name="B", hire_date=None, remark=None, created_by=None)
    c = await svc.allocate(db_session, name="C", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert (a.employee_no, c.employee_no) == ("P00001", "P00003")

    # Delete P00001 (below the max P00003).
    await svc.delete_number(db_session, number_id=a.id, actor_id=uuid.uuid4(), is_admin=True)
    await db_session.commit()

    # Max is still P00003 -> next is P00004, the P00001 gap stays open.
    assert (await svc.get_next(db_session))[0] == "P00004"
    d = await svc.allocate(db_session, name="D", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert d.employee_no == "P00004"


@pytest.mark.asyncio
async def test_update_changes_fields_but_not_number(db_session):
    row = await svc.allocate(
        db_session, name="Old", hire_date=date(2020, 1, 1), remark="r0", created_by=None,
    )
    await db_session.commit()
    no, seq = row.employee_no, row.seq

    updated, before, after = await svc.update_number(
        db_session, number_id=row.id, actor_id=uuid.uuid4(), is_admin=True,
        name="New", hire_date=date(2021, 2, 2), remark="r1", bu=None,
    )
    await db_session.commit()

    # 工号 immutable; descriptive fields changed.
    assert updated.employee_no == no and updated.seq == seq
    assert updated.name == "New" and updated.remark == "r1"
    assert before["name"] == "Old" and after["name"] == "New"


@pytest.mark.asyncio
async def test_owner_authz_on_update_and_delete(db_session):
    """Only the creator (or a global_admin) may maintain/delete a row."""
    owner = await _make_person(db_session, email="owner@x.io", name="Owner")
    other = await _make_person(db_session, email="other@x.io", name="Other")
    row = await svc.allocate(
        db_session, name="X", hire_date=None, remark=None, created_by=owner,
    )
    await db_session.commit()

    # A non-owner, non-admin is refused on both verbs.
    for verb in (
        lambda: svc.update_number(
            db_session, number_id=row.id, actor_id=other, is_admin=False,
            name="hack", hire_date=None, remark=None, bu=None,
        ),
        lambda: svc.delete_number(
            db_session, number_id=row.id, actor_id=other, is_admin=False,
        ),
    ):
        with pytest.raises(AppException) as ei:
            await verb()
        assert ei.value.status_code == 403

    # The owner themselves may edit.
    await svc.update_number(
        db_session, number_id=row.id, actor_id=owner, is_admin=False,
        name="ok", hire_date=None, remark=None, bu=None,
    )
    await db_session.commit()

    # A global_admin may delete someone else's row.
    await svc.delete_number(
        db_session, number_id=row.id, actor_id=other, is_admin=True,
    )
    await db_session.commit()


@pytest.mark.asyncio
async def test_delete_missing_row_404(db_session):
    with pytest.raises(AppException) as ei:
        await svc.delete_number(
            db_session, number_id=uuid.uuid4(), actor_id=uuid.uuid4(), is_admin=True,
        )
    assert ei.value.status_code == 404


# ── API + RBAC ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_create_and_next_via_api(client, hr_user):
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}

    resp = await client.post(
        "/api/v1/hr/employee-numbers",
        json={"name": "张三", "hire_date": "2026-06-04", "remark": "test"},
        headers=h,
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["employee_no"] == "P00001"
    assert data["source"] == "manual"
    assert data["created_by_name"] == "HR User"

    nxt = await client.get("/api/v1/hr/employee-numbers/next", headers=h)
    assert nxt.json()["data"]["next_employee_no"] == "P00002"

    lst = await client.get("/api/v1/hr/employee-numbers", headers=h)
    body = lst.json()
    assert body["meta"]["total"] == 1
    assert body["data"][0]["employee_no"] == "P00001"


@pytest.mark.asyncio
async def test_creator_edit_delete_and_reclaim_via_api(client, hr_user):
    """Creator maintains then deletes their own 工号; the deleted tail number
    is reclaimed by the next allocation."""
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}

    created = (await client.post(
        "/api/v1/hr/employee-numbers",
        json={"name": "张三", "remark": "v0"}, headers=h,
    )).json()["data"]
    nid = created["id"]
    assert created["employee_no"] == "P00001"

    # 维护: change descriptive fields; the number stays put.
    patched = await client.patch(
        f"/api/v1/hr/employee-numbers/{nid}",
        json={"name": "李四", "remark": "v1"}, headers=h,
    )
    assert patched.status_code == 200, patched.text
    pd = patched.json()["data"]
    assert pd["employee_no"] == "P00001" and pd["name"] == "李四" and pd["remark"] == "v1"

    # Next number had advanced to P00002 while the row existed.
    assert (await client.get("/api/v1/hr/employee-numbers/next", headers=h)
            ).json()["data"]["next_employee_no"] == "P00002"

    # 删除: the tail number is freed and reclaimed.
    deleted = await client.delete(f"/api/v1/hr/employee-numbers/{nid}", headers=h)
    assert deleted.status_code == 200, deleted.text
    assert deleted.json()["data"]["employee_no"] == "P00001"
    assert (await client.get("/api/v1/hr/employee-numbers/next", headers=h)
            ).json()["data"]["next_employee_no"] == "P00001"

    # Gone from the ledger; re-creating re-mints the freed number.
    assert (await client.get("/api/v1/hr/employee-numbers", headers=h)
            ).json()["meta"]["total"] == 0
    again = (await client.post(
        "/api/v1/hr/employee-numbers", json={"name": "王五"}, headers=h,
    )).json()["data"]
    assert again["employee_no"] == "P00001"


@pytest.mark.asyncio
async def test_employee_forbidden(client, employee_user):
    _user, token = employee_user
    h = {"Authorization": f"Bearer {token}"}

    # Read gate excludes plain employees.
    assert (await client.get("/api/v1/hr/employee-numbers", headers=h)).status_code == 403
    # Write gate too.
    create = await client.post(
        "/api/v1/hr/employee-numbers",
        json={"name": "X"},
        headers=h,
    )
    assert create.status_code == 403


@pytest.mark.asyncio
async def test_import_and_export_via_api(client, hr_user):
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}

    csv_bytes = (
        "工号,姓名,入职日期,备注\n"
        ",李四,2026-02-03,\n"
        "P05000,王五,,vip\n"
    ).encode("utf-8-sig")

    imp = await client.post(
        "/api/v1/hr/employee-numbers/import",
        files={"file": ("numbers.csv", csv_bytes, "text/csv")},
        headers=h,
    )
    assert imp.status_code == 200, imp.text
    res = imp.json()["data"]
    assert res["created"] == 2 and res["skipped"] == 0

    exp = await client.get("/api/v1/hr/employee-numbers/export", headers=h)
    assert exp.status_code == 200
    text = exp.content.decode("utf-8-sig")
    assert "工号,姓名,入职日期,备注" in text
    assert "P05000" in text and "王五" in text


@pytest.mark.asyncio
async def test_import_rejects_non_utf8_with_stable_client_error(client, hr_user):
    _user, token = hr_user
    response = await client.post(
        "/api/v1/hr/employee-numbers/import",
        files={"file": ("numbers.csv", b"\xff\xfe\x00", "text/csv")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 400
    assert response.json()["errors"][0]["code"] == "EMPLOYEE_NUMBER_IMPORT_INVALID_CSV"


@pytest.mark.asyncio
async def test_import_rejects_row_amplification_before_database_work(
    client, hr_user, monkeypatch,
):
    from app.domains.hr import employee_number_router as router_module

    _user, token = hr_user
    monkeypatch.setattr(router_module, "_MAX_IMPORT_ROWS", 2)
    response = await client.post(
        "/api/v1/hr/employee-numbers/import",
        files={
            "file": (
                "numbers.csv",
                "工号,姓名\n,一\n,二\n,三\n".encode(),
                "text/csv",
            ),
        },
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 413
    assert response.json()["errors"][0]["code"] == (
        "EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED"
    )


@pytest.mark.asyncio
async def test_create_requires_name(client, hr_user):
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}
    resp = await client.post(
        "/api/v1/hr/employee-numbers", json={"name": "  "}, headers=h,
    )
    assert resp.status_code == 422


# ── Manual next-number override (工号维护) ──────────────────────────────


@pytest.mark.asyncio
async def test_set_next_raises_floor_then_goes_inert(db_session):
    await svc.allocate(db_session, name="A", hire_date=None, remark=None, created_by=None)
    await db_session.commit()  # P00001

    ov = await svc.set_next(
        db_session, next_employee_no="P05000", reason="reserve range", created_by=None,
    )
    await db_session.commit()
    assert ov.next_seq == 5000 and ov.next_employee_no == "P05000"
    # before→after diff captured for the log.
    assert ov.previous_next_employee_no == "P00002"

    nxt, seq, mx = await svc.get_next(db_session)
    assert nxt == "P05000" and seq == 5000 and mx == "P00001"

    r = await svc.allocate(db_session, name="B", hire_date=None, remark=None, created_by=None)
    await db_session.commit()
    assert r.employee_no == "P05000"

    # Once allocations reach the floor it becomes inert — MAX(seq)+1 wins.
    nxt2, seq2, _ = await svc.get_next(db_session)
    assert nxt2 == "P05001" and seq2 == 5001


@pytest.mark.asyncio
async def test_set_next_canonicalizes_unpadded(db_session):
    ov = await svc.set_next(
        db_session, next_employee_no="P900", reason="x", created_by=None,
    )
    await db_session.commit()
    assert ov.next_employee_no == "P00900" and ov.next_seq == 900
    nxt, _seq, _mx = await svc.get_next(db_session)
    assert nxt == "P00900"


@pytest.mark.asyncio
async def test_set_next_must_exceed_max(db_session):
    from app.core.exceptions import AppException

    await svc.allocate(db_session, name="A", hire_date=None, remark=None, created_by=None)
    await db_session.commit()  # P00001, max seq = 1

    # Equal to / below the current max would collide with an existing 工号.
    with pytest.raises(AppException):
        await svc.set_next(db_session, next_employee_no="P00001", reason="x", created_by=None)
    # Non-canonical format is rejected.
    with pytest.raises(AppException):
        await svc.set_next(db_session, next_employee_no="garbage", reason="x", created_by=None)


@pytest.mark.asyncio
async def test_latest_override_wins_and_history_newest_first(db_session):
    await svc.set_next(db_session, next_employee_no="P05000", reason="first", created_by=None)
    await db_session.commit()
    await svc.set_next(db_session, next_employee_no="P03000", reason="second", created_by=None)
    await db_session.commit()

    # Latest wins even when lower (still > max=0): HR explicitly lowered it.
    nxt, _seq, _mx = await svc.get_next(db_session)
    assert nxt == "P03000"

    rows, total = await svc.list_overrides(db_session, page=1, page_size=10)
    assert total == 2
    assert rows[0]["next_employee_no"] == "P03000"  # newest first
    assert rows[0]["reason"] == "second"
    assert rows[1]["next_employee_no"] == "P05000"


# ── BU (一级部门) ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_allocate_stores_and_lists_bu(db_session):
    await svc.allocate(
        db_session, name="A", hire_date=None, remark=None, created_by=None,
        bu="Engineering Group",
    )
    await db_session.commit()
    rows, _total = await svc.list_numbers(db_session, page=1, page_size=10)
    assert rows[0]["bu"] == "Engineering Group"


def test_resolve_bu():
    bu_map = {"engineering group": "Engineering Group"}
    assert svc.resolve_bu("", bu_map) == (None, True)
    assert svc.resolve_bu(None, bu_map) == (None, True)
    # Case-insensitive + trimmed, returns the canonical name.
    assert svc.resolve_bu("  engineering GROUP ", bu_map) == ("Engineering Group", True)
    # Unknown -> echoed back with ok=False.
    assert svc.resolve_bu("Marketing", bu_map) == ("Marketing", False)


@pytest.mark.asyncio
async def test_create_unknown_bu_rejected(client, hr_user):
    # The test DB has no org tree, so every non-blank BU is unknown.
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}
    resp = await client.post(
        "/api/v1/hr/employee-numbers",
        json={"name": "张三", "bu": "Nonexistent BU"},
        headers=h,
    )
    assert resp.status_code == 400, resp.text
    assert resp.json()["errors"][0]["code"] == "UNKNOWN_BU"


@pytest.mark.asyncio
async def test_import_unknown_bu_skipped(db_session):
    rows = [
        {"工号": "", "姓名": "甲", "入职日期": "", "备注": "", "一级部门": "鬼部门"},
        {"工号": "", "姓名": "乙", "入职日期": "", "备注": "", "一级部门": ""},
    ]
    result = await svc.import_rows(db_session, rows, created_by=None)
    assert result["created"] == 1
    assert result["errors"] == [{"row": 1, "employee_no": None, "reason": "unknown_bu"}]


# ── Set-next + history via API ─────────────────────────────────────────


@pytest.mark.asyncio
async def test_set_next_and_history_via_api(client, hr_user):
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}

    resp = await client.post(
        "/api/v1/hr/employee-numbers/next",
        json={"next_employee_no": "P09000", "reason": "对齐 HRIS"},
        headers=h,
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["next_employee_no"] == "P09000"

    nxt = await client.get("/api/v1/hr/employee-numbers/next", headers=h)
    assert nxt.json()["data"]["next_employee_no"] == "P09000"

    hist = await client.get("/api/v1/hr/employee-numbers/next/history", headers=h)
    body = hist.json()
    assert body["meta"]["total"] == 1
    assert body["data"][0]["next_employee_no"] == "P09000"
    assert body["data"][0]["reason"] == "对齐 HRIS"
    assert body["data"][0]["created_by_name"] == "HR User"


@pytest.mark.asyncio
async def test_set_next_requires_reason(client, hr_user):
    _user, token = hr_user
    h = {"Authorization": f"Bearer {token}"}
    resp = await client.post(
        "/api/v1/hr/employee-numbers/next",
        json={"next_employee_no": "P09000", "reason": "  "},
        headers=h,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_set_next_forbidden_for_employee(client, employee_user):
    _user, token = employee_user
    h = {"Authorization": f"Bearer {token}"}
    resp = await client.post(
        "/api/v1/hr/employee-numbers/next",
        json={"next_employee_no": "P09000", "reason": "x"},
        headers=h,
    )
    assert resp.status_code == 403
    # History read is also gated.
    assert (
        await client.get("/api/v1/hr/employee-numbers/next/history", headers=h)
    ).status_code == 403