timemaster/backend/tests/test_coverage_gaps.py

"""Gezielte Tests für kritische, ungetestete Pfade (postgres-expert Audit).

Deckt:
1. Carryover-Expiry über create_absence (kein Fehlsignal "Konto reicht")
2. Teilzeit + Pro-rata gleichzeitig (_compute_entitlement kombiniert)
3. FZA-Rückbuchung via Cancellation-Request-Flow (statt Admin-Direkt-Storno)
4. Zwei-Stufen-Genehmigung: kein Doppelabzug/-gutschrift bei Storno in FIRST_APPROVED
   + Schwellwert-Grenzfall working_days == two_stage_min_days
5. DATEV-Export: Feiertag-Kürzel "F" schlägt Urlaubs-Kürzel "U"; working_days konsistent
6. _categorize_hours: Nachtschicht über Mitternacht + Sonntag+Feiertag (höchster Zuschlag gewinnt)
7. POST /absences/ reicht warnings jetzt durch (API-Konsistenz-Fix)

Nebenbei gefixte Bugs (während dieser Coverage-Arbeit gefunden):
- _get_holiday_dates() rechnete Feiertage im Folgejahr bei Anträgen über den
  Jahreswechsel nicht aus working_days heraus (siehe
  test_holiday_dates_cover_year_boundary + absence_service.py year_to-Parameter)
- POST /absences/ verwarf die warnings-Liste von create_absence() (siehe Test 7)
"""
import pytest
import pytest_asyncio
from datetime import date, time, timedelta
from decimal import Decimal
from uuid import UUID
from httpx import AsyncClient
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.company import Company
from app.models.overtime_balance import OvertimeBalance
from app.models.user import User
from app.models.vacation_balance import VacationBalance


# ── Fixtures ──────────────────────────────────────────────────────────────────

@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_company(client: AsyncClient):
    resp = await client.post("/api/v1/auth/register", json={
        "company_name": "Coverage Gaps GmbH",
        "first_name": "Cov",
        "last_name": "Admin",
        "email": "admin@covgaps.de",
        "password": "Secret123",
    })
    assert resp.status_code == 201, resp.text
    tokens = resp.json()
    me = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {tokens['access_token']}"},
    )
    return {"tokens": tokens, "user": me.json()}


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_headers(cov_company):
    return {"Authorization": f"Bearer {cov_company['tokens']['access_token']}"}


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_approver_headers(client: AsyncClient, cov_headers):
    """Zweiter Admin – für Genehmigungen, die nicht self-approval sein dürfen."""
    resp = await client.post("/api/v1/users/invite", json={
        "first_name": "Cov", "last_name": "Approver",
        "email": "approver@covgaps.de", "role": "COMPANY_ADMIN",
        "initial_password": "Secret123",
    }, headers=cov_headers)
    assert resp.status_code == 201, resp.text
    login = await client.post("/api/v1/auth/login", json={
        "email": "approver@covgaps.de", "password": "Secret123",
    })
    assert login.status_code == 200, login.text
    return {"Authorization": f"Bearer {login.json()['access_token']}"}


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_third_approver_headers(client: AsyncClient, cov_headers):
    """Dritter Admin – für die finale Zwei-Stufen-Genehmigung (andere Person als Stufe 1)."""
    resp = await client.post("/api/v1/users/invite", json={
        "first_name": "Cov", "last_name": "Third",
        "email": "third@covgaps.de", "role": "HR",
        "initial_password": "Secret123",
    }, headers=cov_headers)
    assert resp.status_code == 201, resp.text
    login = await client.post("/api/v1/auth/login", json={
        "email": "third@covgaps.de", "password": "Secret123",
    })
    assert login.status_code == 200, login.text
    return {"Authorization": f"Bearer {login.json()['access_token']}"}


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_vacation_type_id(client: AsyncClient, cov_headers):
    resp = await client.get("/api/v1/absence-types/", headers=cov_headers)
    types = resp.json()
    vacation = next((t for t in types if t["name"] == "Urlaub"), types[0])
    return vacation["id"]


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def cov_fza_type_id(client: AsyncClient, cov_headers):
    resp = await client.post("/api/v1/absence-types/", json={
        "name": "FZA Coverage",
        "category": "overtime_comp",
        "color": "#f97316",
        "requires_approval": True,
        "deducts_vacation": False,
        "affects_overtime_balance": True,
    }, headers=cov_headers)
    assert resp.status_code == 201, resp.text
    return resp.json()["id"]


def _future_monday(weeks: int) -> date:
    return date.today() + timedelta(days=(7 - date.today().weekday()) + 7 * weeks)


