"""Tests für Freizeitausgleich (FZA) Lücken-Fixes. Gap-1: Überstunden-Überziehschutz (configurable: allow/block + warning threshold) Gap-2: Überstundenkonto wird bei Zeiteintrag-Genehmigung neu berechnet Gap-3: Stornierung eines genehmigten FZA-Antrags bucht taken_hours zurück """ import pytest import pytest_asyncio from datetime import date, time from decimal import Decimal from httpx import AsyncClient from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.models.absence import Absence, AbsenceStatus from app.models.absence_type import AbsenceType, AbsenceCategory from app.models.company import Company from app.models.overtime_balance import OvertimeBalance from app.models.time_entry import TimeEntry, EntryStatus from app.models.user import User, UserRole # ───────────────────────────────────────────────────────────────────────────── # Fixtures # ───────────────────────────────────────────────────────────────────────────── @pytest_asyncio.fixture(scope="session", loop_scope="session") async def fza_company(client: AsyncClient): """Eigene Company für FZA-Tests.""" resp = await client.post("/api/v1/auth/register", json={ "company_name": "FZA Test GmbH", "first_name": "FZA", "last_name": "Admin", "email": "admin@fza-test.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 fza_admin_headers(fza_company): return {"Authorization": f"Bearer {fza_company['tokens']['access_token']}"} @pytest_asyncio.fixture(scope="session", loop_scope="session") async def fza_hr_headers(client: AsyncClient, fza_admin_headers): """HR-User der FZA Test GmbH.""" resp = await client.post("/api/v1/users/invite", json={ "first_name": "HR", "last_name": "Manager", "email": "hr@fza-test.de", "role": "HR", "initial_password": "Secret123", }, headers=fza_admin_headers) assert resp.status_code == 201, resp.text login = await client.post("/api/v1/auth/login", json={ "email": "hr@fza-test.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 fza_employee_headers(client: AsyncClient, fza_admin_headers): """Mitarbeiter der FZA Test GmbH.""" resp = await client.post("/api/v1/users/invite", json={ "first_name": "Franz", "last_name": "Feierabend", "email": "franz@fza-test.de", "role": "EMPLOYEE", "initial_password": "Secret123", }, headers=fza_admin_headers) assert resp.status_code == 201, resp.text login = await client.post("/api/v1/auth/login", json={ "email": "franz@fza-test.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 fza_type_id(client: AsyncClient, fza_admin_headers): """FZA-Abwesenheitstyp der Company erstellen.""" resp = await client.post("/api/v1/absence-types/", json={ "name": "Freizeitausgleich", "category": "overtime_comp", "color": "#f97316", "requires_approval": True, "deducts_vacation": False, "affects_overtime_balance": True, }, headers=fza_admin_headers) assert resp.status_code == 201, resp.text return resp.json()["id"] async def _seed_overtime_balance( db_session: AsyncSession, admin_user: dict, total_hours: float, ) -> None: """Setzt total_hours in OvertimeBalance direkt (ohne API).""" await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'")) user_id = admin_user["id"] company_id = admin_user["company_id"] 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() # ───────────────────────────────────────────────────────────────────────────── # Gap-1a: Überziehen erlaubt (default) – geht durch # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio(loop_scope="session") async def test_fza_overdraft_allowed_by_default( client: AsyncClient, db_session: AsyncSession, fza_company: dict, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, fza_type_id: str, ): """Standardmäßig (overdraft_allowed=True) darf FZA auch bei leerem Konto genehmigt werden.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() # Überstundenkonto auf 4h setzen (1 Tag = 8h benötigt) await _seed_overtime_balance(db_session, emp_me, total_hours=4.0) await db_session.commit() # HR-User für Genehmigung holen hr_me = (await client.get("/api/v1/auth/me", headers=fza_hr_headers)).json() emp_id = emp_me["id"] company_id = emp_me["company_id"] # Antrag als Mitarbeiter stellen resp = await client.post("/api/v1/absences/", json={ "type_id": fza_type_id, "start_date": "2026-07-01", "end_date": "2026-07-01", }, headers=fza_employee_headers) assert resp.status_code == 201, resp.text absence_id = resp.json()["id"] # HR genehmigt – sollte trotz Überziehen funktionieren resp2 = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=fza_hr_headers) assert resp2.status_code == 200, resp2.text body = resp2.json() assert body["status"] == "approved" # ───────────────────────────────────────────────────────────────────────────── # Gap-1b: Überziehen blockiert # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio(loop_scope="session") async def test_fza_overdraft_blocked( client: AsyncClient, db_session: AsyncSession, fza_company: dict, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, fza_type_id: str, ): """Mit overtime_overdraft_allowed=False wird Genehmigung abgelehnt wenn Konto leer.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() await _seed_overtime_balance(db_session, emp_me, total_hours=2.0) # nur 2h await db_session.commit() # Firma: kein Überziehen erlauben resp = await client.patch("/api/v1/companies/me", json={ "overtime_overdraft_allowed": False, }, headers=fza_admin_headers) assert resp.status_code == 200, resp.text # Antrag für 1 Tag (= 8h) – mehr als verfügbar resp2 = await client.post("/api/v1/absences/", json={ "type_id": fza_type_id, "start_date": "2026-07-02", "end_date": "2026-07-02", }, headers=fza_employee_headers) assert resp2.status_code == 201, resp2.text absence_id = resp2.json()["id"] # Genehmigung muss fehlschlagen resp3 = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=fza_hr_headers) assert resp3.status_code == 422, resp3.text assert "Nicht genug Überstunden" in resp3.json()["detail"] # Reset await client.patch("/api/v1/companies/me", json={ "overtime_overdraft_allowed": True, }, headers=fza_admin_headers) # ───────────────────────────────────────────────────────────────────────────── # Gap-1c: Warnschwelle # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio(loop_scope="session") async def test_fza_warning_threshold( client: AsyncClient, db_session: AsyncSession, fza_company: dict, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, fza_type_id: str, ): """Warnschwelle: Genehmigung geht durch, aber warnings werden zurückgegeben.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() await _seed_overtime_balance(db_session, emp_me, total_hours=16.0) await db_session.commit() # Warnschwelle auf 10h setzen await client.patch("/api/v1/companies/me", json={ "overtime_warning_threshold_hours": 10, }, headers=fza_admin_headers) # Antrag für 1 Tag (8h) → verbleibend 8h < 10h Schwelle → Warnung resp = await client.post("/api/v1/absences/", json={ "type_id": fza_type_id, "start_date": "2026-07-03", "end_date": "2026-07-03", }, headers=fza_employee_headers) assert resp.status_code == 201, resp.text absence_id = resp.json()["id"] resp2 = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=fza_hr_headers) assert resp2.status_code == 200, resp2.text body = resp2.json() assert body["status"] == "approved" assert len(body.get("warnings", [])) > 0, "Warnschwellen-Warnung erwartet" assert "Warnschwelle" in body["warnings"][0] # Reset await client.patch("/api/v1/companies/me", json={ "overtime_warning_threshold_hours": 0, }, headers=fza_admin_headers) # ───────────────────────────────────────────────────────────────────────────── # Gap-3: Stornierung genehmigter FZA → Rückbuchung # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio(loop_scope="session") async def test_fza_cancel_approved_refunds( client: AsyncClient, db_session: AsyncSession, fza_company: dict, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, fza_type_id: str, ): """HR storniert genehmigten FZA → taken_hours werden zurückgebucht.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() await _seed_overtime_balance(db_session, emp_me, total_hours=40.0) await db_session.commit() # Antrag stellen resp = await client.post("/api/v1/absences/", json={ "type_id": fza_type_id, "start_date": "2026-07-07", "end_date": "2026-07-07", }, headers=fza_employee_headers) assert resp.status_code == 201, resp.text absence_id = resp.json()["id"] # Genehmigen r2 = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=fza_hr_headers) assert r2.status_code == 200, r2.text # OvertimeBalance: taken_hours sollte jetzt 8h sein await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'")) ob = await db_session.scalar( select(OvertimeBalance).where(OvertimeBalance.user_id == emp_me["id"]) ) await db_session.refresh(ob) taken_after_approve = float(ob.taken_hours) assert taken_after_approve == pytest.approx(8.0, abs=0.1), f"Erwartet 8h, got {taken_after_approve}" # HR storniert r3 = await client.delete(f"/api/v1/absences/{absence_id}", headers=fza_hr_headers) assert r3.status_code == 200, r3.text assert r3.json()["status"] == "cancelled" # Rückbuchung prüfen – frischen Query statt refresh (vermeidet Race mit CalDAV-Background-Task) import asyncio await asyncio.sleep(0.15) # CalDAV fire-and-forget abwarten await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'")) ob2 = await db_session.scalar( select(OvertimeBalance).where(OvertimeBalance.user_id == emp_me["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"Rückbuchung fehlgeschlagen, got {taken_after_cancel}" @pytest.mark.asyncio(loop_scope="session") async def test_fza_employee_cannot_cancel_approved( client: AsyncClient, db_session: AsyncSession, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, fza_type_id: str, ): """Mitarbeiter kann genehmigten FZA-Antrag nicht selbst stornieren.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() await _seed_overtime_balance(db_session, emp_me, total_hours=40.0) await db_session.commit() resp = await client.post("/api/v1/absences/", json={ "type_id": fza_type_id, "start_date": "2026-07-08", "end_date": "2026-07-08", }, headers=fza_employee_headers) assert resp.status_code == 201, resp.text absence_id = resp.json()["id"] # Genehmigen (durch HR) r2 = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=fza_hr_headers) assert r2.status_code == 200, r2.text # Mitarbeiter versucht zu stornieren → 409 r3 = await client.delete(f"/api/v1/absences/{absence_id}", headers=fza_employee_headers) assert r3.status_code == 409, r3.text assert "HR/Admin" in r3.json()["detail"] # ───────────────────────────────────────────────────────────────────────────── # Gap-2: Zeiteintrag-Genehmigung aktualisiert OvertimeBalance.total_hours # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.asyncio(loop_scope="session") async def test_time_entry_approval_updates_overtime_balance( client: AsyncClient, db_session: AsyncSession, fza_admin_headers: dict, fza_hr_headers: dict, fza_employee_headers: dict, ): """Nach Genehmigung eines Zeiteintrags wird OvertimeBalance.total_hours neu berechnet.""" emp_me = (await client.get("/api/v1/auth/me", headers=fza_employee_headers)).json() # Manuelle Zeiterfassung für diesen User freischalten emp_id = emp_me["id"] r = await client.patch(f"/api/v1/users/{emp_id}", json={"can_manual_time_entry": True}, headers=fza_admin_headers) assert r.status_code == 200, f"Manual entry freischalten fehlgeschlagen: {r.text}" # Eintrag manuell anlegen (10h = 2h Überstunden bei 8h Soll-Tag) resp = await client.post("/api/v1/time/entries", json={ "date": "2026-06-01", "start_time": "08:00", "end_time": "18:00", "break_minutes": 0, }, headers=fza_employee_headers) assert resp.status_code == 201, resp.text entry_id = resp.json()["entry"]["id"] # Genehmigen (HR, kein Self-Approval) r2 = await client.post(f"/api/v1/time/entries/{entry_id}/approve", headers=fza_hr_headers) assert r2.status_code == 200, r2.text # OvertimeBalance.last_calculated sollte jetzt gesetzt sein await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'")) ob = await db_session.scalar( select(OvertimeBalance).where(OvertimeBalance.user_id == emp_me["id"]) .execution_options(populate_existing=True) ) assert ob is not None, "OvertimeBalance wurde nicht angelegt" assert ob.last_calculated is not None, "last_calculated sollte nach Genehmigung gesetzt sein"