scripte/timemaster
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Freizeitausgleich-Lücken geschlossen (Gap 1-3) + konfigurierbare Schwellwerte

Browse Source 
Gap-1: Überziehschutz für Überstundenkonto
  - Company.overtime_overdraft_allowed (default: true) – blockiert FZA wenn deaktiviert
  - Company.overtime_warning_threshold_hours (default: 0) – Warnung wenn Konto unter Schwelle fällt
  - warnings[] jetzt in approve_absence Response (AbsenceApproveOut)
  - Migration 0028_overtime_fza_config.py

Gap-2: total_hours wird bei Zeiteintrag-Genehmigung neu berechnet
  - time_service.approve_entry() ruft _recalculate_overtime_balance() auf
  - last_calculated Timestamp wird gesetzt

Gap-3: Stornierung genehmigter FZA-Anträge bucht taken_hours zurück
  - _refund_overtime() Helfer hinzugefügt
  - cancel_absence() erlaubt jetzt HR/Admin auch genehmigte Abwesenheiten zu stornieren
  - DELETE /absences/{id} gibt jetzt AbsenceOut zurück (statt 204)
  - Mitarbeiter können genehmigte FZA-Anträge nicht selbst stornieren (409)

Frontend:
  - CompanySettingsPage: neuer Abschnitt 'Freizeitausgleich' mit Toggle + Schwellwert-Eingabe

Tests: backend/tests/test_fza.py mit 6 Tests (alle 3 Gaps)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
patrick
2026-05-25 00:08:03 +02:00
parent 0ba16bb6af
commit 345002944e
9 changed files with 619 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_fza.py
+382
View File
@@ -0,0 +1,382 @@
"""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
await db_session.refresh(ob)
taken_after_cancel = float(ob.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()
hr_me = (await client.get("/api/v1/auth/me", headers=fza_hr_headers)).json()
# Eintrag manuell anlegen (10h, 2h Überstunden bei 8h Soll)
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)
# Kann 201 oder 200 sein je nach Implementation
assert resp.status_code in (200, 201), resp.text
entry_id = resp.json()["id"]
# Genehmigen (HR != EMPLOYEE, Self-Approval ist geblockt)
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"])
)
if ob:
await db_session.refresh(ob)
assert ob.last_calculated is not None, "last_calculated sollte nach Genehmigung gesetzt sein"