# ── 1. Carryover-Expiry über echten Integrationspfad ───────────────────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_carryover_expiry_via_create_absence(
    client: AsyncClient, db_session: AsyncSession, cov_company, cov_headers, cov_vacation_type_id,
):
    """Ein User mit abgelaufenem carried_over-Betrag darf im create_absence-Pfad
    nicht das Signal 'Konto reicht' bekommen, obwohl entitled+carried_over rechnerisch
    reichen würde – der verfallene Übertrag darf effective_available nicht aufblähen.

    Ruft absence_service.create_absence() direkt auf (statt über HTTP) um den
    warnings-Rückgabewert isoliert vom HTTP-Layer zu prüfen (Test 7 unten deckt
    den HTTP-Pfad inkl. warnings-Durchreichung separat ab)."""
    from app.services.absence_service import absence_service
    from app.schemas.absence import AbsenceCreate

    user_id = cov_company["user"]["id"]
    year = date.today().year

    # Firma: Verfallsdatum in der Vergangenheit setzen (z.B. 31.01. dieses Jahr)
    r = await client.patch("/api/v1/companies/me", json={
        "settings": {"carryover_expires_month": 1, "carryover_expires_day": 31},
    }, headers=cov_headers)
    assert r.status_code == 200, r.text

    # RLS-Bypass gilt nur innerhalb der aktuellen Transaktion (SET LOCAL) -- muss
    # daher vor JEDER Query auf diesem "rohen" db_session (ohne app.company_id-
    # Kontext eines eingeloggten Requests) neu gesetzt werden, siehe
    # project_rls_bypass_commit_pitfall in Memory.
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    user = await db_session.get(User, user_id)
    assert user is not None

    # VacationBalance direkt präparieren: wenig entitled, hoher (verfallener) Übertrag,
    # kaum genutzt -> ohne Verfalls-Logik würde das Konto "reichen"
    balance = await db_session.scalar(
        select(VacationBalance).where(VacationBalance.user_id == user_id, VacationBalance.year == year)
    )
    if balance is None:
        balance = VacationBalance(user_id=user_id, year=year)
        db_session.add(balance)
        await db_session.flush()
    balance.entitled_days = 5
    balance.carried_over = 20
    balance.used_days = 0
    balance.special_days = 0
    await db_session.flush()

    # Antrag über 10 Arbeitstage im selben Jahr wie die präparierte Bilanz (nah
    # in der Zukunft, damit wir nicht ins nächste Jahr rutschen)
    start = _future_monday(1)
    assert start.year == year, "Testannahme verletzt: Zeitraum muss im selben Jahr wie die Bilanz liegen"
    create_data = AbsenceCreate(
        type_id=cov_vacation_type_id,
        start_date=start, end_date=start + timedelta(days=13),  # 2 volle Wochen = 10 Werktage
    )
    absence, warnings = await absence_service.create_absence(create_data, user, db_session)
    assert absence.working_days == 10

    assert any("Urlaubskonto reicht" in w for w in warnings), (
        f"Erwartete Verfalls-Warnung fehlt (Bug: effective_available berücksichtigt Verfall nicht "
        f"im create_absence-Pfad?), warnings={warnings}"
    )
    await db_session.commit()

    # Reset für nachfolgende Tests
    await client.patch("/api/v1/companies/me", json={"settings": {}}, headers=cov_headers)
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    bal2 = await db_session.scalar(
        select(VacationBalance).where(VacationBalance.user_id == user_id, VacationBalance.year == year)
    )
    if bal2:
        bal2.entitled_days = 30
        bal2.carried_over = 0
    await db_session.commit()


