sysops
1fedd683e0
Stand: agent-06 (Audit-Log), agent-05 (Krankmeldung), agent-07 Phase 1 (Personalnummer), Busylight-Pull-Integration, TOTP/2FA, Abwesenheiten, Zeiterfassung, Kiosk-Grundgerüst. Migrations 0001–0023 deployed auf 192.168.1.137 + .164. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
463 lines
16 KiB
Python
463 lines
16 KiB
Python
"""Tests für agent-03-abwesenheit"""
|
||
import pytest
|
||
import pytest_asyncio
|
||
from datetime import date, timedelta
|
||
from httpx import AsyncClient
|
||
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def abs_company(client: AsyncClient):
|
||
"""Eigene Company für Absence-Tests."""
|
||
resp = await client.post("/api/v1/auth/register", json={
|
||
"company_name": "Absence AG",
|
||
"first_name": "Admin",
|
||
"last_name": "Test",
|
||
"email": "admin@absenceag.de",
|
||
"password": "Secret123",
|
||
})
|
||
assert resp.status_code == 201
|
||
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 abs_headers(abs_company):
|
||
return {"Authorization": f"Bearer {abs_company['tokens']['access_token']}"}
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def vacation_type_id(client: AsyncClient, abs_headers):
|
||
"""Urlaubs-Typ der Company holen."""
|
||
resp = await client.get("/api/v1/absence-types/", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
types = resp.json()
|
||
assert len(types) > 0, "Default-Abwesenheitstypen fehlen"
|
||
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 sick_type_id(client: AsyncClient, abs_headers):
|
||
resp = await client.get("/api/v1/absence-types/", headers=abs_headers)
|
||
types = resp.json()
|
||
sick = next((t for t in types if t["name"] == "Krankheit"), types[0])
|
||
return sick["id"]
|
||
|
||
|
||
# ── Default Types ──────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_absence_types_created_on_register(client: AsyncClient, abs_headers):
|
||
"""Nach der Registrierung müssen Default-Typen vorhanden sein."""
|
||
resp = await client.get("/api/v1/absence-types/", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
types = resp.json()
|
||
names = {t["name"] for t in types}
|
||
assert "Urlaub" in names
|
||
assert "Krankheit" in names
|
||
assert "Homeoffice" in names
|
||
assert "Dienstreise" in names
|
||
assert "Sonderurlaub" in names
|
||
|
||
|
||
# ── Absence Types CRUD ─────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_absence_type(client: AsyncClient, abs_headers):
|
||
resp = await client.post(
|
||
"/api/v1/absence-types/",
|
||
json={
|
||
"name": "Elternzeit",
|
||
"color": "#F59E0B",
|
||
"requires_approval": True,
|
||
"deducts_vacation": False,
|
||
"is_paid": False,
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
assert resp.json()["name"] == "Elternzeit"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_absence_type(client: AsyncClient, abs_headers, vacation_type_id):
|
||
resp = await client.patch(
|
||
f"/api/v1/absence-types/{vacation_type_id}",
|
||
json={"max_days_per_year": 30},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["max_days_per_year"] == 30
|
||
|
||
|
||
# ── Absences ───────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_absence_vacation(client: AsyncClient, abs_headers, vacation_type_id):
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()))
|
||
resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=4)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert data["status"] == "pending"
|
||
assert data["working_days"] > 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_absence_sick_auto_approved(client: AsyncClient, abs_headers, sick_type_id):
|
||
"""Krankheit hat requires_approval=False → sofort approved."""
|
||
resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(sick_type_id),
|
||
"start_date": str(date.today()),
|
||
"end_date": str(date.today()),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
# Krankheit erfordert keine Genehmigung → approved
|
||
data = resp.json()
|
||
assert data["status"] == "approved"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_absences(client: AsyncClient, abs_headers):
|
||
resp = await client.get("/api/v1/absences/", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "total" in data
|
||
assert "items" in data
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_absence_by_id(client: AsyncClient, abs_headers, vacation_type_id):
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()) + 7)
|
||
create_resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=2)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create_resp.json()["id"]
|
||
|
||
resp = await client.get(f"/api/v1/absences/{absence_id}", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["id"] == absence_id
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_absence(client: AsyncClient, abs_headers, vacation_type_id):
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()) + 14)
|
||
create_resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=4)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create_resp.json()["id"]
|
||
|
||
resp = await client.post(f"/api/v1/absences/{absence_id}/approve", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "approved"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reject_absence(client: AsyncClient, abs_headers, vacation_type_id):
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()) + 21)
|
||
create_resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=4)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create_resp.json()["id"]
|
||
|
||
resp = await client.post(
|
||
f"/api/v1/absences/{absence_id}/reject",
|
||
json={"rejection_reason": "Urlaubssperre in diesem Zeitraum"},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "rejected"
|
||
assert resp.json()["rejection_reason"] == "Urlaubssperre in diesem Zeitraum"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancel_absence(client: AsyncClient, abs_headers, vacation_type_id):
|
||
"""Eigenen ausstehenden Antrag stornieren."""
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()) + 28)
|
||
create_resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=4)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create_resp.json()["id"]
|
||
|
||
resp = await client.delete(f"/api/v1/absences/{absence_id}", headers=abs_headers)
|
||
assert resp.status_code == 204
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_calendar(client: AsyncClient, abs_headers):
|
||
resp = await client.get(
|
||
"/api/v1/absences/calendar",
|
||
params={"year": 2026, "month": 4},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json(), list)
|
||
|
||
|
||
# ── Vacation Balance ───────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_own_balance(client: AsyncClient, abs_headers):
|
||
resp = await client.get("/api/v1/absences/balance", params={"year": 2026}, headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "entitled_days" in data
|
||
assert "remaining_days" in data
|
||
assert data["entitled_days"] == 30
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_balance_deducted_after_approve(client: AsyncClient, abs_headers, vacation_type_id, abs_company):
|
||
"""Urlaubskonto wird nach Genehmigung abgezogen."""
|
||
user_id = abs_company["user"]["id"]
|
||
|
||
# Initiales Konto
|
||
before = await client.get("/api/v1/absences/balance", params={"year": 2026}, headers=abs_headers)
|
||
used_before = before.json()["used_days"]
|
||
|
||
# Urlaub beantragen (Mo-Fr = 5 Tage)
|
||
next_monday = date.today() + timedelta(days=(7 - date.today().weekday()) + 35)
|
||
create_resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(vacation_type_id),
|
||
"start_date": str(next_monday),
|
||
"end_date": str(next_monday + timedelta(days=4)),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create_resp.json()["id"]
|
||
working_days = create_resp.json()["working_days"]
|
||
|
||
# Genehmigen
|
||
await client.post(f"/api/v1/absences/{absence_id}/approve", headers=abs_headers)
|
||
|
||
# Konto prüfen
|
||
after = await client.get("/api/v1/absences/balance", params={"year": 2026}, headers=abs_headers)
|
||
assert after.json()["used_days"] == used_before + working_days
|
||
|
||
|
||
# ── Public Holidays ────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_and_list_public_holiday(client: AsyncClient, abs_headers):
|
||
resp = await client.post(
|
||
"/api/v1/public-holidays/",
|
||
json={
|
||
"country": "DE",
|
||
"state": None,
|
||
"date": "2026-01-01",
|
||
"name": "Neujahr",
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
assert resp.json()["name"] == "Neujahr"
|
||
|
||
list_resp = await client.get(
|
||
"/api/v1/public-holidays/",
|
||
params={"year": 2026, "country": "DE"},
|
||
headers=abs_headers,
|
||
)
|
||
assert list_resp.status_code == 200
|
||
holidays = list_resp.json()
|
||
assert any(h["name"] == "Neujahr" for h in holidays)
|
||
|
||
|
||
# ── Working Days Calculation ───────────────────────────────────────────────────
|
||
|
||
def test_calc_working_days_full_week():
|
||
from app.services.absence_service import AbsenceService
|
||
svc = AbsenceService()
|
||
# Mo-Fr
|
||
days = svc._calc_working_days(
|
||
date(2026, 3, 30), date(2026, 4, 3), set(), False, False
|
||
)
|
||
assert days == 5.0
|
||
|
||
|
||
def test_calc_working_days_with_holiday():
|
||
from app.services.absence_service import AbsenceService
|
||
svc = AbsenceService()
|
||
# Mo-Fr mit Feiertag am Mittwoch
|
||
days = svc._calc_working_days(
|
||
date(2026, 3, 30), date(2026, 4, 3),
|
||
{date(2026, 4, 1)}, # Mittwoch = Feiertag
|
||
False, False
|
||
)
|
||
assert days == 4.0
|
||
|
||
|
||
def test_calc_working_days_half_day():
|
||
from app.services.absence_service import AbsenceService
|
||
svc = AbsenceService()
|
||
# Mo-Fr, halber Montag
|
||
days = svc._calc_working_days(
|
||
date(2026, 3, 30), date(2026, 4, 3), set(), True, False
|
||
)
|
||
assert days == 4.5
|
||
|
||
|
||
def test_calc_working_days_weekend_only():
|
||
from app.services.absence_service import AbsenceService
|
||
svc = AbsenceService()
|
||
days = svc._calc_working_days(
|
||
date(2026, 3, 28), date(2026, 3, 29), set(), False, False
|
||
)
|
||
assert days == 0.0
|
||
|
||
|
||
# ── Krankmeldung (agent-05) ───────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_quick_sick_creates_approved_absence(client: AsyncClient, abs_headers):
|
||
"""POST /absences/quick-sick → sofort approved, ohne Typ-Auswahl."""
|
||
today = date.today()
|
||
resp = await client.post(
|
||
"/api/v1/absences/quick-sick",
|
||
json={"start_date": str(today), "end_date": str(today)},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert data["status"] == "approved"
|
||
assert data["certificate_required_by"] is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_certificate_required_by_uses_type_threshold(
|
||
client: AsyncClient, abs_headers, sick_type_id
|
||
):
|
||
"""certificate_required_by = start_date + AbsenceType.certificate_after_days (default 3)."""
|
||
start = date(2027, 6, 1)
|
||
resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(sick_type_id),
|
||
"start_date": str(start),
|
||
"end_date": str(start),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
expected = start + timedelta(days=3)
|
||
assert data["certificate_required_by"] == str(expected)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_certificate_required_by_respects_type_override(
|
||
client: AsyncClient, abs_headers, sick_type_id
|
||
):
|
||
"""PATCH AbsenceType.certificate_after_days = 7 → neue Absences nutzen 7."""
|
||
upd = await client.patch(
|
||
f"/api/v1/absence-types/{sick_type_id}",
|
||
json={"certificate_after_days": 7},
|
||
headers=abs_headers,
|
||
)
|
||
assert upd.status_code == 200
|
||
assert upd.json()["certificate_after_days"] == 7
|
||
|
||
start = date(2027, 7, 5)
|
||
resp = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(sick_type_id),
|
||
"start_date": str(start),
|
||
"end_date": str(start),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
expected = start + timedelta(days=7)
|
||
assert resp.json()["certificate_required_by"] == str(expected)
|
||
|
||
# Reset auf default für nachfolgende Tests
|
||
await client.patch(
|
||
f"/api/v1/absence-types/{sick_type_id}",
|
||
json={"certificate_after_days": 3},
|
||
headers=abs_headers,
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_mark_certificate_received(client: AsyncClient, abs_headers, sick_type_id):
|
||
"""HR/Admin markiert Attest-Eingang per PATCH /absences/{id}/certificate."""
|
||
start = date(2027, 8, 10)
|
||
create = await client.post(
|
||
"/api/v1/absences/",
|
||
json={
|
||
"type_id": str(sick_type_id),
|
||
"start_date": str(start),
|
||
"end_date": str(start),
|
||
},
|
||
headers=abs_headers,
|
||
)
|
||
absence_id = create.json()["id"]
|
||
assert create.json()["certificate_received_at"] is None
|
||
|
||
resp = await client.patch(
|
||
f"/api/v1/absences/{absence_id}/certificate",
|
||
json={}, # default = today
|
||
headers=abs_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["certificate_received_at"] == str(date.today())
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sick_stats_bradford_factor(client: AsyncClient, abs_headers):
|
||
"""GET /absences/sick-stats: Bradford = episodes² × total_days, rolling 12 Monate."""
|
||
resp = await client.get("/api/v1/absences/sick-stats", headers=abs_headers)
|
||
assert resp.status_code == 200
|
||
stats = resp.json()
|
||
# Vorherige Tests haben mindestens eine Krankmeldung im 12-Monats-Fenster erzeugt
|
||
assert len(stats) >= 1
|
||
row = stats[0]
|
||
assert row["episodes"] >= 1
|
||
assert row["total_days"] >= 0.0
|
||
# Bradford-Formel verifizieren
|
||
expected = float(row["episodes"]) ** 2 * row["total_days"]
|
||
assert abs(row["bradford_factor"] - expected) < 0.001
|