# ── 2. Teilzeit + Pro-rata gleichzeitig ─────────────────────────────────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_part_time_prorate_combined(client: AsyncClient, cov_headers):
    """company.vacation_from_schedule=True UND vacation_prorate_first_year=True gleichzeitig:
    Anspruch = default_days * (Arbeitstage/Woche / 5) * (Monate ab Eintritt / 12), gerundet."""
    sched = await client.post("/api/v1/time/schedules", json={
        "name": "Teilzeit 3 Tage Cov", "mon_h": 8, "tue_h": 8, "wed_h": 8,
        "thu_h": 0, "fri_h": 0, "valid_from": "2026-01-01",
    }, headers=cov_headers)
    assert sched.status_code == 201, sched.text
    sid = sched.json()["id"]

    await client.patch("/api/v1/companies/me", json={
        "vacation_from_schedule": True,
        "vacation_prorate_first_year": True,
        "vacation_default_days": 30,
    }, headers=cov_headers)

    yr = date.today().year
    inv = await client.post("/api/v1/users/invite", json={
        "first_name": "Combo", "last_name": "Case", "email": "combo@covgaps.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=cov_headers)
    assert inv.status_code == 201, inv.text
    uid = inv.json()["id"]
    await client.patch(f"/api/v1/users/{uid}", json={
        "entry_date": f"{yr}-07-01", "work_schedule_id": sid,
    }, headers=cov_headers)

    bal = await client.get(f"/api/v1/absences/balance/{uid}?year={yr}", headers=cov_headers)
    assert bal.status_code == 200, bal.text

    # 30 * 3/5 = 18 (Teilzeit) * 6/12 (Eintritt Juli -> Monate Jul..Dez = 6) = 9
    # math.floor(9 + 0.5) = 9
    days = 30.0 * 3 / 5.0
    days = days * 6 / 12.0
    import math
    expected = int(math.floor(days + 0.5))
    assert bal.json()["entitled_days"] == expected == 9

    await client.patch("/api/v1/companies/me", json={
        "vacation_from_schedule": False, "vacation_prorate_first_year": False,
    }, headers=cov_headers)


# ── 3. FZA-Rückbuchung via Cancellation-Request-Flow ────────────────────────────

async def _seed_overtime_balance(db_session: AsyncSession, user_id: str, company_id: str, total_hours: float) -> None:
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    ob = await db_session.scalar(select(OvertimeBalance).where(OvertimeBalance.user_id == user_id))
    if ob is None:
        ob = OvertimeBalance(user_id=user_id, company_id=company_id, total_hours=Decimal(str(total_hours)))
        db_session.add(ob)
    else:
        ob.total_hours = Decimal(str(total_hours))
        ob.taken_hours = Decimal("0")
    await db_session.flush()


@pytest.mark.asyncio(loop_scope="session")
async def test_fza_refund_via_cancellation_request_flow(
    client: AsyncClient, db_session: AsyncSession, cov_company, cov_headers,
    cov_approver_headers, cov_fza_type_id,
):
    """Kompletter Storno-Antrags-Flow (nicht der Admin-Direkt-Pfad) für einen
    FZA-Antrag: Antrag -> genehmigt (Konto sinkt) -> Storno beantragt -> Storno
    genehmigt -> Konto muss zurückgebucht sein."""
    emp = await client.post("/api/v1/users/invite", json={
        "first_name": "Fza", "last_name": "Requestor", "email": "fzareq@covgaps.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=cov_headers)
    assert emp.status_code == 201, emp.text
    emp_id = emp.json()["id"]
    company_id = cov_company["user"]["company_id"]

    await _seed_overtime_balance(db_session, emp_id, company_id, total_hours=40.0)
    await db_session.commit()

    login = await client.post("/api/v1/auth/login", json={
        "email": "fzareq@covgaps.de", "password": "Secret123",
    })
    emp_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

    start = _future_monday(25)
    resp = await client.post("/api/v1/absences/", json={
        "type_id": cov_fza_type_id, "start_date": str(start), "end_date": str(start),
    }, headers=emp_headers)
    assert resp.status_code == 201, resp.text
    aid = resp.json()["id"]

    approve = await client.post(f"/api/v1/absences/{aid}/approve", headers=cov_approver_headers)
    assert approve.status_code == 200, approve.text

    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    ob = await db_session.scalar(
        select(OvertimeBalance).where(OvertimeBalance.user_id == emp_id)
        .execution_options(populate_existing=True)
    )
    taken_after_approve = float(ob.taken_hours)
    assert taken_after_approve == pytest.approx(8.0, abs=0.1)

    # Mitarbeiter beantragt Storno statt Admin-Direktstorno
    req = await client.post(
        f"/api/v1/absences/{aid}/request-cancellation", json={"reason": "Plan geaendert"},
        headers=emp_headers,
    )
    assert req.status_code == 200, req.text
    assert req.json()["status"] == "cancellation_requested"

    ok = await client.post(f"/api/v1/absences/{aid}/approve", headers=cov_approver_headers)
    assert ok.status_code == 200, ok.text
    assert ok.json()["status"] == "cancelled"

    import asyncio
    await asyncio.sleep(0.15)  # fire-and-forget CalDAV
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    ob2 = await db_session.scalar(
        select(OvertimeBalance).where(OvertimeBalance.user_id == emp_id)
        .execution_options(populate_existing=True)
    )
    taken_after_cancel = float(ob2.taken_hours)
    assert taken_after_cancel == pytest.approx(0.0, abs=0.1), (
        f"FZA-Rückbuchung über Cancellation-Request-Flow fehlgeschlagen, got {taken_after_cancel}"
    )


# ── 4. Zwei-Stufen-Genehmigung + Storno-Grenzfälle ──────────────────────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_two_stage_no_double_credit_on_cancellation(
    client: AsyncClient, db_session: AsyncSession, cov_company, cov_headers,
    cov_approver_headers, cov_vacation_type_id,
):
    """Antrag -> FIRST_APPROVED (Stufe 1, KEIN Urlaubsabzug) -> Storno in diesem
    Zustand darf keine Rückbuchung auslösen (es wurde ja nichts gebucht) -> Antrag
    entweder direkt stornierbar oder Cancellation-Request nicht möglich (Business-Regel:
    request_cancellation verlangt Status APPROVED). Konto wird bei jedem Schritt geprüft."""
    await client.patch("/api/v1/companies/me",
                        json={"two_stage_approval_enabled": True, "two_stage_min_days": 0},
                        headers=cov_headers)

    inv = await client.post("/api/v1/users/invite", json={
        "first_name": "Two", "last_name": "Cov", "email": "twocov@covgaps.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=cov_headers)
    assert inv.status_code == 201, inv.text
    emp_id = inv.json()["id"]
    login = await client.post("/api/v1/auth/login", json={
        "email": "twocov@covgaps.de", "password": "Secret123",
    })
    emp_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

    yr = date.today().year
    start = _future_monday(27)
    used_before = (await client.get(
        f"/api/v1/absences/balance/{emp_id}?year={yr}", headers=cov_headers
    )).json()["used_days"]

    cr = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start),
        "end_date": str(start + timedelta(days=2)),  # 3 Werktage
    }, headers=emp_headers)
    assert cr.status_code == 201, cr.text
    aid = cr.json()["id"]

    # Stufe 1
    r1 = await client.post(f"/api/v1/absences/{aid}/approve", headers=cov_approver_headers)
    assert r1.status_code == 200, r1.text
    assert r1.json()["status"] == "first_approved"

    used_after_first = (await client.get(
        f"/api/v1/absences/balance/{emp_id}?year={yr}", headers=cov_headers
    )).json()["used_days"]
    assert used_after_first == used_before, (
        "Urlaub wurde bereits bei FIRST_APPROVED abgezogen -- Doppelgutschriftsrisiko bei Storno!"
    )

    # Business-Regel prüfen: Storno-Antrag ist bei FIRST_APPROVED (noch nicht APPROVED)
    # explizit nicht erlaubt (nur genehmigte Anträge, siehe request_cancellation).
    req = await client.post(
        f"/api/v1/absences/{aid}/request-cancellation", json={}, headers=emp_headers,
    )
    assert req.status_code == 409, (
        f"Erwartet: Cancellation-Request bei FIRST_APPROVED wird abgelehnt (409), got {req.status_code}"
    )

    # Der Owner kann den Antrag aber direkt stornieren (Storno von PENDING/FIRST_APPROVED
    # ist über cancel_absence/DELETE erlaubt und darf NICHTS zurückbuchen, da nie
    # etwas abgezogen wurde).
    cancel = await client.delete(f"/api/v1/absences/{aid}", headers=emp_headers)
    assert cancel.status_code == 200, cancel.text
    assert cancel.json()["status"] == "cancelled"

    used_after_cancel = (await client.get(
        f"/api/v1/absences/balance/{emp_id}?year={yr}", headers=cov_headers
    )).json()["used_days"]
    assert used_after_cancel == used_before, (
        f"Doppelgutschrift-Bug: used_days veränderte sich durch Storno eines nie abgezogenen "
        f"FIRST_APPROVED-Antrags ({used_before} -> {used_after_cancel})"
    )

    await client.patch("/api/v1/companies/me",
                        json={"two_stage_approval_enabled": False}, headers=cov_headers)


@pytest.mark.asyncio(loop_scope="session")
async def test_two_stage_threshold_boundary_equal_triggers_two_stage(
    client: AsyncClient, cov_headers, cov_approver_headers, cov_third_approver_headers, cov_vacation_type_id,
):
    """Grenzfall working_days == two_stage_min_days: Code nutzt >= -> muss zwei Stufen auslösen."""
    await client.patch("/api/v1/companies/me",
                        json={"two_stage_approval_enabled": True, "two_stage_min_days": 3},
                        headers=cov_headers)

    inv = await client.post("/api/v1/users/invite", json={
        "first_name": "Boundary", "last_name": "Case", "email": "boundary@covgaps.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=cov_headers)
    assert inv.status_code == 201, inv.text
    login = await client.post("/api/v1/auth/login", json={
        "email": "boundary@covgaps.de", "password": "Secret123",
    })
    emp_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

    start = _future_monday(29)
    cr = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start),
        "end_date": str(start + timedelta(days=2)),  # exakt 3 Werktage = Schwelle
    }, headers=emp_headers)
    assert cr.status_code == 201, cr.text
    assert cr.json()["working_days"] == 3
    aid = cr.json()["id"]

    r1 = await client.post(f"/api/v1/absences/{aid}/approve", headers=cov_approver_headers)
    assert r1.status_code == 200, r1.text
    assert r1.json()["status"] == "first_approved", (
        "working_days == two_stage_min_days sollte zwei Stufen auslösen (Code nutzt >=)"
    )

    r2 = await client.post(f"/api/v1/absences/{aid}/approve", headers=cov_third_approver_headers)
    assert r2.status_code == 200, r2.text
    assert r2.json()["status"] == "approved"

    await client.patch("/api/v1/companies/me",
                        json={"two_stage_approval_enabled": False}, headers=cov_headers)


# ── 5. DATEV-Export: Feiertag vs. Urlaub, working_days-Konsistenz ─────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_datev_holiday_takes_precedence_over_vacation(
    client: AsyncClient, db_session: AsyncSession, cov_company, cov_headers, cov_vacation_type_id,
):
    """Absence, die exakt auf einen Feiertag fällt: DATEV-Zeile zeigt 'F' statt 'U',
    und working_days (Antrag) rechnet den Feiertag korrekt heraus."""
    from app.services.holiday_service import ensure_holidays_for_year
    from app.services.report_service import report_service

    from uuid import UUID as _UUID
    company_id = _UUID(cov_company["user"]["company_id"])
    user_id = _UUID(cov_company["user"]["id"])

    r = await client.patch("/api/v1/companies/me", json={"state": "BY"}, headers=cov_headers)
    assert r.status_code == 200, r.text
    r2 = await client.patch(f"/api/v1/users/{user_id}", json={"personnel_number": "4711"}, headers=cov_headers)
    assert r2.status_code == 200, r2.text

    # Feiertage für 2027 vorab befüllen, damit absence_service._get_holiday_dates
    # (fragt PublicHoliday direkt ab, ohne Auto-Generierung) sie kennt.
    # Jahreswechsel-Fall (Feiertag im Folgejahr) siehe separat
    # test_holiday_dates_cover_year_boundary weiter unten -- hier bewusst
    # innerhalb eines einzigen Jahres, um Feiertag-vs-Urlaub-Vorrang isoliert
    # zu prüfen.
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    await ensure_holidays_for_year(2027, "BY", db_session)
    await db_session.commit()

    # Fr 01.01.2027 (Neujahr, Feiertag) - Di 05.01.2027: 3 Wochentage (Fr/Mo/Di),
    # davon Neujahr ein Feiertag -> 2 Werktage
    start = date(2027, 1, 1)
    end = date(2027, 1, 5)
    resp = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start), "end_date": str(end),
    }, headers=cov_headers)
    assert resp.status_code == 201, resp.text
    data = resp.json()
    assert data["working_days"] == 2, "Feiertag (Neujahr) sollte working_days rausrechnen"

    # Selbst genehmigen geht nicht -> zweiter Admin holen wäre Overkill hier;
    # DATEV-Report berücksichtigt PENDING nicht (nur APPROVED/FIRST_APPROVED),
    # daher separat genehmigen.
    resp2 = await client.post("/api/v1/users/invite", json={
        "first_name": "Datev", "last_name": "Approver", "email": "datevapprover@covgaps.de",
        "role": "COMPANY_ADMIN", "initial_password": "Secret123",
    }, headers=cov_headers)
    assert resp2.status_code == 201, resp2.text
    login = await client.post("/api/v1/auth/login", json={
        "email": "datevapprover@covgaps.de", "password": "Secret123",
    })
    approver_headers = {"Authorization": f"Bearer {login.json()['access_token']}"}
    approve = await client.post(f"/api/v1/absences/{data['id']}/approve", headers=approver_headers)
    assert approve.status_code == 200, approve.text

    # RLS-Bypass gilt nur innerhalb der aktuellen Transaktion (SET LOCAL) -- der
    # vorherige Bypass ist mit dem commit() oben verfallen. expire_all() erzwingt
    # frische Queries unter dem neuen Bypass (statt gecachter Identity-Map-Objekte).
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    db_session.expire_all()
    sheet = await report_service.datev_monthly_report(company_id, user_id, 2027, 1, db_session)
    jan1_row = next(row for row in sheet.rows if row.day == 1)
    assert jan1_row.code == "F", f"Feiertag muss Vorrang vor Urlaubs-Kürzel haben, got code={jan1_row.code!r}"
    assert "Neujahr" in (jan1_row.note or "")

    # working_days-Konsistenz: Feiertag zählt nicht als Urlaubstag -> Mo/Di (4./5.1.)
    # müssen als 'U' erscheinen (die tatsächlich abgezogenen Werktage).
    day_codes = {row.day: row.code for row in sheet.rows}
    assert day_codes.get(4) == "U"
    assert day_codes.get(5) == "U"

    await client.patch("/api/v1/companies/me", json={"state": None}, headers=cov_headers)


async def test_holiday_dates_cover_year_boundary(
    client: AsyncClient, db_session: AsyncSession, cov_company, cov_headers, cov_vacation_type_id,
):
    """Regressionstest für den in test_datev_holiday_takes_precedence_over_vacation
    dokumentierten Bug: ein Antrag über den Jahreswechsel muss den Feiertag im
    Folgejahr (Neujahr) korrekt aus working_days herausrechnen. Fix:
    _get_holiday_dates(..., year_to=end_date.year) statt nur start_date.year."""
    from app.services.holiday_service import ensure_holidays_for_year

    r = await client.patch("/api/v1/companies/me", json={"state": "BY"}, headers=cov_headers)
    assert r.status_code == 200, r.text

    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    await ensure_holidays_for_year(2028, "BY", db_session)
    await ensure_holidays_for_year(2029, "BY", db_session)
    await db_session.commit()

    # Do 28.12.2028 - Mo 01.01.2029: Wochentage Do/Fr (28./29.12.) + Mo (01.01.),
    # Sa/So (30./31.12.) fallen ohnehin raus. Montag 01.01.2029 ist Neujahr
    # (Feiertag) -> 2 Werktage.
    start = date(2028, 12, 28)
    end = date(2029, 1, 1)
    resp = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start), "end_date": str(end),
    }, headers=cov_headers)
    assert resp.status_code == 201, resp.text
    data = resp.json()
    assert data["working_days"] == 2, (
        f"Neujahr (Folgejahr) muss aus working_days herausgerechnet werden, "
        f"got {data['working_days']}"
    )

    await client.patch("/api/v1/companies/me", json={"state": None}, headers=cov_headers)


# ── 6. _categorize_hours: Nacht über Mitternacht + Sonntag/Feiertag ─────────────

def test_categorize_hours_night_shift_over_midnight():
    from app.services.report_service import _categorize_hours

    # 22:00 - 06:00, kein Feiertag, kein Sonntag (Mo -> Di)
    entry_date = date(2026, 6, 1)  # Montag
    result = _categorize_hours(entry_date, time(22, 0), time(6, 0), 0, {})

    # 22:00-24:00 (2h) = night_25, 00:00-04:00 (4h) = night_40, 04:00-06:00 (2h) = night_25
    assert result.night_40_hours == pytest.approx(4.0)
    assert result.night_25_hours == pytest.approx(4.0)
    assert result.normal_hours == pytest.approx(0.0)
    total = (result.normal_hours + result.night_25_hours + result.night_40_hours
             + result.sunday_hours + result.holiday_125_hours + result.holiday_150_hours)
    assert total == pytest.approx(8.0)


def test_categorize_hours_sunday_and_holiday_highest_wins():
    from app.services.report_service import _categorize_hours

    # Sonntag, der gleichzeitig Feiertag (hoher Zuschlag) ist -> holiday_150, nicht
    # sunday + holiday addiert.
    sunday_holiday = date(2026, 12, 25)  # 1. Weihnachtsfeiertag, Freitag in 2026 -> nutze echten Sonntag
    # 2026-12-25 ist ein Freitag; für einen Sonntags-Feiertag nehmen wir ein
    # künstliches Beispiel (Datum ist beliebig, Zuschlagslogik hängt nur vom
    # holidays-dict + weekday ab)
    d = date(2026, 11, 1)  # ein Sonntag
    assert d.weekday() == 6
    holidays = {d: ("Fiktiver Feiertag hoch", True)}

    result = _categorize_hours(d, time(10, 0), time(18, 0), 0, holidays)  # 8h, tagsüber

    assert result.holiday_150_hours == pytest.approx(8.0)
    assert result.sunday_hours == pytest.approx(0.0)
    assert result.holiday_125_hours == pytest.approx(0.0)


# ── 7. POST /absences/ gibt warnings zurück (API-Konsistenz-Fix) ────────────────

async def test_create_absence_returns_warnings(
    client: AsyncClient, cov_headers, cov_vacation_type_id,
):
    """Regressionstest: POST /absences/ verwarf bisher die von create_absence()
    zurückgegebenen warnings (im Unterschied zu approve_absence, das sie über
    AbsenceApproveOut.warnings durchreicht). Fix: neues AbsenceCreateOut-Schema.
    Trigger hier: Überschneidungs-Warnung bei zwei sich überlappenden Anträgen."""
    start = _future_monday(20)
    end = start + timedelta(days=2)

    resp1 = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start), "end_date": str(end),
    }, headers=cov_headers)
    assert resp1.status_code == 201, resp1.text
    assert resp1.json().get("warnings") == []

    # Überlappender zweiter Antrag im selben Zeitraum -> Warnung erwartet
    resp2 = await client.post("/api/v1/absences/", json={
        "type_id": str(cov_vacation_type_id), "start_date": str(start), "end_date": str(end + timedelta(days=1)),
    }, headers=cov_headers)
    assert resp2.status_code == 201, resp2.text
    data2 = resp2.json()
    assert "warnings" in data2
    assert any("Überschneidung" in w for w in data2["warnings"]), data2["warnings"]


# ── P1-Rest: postgres-expert Audit, 4 gezielte Tests ────────────────────────────
#
# 1. Scheduler-Idempotenz Überstunden-Verfall (report_service.apply_overtime_expiry_if_needed)
# 2. Bradford-Faktor Fensterrand + MANAGER-Department-Scope (absence_service.get_sick_stats)
# 3. Sondervertretungs-Faktor + Überstunden-Kappung kombiniert (report_service._recalculate_overtime_balance)
# 4. Cross-Company-RLS bei Vertretern (absence_service.list_absences as_substitute=True)

from types import SimpleNamespace


# ── 1. Scheduler-Idempotenz (Überstunden-Verfall) ───────────────────────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_overtime_expiry_idempotent_second_call_noop(db_session: AsyncSession):
    """Ein zweiter Aufruf am selben Tag darf keinen zweiten Abzug auslösen
    (verhindert Doppel-Kappung bei doppeltem Scheduler-Lauf)."""
    from app.services.report_service import apply_overtime_expiry_if_needed

    company = SimpleNamespace(
        overtime_expiry_enabled=True,
        overtime_expiry_month=1,
        overtime_expiry_day=1,
        overtime_max_carryover_hours=5,
    )
    bal = SimpleNamespace(total_hours=Decimal("20"), taken_hours=Decimal("0"), last_expiry_applied_at=None)

    changed_first = await apply_overtime_expiry_if_needed(bal, company, db_session)
    assert changed_first is True
    assert bal.total_hours == Decimal("5"), "Kappung auf max_carryover_hours erwartet"

    # Zweiter Aufruf am selben Tag (last_expiry_applied_at wurde eben gesetzt) -> no-op
    changed_second = await apply_overtime_expiry_if_needed(bal, company, db_session)
    assert changed_second is False
    assert bal.total_hours == Decimal("5"), "Kein zweiter Abzug bei erneutem Aufruf am selben Tag"


@pytest.mark.asyncio(loop_scope="session")
async def test_overtime_expiry_invalid_date_silent_skip(db_session: AsyncSession):
    """Ungültiges Verfallsdatum (29. Februar in einem Nicht-Schaltjahr) darf nicht
    crashen (kein propagierter ValueError), sondern still übersprungen werden.

    Hinweis: nutzt das reale Systemdatum (Funktion ruft intern date.today() auf,
    kein Mock-Punkt vorhanden). Das aktuelle Jahr (Testlauf 2026) ist kein
    Schaltjahr, ebenso das Vorjahr 2025 -- Test bleibt bis 2027 gültig
    (2028 ist wieder ein Schaltjahr, dann müsste der Test angepasst werden)."""
    from app.services.report_service import apply_overtime_expiry_if_needed

    company = SimpleNamespace(
        overtime_expiry_enabled=True,
        overtime_expiry_month=2,
        overtime_expiry_day=29,
        overtime_max_carryover_hours=5,
    )
    bal = SimpleNamespace(total_hours=Decimal("20"), taken_hours=Decimal("0"), last_expiry_applied_at=None)

    changed = await apply_overtime_expiry_if_needed(bal, company, db_session)
    assert changed is False
    assert bal.total_hours == Decimal("20"), "Bei ungültigem Datum darf keine Kappung stattfinden"


# ── 2. Bradford-Faktor Fensterrand + MANAGER-Department-Scope ──────────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_bradford_window_boundary_and_manager_department_scope(
    client: AsyncClient, db_session: AsyncSession,
):
    """(a) Eine Krank-Episode die GENAU am Rand des 365-Tage-Fensters endet
    (end_date == window_start) muss mitgezählt werden, eine die einen Tag davor
    endet nicht (Off-by-one, quadratisch empfindlich für den Bradford-Faktor).
    (b) Ein MANAGER mit department_id sieht nur Bradford-Stats des eigenen
    Departments, nicht firmenweit (DSGVO-Datensparsamkeit)."""
    from app.models.absence import Absence, AbsenceStatus
    from app.models.absence_type import AbsenceCategory, AbsenceType
    from app.models.user import UserRole
    from app.services.absence_service import absence_service

    reg = await client.post("/api/v1/auth/register", json={
        "company_name": "Bradford Edge GmbH",
        "first_name": "Brad", "last_name": "Admin",
        "email": "admin@bradfordedge.de", "password": "Secret123",
    })
    assert reg.status_code == 201, reg.text
    admin_headers = {"Authorization": f"Bearer {reg.json()['access_token']}"}
    me = (await client.get("/api/v1/auth/me", headers=admin_headers)).json()
    company_id = me["company_id"]

    dep_a = await client.post("/api/v1/companies/me/departments", json={"name": "Abteilung A"}, headers=admin_headers)
    assert dep_a.status_code == 201, dep_a.text
    dep_b = await client.post("/api/v1/companies/me/departments", json={"name": "Abteilung B"}, headers=admin_headers)
    assert dep_b.status_code == 201, dep_b.text
    dep_a_id = dep_a.json()["id"]
    dep_b_id = dep_b.json()["id"]

    mgr = await client.post("/api/v1/users/invite", json={
        "first_name": "Manager", "last_name": "A", "email": "manager.a@bradfordedge.de",
        "role": "MANAGER", "department_id": dep_a_id, "initial_password": "Secret123",
    }, headers=admin_headers)
    assert mgr.status_code == 201, mgr.text
    mgr_id = mgr.json()["id"]

    emp_a = await client.post("/api/v1/users/invite", json={
        "first_name": "Emp", "last_name": "A", "email": "emp.a@bradfordedge.de",
        "role": "EMPLOYEE", "department_id": dep_a_id, "initial_password": "Secret123",
    }, headers=admin_headers)
    assert emp_a.status_code == 201, emp_a.text
    emp_a_id = emp_a.json()["id"]

    emp_b = await client.post("/api/v1/users/invite", json={
        "first_name": "Emp", "last_name": "B", "email": "emp.b@bradfordedge.de",
        "role": "EMPLOYEE", "department_id": dep_b_id, "initial_password": "Secret123",
    }, headers=admin_headers)
    assert emp_b.status_code == 201, emp_b.text
    emp_b_id = emp_b.json()["id"]

    types_resp = await client.get("/api/v1/absence-types/", headers=admin_headers)
    sick_type = next(t for t in types_resp.json() if t["category"] == "sick")
    sick_type_id = sick_type["id"]

    ref_date = date.today()
    window_start = ref_date - timedelta(days=365)

    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))

    # Emp A: eine Episode GENAU am Fensterrand (muss zählen) + eine Episode
    # einen Tag VOR dem Fensterrand (darf nicht zählen).
    in_window = Absence(
        user_id=emp_a_id, type_id=sick_type_id,
        start_date=window_start, end_date=window_start,
        working_days=Decimal("1"), status=AbsenceStatus.APPROVED,
    )
    out_of_window = Absence(
        user_id=emp_a_id, type_id=sick_type_id,
        start_date=window_start - timedelta(days=3), end_date=window_start - timedelta(days=1),
        working_days=Decimal("3"), status=AbsenceStatus.APPROVED,
    )
    # Emp B (anderes Department): Episode innerhalb des Fensters.
    emp_b_absence = Absence(
        user_id=emp_b_id, type_id=sick_type_id,
        start_date=ref_date - timedelta(days=10), end_date=ref_date - timedelta(days=8),
        working_days=Decimal("3"), status=AbsenceStatus.APPROVED,
    )
    db_session.add_all([in_window, out_of_window, emp_b_absence])
    await db_session.flush()

    # (a) Fensterrand: Admin sieht firmenweit -> Emp A hat genau 1 Episode
    # (nur die am Rand liegende zählt, die davor liegende fällt raus).
    admin_ns = SimpleNamespace(role=UserRole.COMPANY_ADMIN, department_id=None)
    stats_admin = await absence_service.get_sick_stats(
        UUID(company_id) if not isinstance(company_id, UUID) else company_id,
        admin_ns, ref_date, db_session,
    )
    emp_a_stats = next(s for s in stats_admin if str(s["user_id"]) == emp_a_id)
    assert emp_a_stats["episodes"] == 1, (
        f"Off-by-one am Fensterrand: erwartet genau 1 Episode (window_start selbst zählt, "
        f"ein Tag davor nicht), got {emp_a_stats['episodes']}"
    )
    assert emp_a_stats["total_days"] == pytest.approx(1.0)
    # Admin sieht auch Emp B (firmenweit, keine Department-Einschränkung).
    assert any(str(s["user_id"]) == emp_b_id for s in stats_admin)

    # (b) MANAGER mit department_id sieht nur sein eigenes Department.
    mgr_ns = SimpleNamespace(role=UserRole.MANAGER, department_id=UUID(dep_a_id))
    stats_mgr = await absence_service.get_sick_stats(
        UUID(company_id) if not isinstance(company_id, UUID) else company_id,
        mgr_ns, ref_date, db_session,
    )
    assert any(str(s["user_id"]) == emp_a_id for s in stats_mgr), "Manager muss eigenes Department sehen"
    assert not any(str(s["user_id"]) == emp_b_id for s in stats_mgr), (
        "DSGVO-Lücke: MANAGER sieht Bradford-Stats von Mitarbeitern außerhalb des eigenen Departments"
    )

    await db_session.commit()


# ── 3. Sondervertretungs-Faktor + Überstunden-Kappung kombiniert ───────────────

@pytest.mark.asyncio(loop_scope="session")
async def test_special_assignment_factor_combined_with_overtime_cap(
    client: AsyncClient, db_session: AsyncSession,
):
    """SpecialAssignment mit Faktor 1.5 (mode=fza) UND company.overtime_cap_hours
    gleichzeitig: prüft anhand des tatsächlichen Codes, dass die Kappung auf den
    FAKTORISIERTEN Wert angewendet wird (hours_by_date wird bereits mit dem Faktor
    multipliziert berechnet, siehe report_service.py Zeile ~239, bevor die Kappung
    in Zeile ~253 greift). Das ist kein Bug, sondern dokumentiertes Ist-Verhalten:
    die Sondervertretungs-Gewichtung fließt VOR die Kappung ein."""
    from app.models.special_assignment import AssignmentMode, SpecialAssignment
    from app.models.time_entry import EntryStatus, TimeEntry
    from app.services.report_service import _recalculate_overtime_balance

    reg = await client.post("/api/v1/auth/register", json={
        "company_name": "FZA Cap Edge GmbH",
        "first_name": "Cap", "last_name": "Admin",
        "email": "admin@fzacapedge.de", "password": "Secret123",
    })
    assert reg.status_code == 201, reg.text
    admin_headers = {"Authorization": f"Bearer {reg.json()['access_token']}"}

    cap_resp = await client.patch("/api/v1/companies/me", json={"overtime_cap_hours": 5}, headers=admin_headers)
    assert cap_resp.status_code == 200, cap_resp.text

    emp = await client.post("/api/v1/users/invite", json={
        "first_name": "Cap", "last_name": "Employee", "email": "emp@fzacapedge.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=admin_headers)
    assert emp.status_code == 201, emp.text
    emp_id = emp.json()["id"]

    # Vergangener Werktag (Mo-Fr), weit genug in der Vergangenheit um nicht
    # mit anderen Tests zu kollidieren.
    workday = date.today() - timedelta(days=90)
    while workday.weekday() >= 5:
        workday -= timedelta(days=1)

    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))

    user = await db_session.get(User, emp_id)
    assert user is not None

    # 10h gearbeitet (08:00-18:00, keine Pause) an einem Standard-8h-Tag.
    entry = TimeEntry(
        user_id=emp_id, date=workday,
        start_time=time(8, 0), end_time=time(18, 0),
        break_minutes=0, status=EntryStatus.APPROVED,
    )
    db_session.add(entry)

    sa = SpecialAssignment(
        user_id=emp_id, company_id=UUID(str(user.company_id)),
        date_from=workday, date_to=workday,
        factor=Decimal("1.5"), mode=AssignmentMode.fza,
        label="Nachtschicht-Zuschlag Test",
    )
    db_session.add(sa)
    await db_session.flush()

    bal = await _recalculate_overtime_balance(user, None, db_session)

    # Roh: 10h - 8h Soll = 2h Überstunden. Faktorisiert: 10h*1.5=15h - 8h = 7h.
    # Kappung bei 5h -> falls auf faktorisierten Wert angewendet (Ist-Verhalten
    # laut Code) landet total_hours bei genau 5, NICHT bei den rohen 2h.
    assert bal.total_hours == Decimal("5"), (
        f"Erwartetes Ist-Verhalten (Kappung greift auf den faktorisierten Wert, "
        f"nicht auf den rohen ArbZG-Überstundenwert) nicht bestätigt, got {bal.total_hours}. "
        f"Falls dies fachlich falsch ist, ist das ein Bug in report_service._recalculate_overtime_balance "
        f"(Kappung sollte ggf. vor Faktor-Anwendung greifen) -- hier nur dokumentiert, nicht gefixt."
    )

    await db_session.commit()


# ── 4. Cross-Company-RLS bei Vertretern (list_absences as_substitute) ──────────

@pytest.mark.asyncio(loop_scope="session")
async def test_list_absences_as_substitute_cross_company_isolation(
    client: AsyncClient, db_session: AsyncSession,
):
    """Defense-in-Depth auf Service-Ebene (nicht nur Erstell-Validierung):
    list_absences(..., as_substitute=True) mit company_id von Firma B darf
    KEINE Anträge aus Firma A liefern, selbst wenn (hypothetisch, z.B. durch
    einen Bug anderswo) ein User aus Firma A als substitute_id auf einen
    Antrag aus Firma A gesetzt wurde und ein gleich benannter/verwechselter
    User aus Firma B abgefragt wird. Der Query joint über User.company_id ==
    company_id -- geprüft wird hier direkt auf Service-Ebene."""
    from app.models.absence import Absence, AbsenceStatus
    from app.models.user import UserRole
    from app.services.absence_service import absence_service

    # Firma A: Antragsteller + Vertreter, beide in Firma A.
    reg_a = await client.post("/api/v1/auth/register", json={
        "company_name": "Substitute Cross A GmbH",
        "first_name": "SubA", "last_name": "Admin",
        "email": "admin@subcrossa.de", "password": "Secret123",
    })
    assert reg_a.status_code == 201, reg_a.text
    headers_a = {"Authorization": f"Bearer {reg_a.json()['access_token']}"}
    me_a = (await client.get("/api/v1/auth/me", headers=headers_a)).json()
    company_a_id = me_a["company_id"]

    sub_a = await client.post("/api/v1/users/invite", json={
        "first_name": "Sub", "last_name": "InA", "email": "sub.a@subcrossa.de",
        "role": "EMPLOYEE", "initial_password": "Secret123",
    }, headers=headers_a)
    assert sub_a.status_code == 201, sub_a.text
    sub_a_id = sub_a.json()["id"]

    types_a = await client.get("/api/v1/absence-types/", headers=headers_a)
    vac_type_a = next(t for t in types_a.json() if t["name"] == "Urlaub")["id"]

    # Firma B: eigener Vertreter-Kandidat (gleiche Rolle, andere Firma).
    reg_b = await client.post("/api/v1/auth/register", json={
        "company_name": "Substitute Cross B GmbH",
        "first_name": "SubB", "last_name": "Admin",
        "email": "admin@subcrossb.de", "password": "Secret123",
    })
    assert reg_b.status_code == 201, reg_b.text
    headers_b = {"Authorization": f"Bearer {reg_b.json()['access_token']}"}
    me_b = (await client.get("/api/v1/auth/me", headers=headers_b)).json()
    company_b_id = me_b["company_id"]

    # Absence in Firma A, mit sub_a (Firma A) als Vertretung -- über Model
    # direkt angelegt um Overlap-/Validierungslogik des Create-Endpunkts zu
    # umgehen und die Service-Query isoliert zu prüfen.
    start = date.today() + timedelta(days=200)
    await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
    absence_a = Absence(
        user_id=me_a["id"], type_id=vac_type_a,
        start_date=start, end_date=start,
        working_days=Decimal("1"), status=AbsenceStatus.PENDING,
        substitute_id=sub_a_id,
    )
    db_session.add(absence_a)
    await db_session.flush()

    # Query 1: sub_a (echter Vertreter, Firma A) mit company_id=Firma A -> sieht den Antrag.
    sub_a_user = await db_session.get(User, sub_a_id)
    total_a, absences_a = await absence_service.list_absences(
        UUID(company_a_id), sub_a_user, db_session, as_substitute=True,
    )
    assert total_a == 1
    assert absences_a[0].id == absence_a.id

    # Query 2: Admin von Firma B (current_user.id != sub_a_id, aber selbst wenn
    # ein Aufrufer aus Firma B versucht mit fremder company_id_a zu fragen, ist
    # der Sicherheits-Layer die current_user.id im WHERE Absence.substitute_id
    # == current_user.id -- entscheidend ist aber der company_id-Join). Hier:
    # regulärer Aufruf mit Firma B's eigener company_id liefert 0 Treffer,
    # obwohl der Antrag aus Firma A existiert.
    admin_b_user = await db_session.get(User, me_b["id"])
    # current_user hier künstlich auf sub_a_id "verfälscht" simulieren ist nicht
    # möglich ohne DB-User -- stattdessen der eigentliche Defense-in-Depth-Check:
    # company_id explizit auf Firma B gesetzt, obwohl ein (hypothetischer)
    # Vertreter mit gleicher User-Id in Firma A existieren würde. Da UUIDs global
    # eindeutig sind, testen wir die Kernaussage direkt: Query mit company_b_id
    # + sub_a_user (der in Wahrheit zu Firma A gehört, aber hier als current_user
    # übergeben wird) liefert trotzdem 0 Treffer, weil der Join User.company_id
    # == company_id (Firma B) den Antrag aus Firma A aussortiert.
    total_cross, absences_cross = await absence_service.list_absences(
        UUID(company_b_id), sub_a_user, db_session, as_substitute=True,
    )
    assert total_cross == 0, (
        "RLS-Lücke: Vertreter-Query lieferte einen Antrag aus einer anderen Firma zurück "
        "(company_id-Join in list_absences greift nicht als Defense-in-Depth)"
    )
    assert absences_cross == []

    await db_session.commit()