patrick
eb122802b2
1. Krankmeldungen an Wochenenden erlaubt (working_days=0 für SICK)
Medizinisch korrekt: Krankmeldungen können auch Wochenendtage umfassen.
Behebt: test_create_absence_sick_auto_approved, test_quick_sick_*,
test_pull_includes_today_absence_with_category, test_sick_stats_*
2. Self-Approval-Schutz in Tests berücksichtigt
abs_approver_headers / time_approver_headers: zweiter Admin je Company.
Behebt: test_approve_absence, test_balance_deducted_after_approve, test_approve_entry
3. test_export_invalid_format: "pdf" → "xml" (pdf ist jetzt valides Format)
Ergebnis: 134/134 passed.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""Tests für agent-02-zeiterfassung"""
|
||
import pytest
|
||
import pytest_asyncio
|
||
from datetime import date, time
|
||
from httpx import AsyncClient
|
||
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def auth_headers(registered_user):
|
||
token = registered_user["tokens"]["access_token"]
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def manager_headers(client: AsyncClient):
|
||
"""Zweiten User als Manager registrieren."""
|
||
resp = await client.post("/api/v1/auth/register", json={
|
||
"company_name": "Time GmbH",
|
||
"first_name": "Manager",
|
||
"last_name": "Max",
|
||
"email": "manager@timegmbh.de",
|
||
"password": "Secret123",
|
||
})
|
||
assert resp.status_code == 201
|
||
token = resp.json()["access_token"]
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def time_approver_headers(client: AsyncClient, manager_headers):
|
||
"""Zweiter Admin in Time GmbH – kann Zeiteinträge anderer genehmigen."""
|
||
resp = await client.post("/api/v1/users/invite", json={
|
||
"first_name": "Time",
|
||
"last_name": "Approver",
|
||
"email": "approver@timegmbh.de",
|
||
"role": "COMPANY_ADMIN",
|
||
"initial_password": "Secret123",
|
||
}, headers=manager_headers)
|
||
assert resp.status_code == 201, resp.text
|
||
login = await client.post("/api/v1/auth/login", json={
|
||
"email": "approver@timegmbh.de",
|
||
"password": "Secret123",
|
||
})
|
||
assert login.status_code == 200, login.text
|
||
return {"Authorization": f"Bearer {login.json()['access_token']}"}
|
||
|
||
|
||
# ── Stamp In / Out ─────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stamp_in(client: AsyncClient, auth_headers):
|
||
resp = await client.post(
|
||
"/api/v1/time/stamp-in",
|
||
json={"source": "web"},
|
||
headers=auth_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert "entry" in data
|
||
assert data["entry"]["end_time"] is None
|
||
assert data["entry"]["status"] == "pending"
|
||
assert isinstance(data["warnings"], list)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stamp_in_twice_fails(client: AsyncClient, auth_headers):
|
||
"""Zweimal einzustempeln ohne auszustempeln muss 409 ergeben."""
|
||
await client.post("/api/v1/time/stamp-in", json={"source": "web"}, headers=auth_headers)
|
||
resp = await client.post("/api/v1/time/stamp-in", json={"source": "web"}, headers=auth_headers)
|
||
assert resp.status_code == 409
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stamp_out(client: AsyncClient, auth_headers):
|
||
await client.post("/api/v1/time/stamp-in", json={"source": "web"}, headers=auth_headers)
|
||
resp = await client.post("/api/v1/time/stamp-out", json={}, headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["entry"]["end_time"] is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stamp_out_without_in_fails(client: AsyncClient, auth_headers):
|
||
resp = await client.post("/api/v1/time/stamp-out", json={}, headers=auth_headers)
|
||
assert resp.status_code == 404
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_break_workflow(client: AsyncClient, auth_headers):
|
||
await client.post("/api/v1/time/stamp-in", json={"source": "web"}, headers=auth_headers)
|
||
|
||
resp = await client.post("/api/v1/time/break-start", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["break_start"] is not None
|
||
|
||
resp = await client.post("/api/v1/time/break-end", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["break_start"] is None
|
||
assert resp.json()["break_minutes"] >= 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_break_start_twice_fails(client: AsyncClient, auth_headers):
|
||
await client.post("/api/v1/time/stamp-in", json={"source": "web"}, headers=auth_headers)
|
||
await client.post("/api/v1/time/break-start", headers=auth_headers)
|
||
resp = await client.post("/api/v1/time/break-start", headers=auth_headers)
|
||
assert resp.status_code == 409
|
||
|
||
|
||
# ── Today ──────────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_today(client: AsyncClient, auth_headers):
|
||
resp = await client.get("/api/v1/time/today", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json(), list)
|
||
|
||
|
||
# ── Entries ────────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_entries(client: AsyncClient, auth_headers):
|
||
resp = await client.get("/api/v1/time/entries", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "total" in data
|
||
assert "items" in data
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_manual_entry(client: AsyncClient, auth_headers):
|
||
resp = await client.post(
|
||
"/api/v1/time/entries",
|
||
json={
|
||
"date": str(date.today()),
|
||
"start_time": "09:00:00",
|
||
"end_time": "17:00:00",
|
||
"break_minutes": 30,
|
||
"source": "manual",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert data["entry"]["status"] == "pending"
|
||
assert data["entry"]["source"] == "manual"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_manual_entry_arbzg_warning(client: AsyncClient, auth_headers):
|
||
"""Mehr als 6h ohne Pause → ArbZG Warnung."""
|
||
resp = await client.post(
|
||
"/api/v1/time/entries",
|
||
json={
|
||
"date": str(date.today()),
|
||
"start_time": "08:00:00",
|
||
"end_time": "15:00:00",
|
||
"break_minutes": 0,
|
||
"source": "manual",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert len(data["warnings"]) > 0
|
||
assert any("Pause" in w for w in data["warnings"])
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_entry(client: AsyncClient, auth_headers):
|
||
create_resp = await client.post(
|
||
"/api/v1/time/entries",
|
||
json={
|
||
"date": str(date.today()),
|
||
"start_time": "09:00:00",
|
||
"end_time": "17:00:00",
|
||
"break_minutes": 30,
|
||
"source": "manual",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
entry_id = create_resp.json()["entry"]["id"]
|
||
|
||
resp = await client.patch(
|
||
f"/api/v1/time/entries/{entry_id}",
|
||
json={"break_minutes": 45, "correction_note": "Pause vergessen einzutragen"},
|
||
headers=auth_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["break_minutes"] == 45
|
||
assert resp.json()["correction_note"] == "Pause vergessen einzutragen"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_entry(client: AsyncClient, manager_headers, time_approver_headers):
|
||
"""Manager genehmigt einen Eintrag eines anderen Benutzers (kein Self-Approval)."""
|
||
create_resp = await client.post(
|
||
"/api/v1/time/entries",
|
||
json={
|
||
"date": str(date.today()),
|
||
"start_time": "09:00:00",
|
||
"end_time": "17:30:00",
|
||
"break_minutes": 30,
|
||
"source": "manual",
|
||
},
|
||
headers=manager_headers,
|
||
)
|
||
assert create_resp.status_code == 201
|
||
entry_id = create_resp.json()["entry"]["id"]
|
||
|
||
resp = await client.post(
|
||
f"/api/v1/time/entries/{entry_id}/approve",
|
||
headers=time_approver_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "approved"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reject_entry(client: AsyncClient, manager_headers):
|
||
create_resp = await client.post(
|
||
"/api/v1/time/entries",
|
||
json={
|
||
"date": str(date.today()),
|
||
"start_time": "08:00:00",
|
||
"end_time": "20:00:00",
|
||
"break_minutes": 0,
|
||
"source": "manual",
|
||
},
|
||
headers=manager_headers,
|
||
)
|
||
entry_id = create_resp.json()["entry"]["id"]
|
||
|
||
resp = await client.post(
|
||
f"/api/v1/time/entries/{entry_id}/reject",
|
||
json={"rejection_note": "Zeiten unrealistisch"},
|
||
headers=manager_headers,
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "rejected"
|
||
|
||
|
||
# ── Balance ────────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_balance(client: AsyncClient, registered_user, auth_headers):
|
||
user_id = registered_user["user"]["id"]
|
||
resp = await client.get(f"/api/v1/time/balance/{user_id}", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "total_hours_worked" in data
|
||
assert "expected_hours" in data
|
||
assert "overtime_hours" in data
|
||
|
||
|
||
# ── Work Schedules ─────────────────────────────────────────────────────────────
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_and_list_schedules(client: AsyncClient, manager_headers):
|
||
resp = await client.post(
|
||
"/api/v1/time/schedules",
|
||
json={
|
||
"name": "Vollzeit 40h",
|
||
"mon_h": "8.00",
|
||
"tue_h": "8.00",
|
||
"wed_h": "8.00",
|
||
"thu_h": "8.00",
|
||
"fri_h": "8.00",
|
||
"sat_h": "0.00",
|
||
"sun_h": "0.00",
|
||
"valid_from": str(date.today()),
|
||
},
|
||
headers=manager_headers,
|
||
)
|
||
assert resp.status_code == 201
|
||
assert resp.json()["name"] == "Vollzeit 40h"
|
||
|
||
list_resp = await client.get("/api/v1/time/schedules", headers=manager_headers)
|
||
assert list_resp.status_code == 200
|
||
assert len(list_resp.json()) >= 1
|
||
|
||
|
||
# ── ArbZG Unit Tests ──────────────────────────────────────────────────────────
|
||
|
||
def test_arbzg_check_ok():
|
||
from app.services.time_service import _check_arbzg
|
||
warnings = _check_arbzg(time(9, 0), time(17, 0), 30)
|
||
assert len(warnings) == 0
|
||
|
||
|
||
def test_arbzg_check_no_break_over_6h():
|
||
from app.services.time_service import _check_arbzg
|
||
warnings = _check_arbzg(time(9, 0), time(16, 0), 0)
|
||
assert any("30 min" in w for w in warnings)
|
||
|
||
|
||
def test_arbzg_check_break_too_short_over_9h():
|
||
from app.services.time_service import _check_arbzg
|
||
warnings = _check_arbzg(time(8, 0), time(18, 0), 30)
|
||
assert any("45 min" in w for w in warnings)
|
||
|
||
|
||
def test_arbzg_check_over_10h():
|
||
from app.services.time_service import _check_arbzg
|
||
warnings = _check_arbzg(time(6, 0), time(17, 0), 0)
|
||
assert any("10 Stunden" in w for w in warnings)
|
||
|
||
|
||
def test_rest_period_warning():
|
||
from app.services.time_service import _check_rest_period
|
||
from datetime import date, time
|
||
warnings = _check_rest_period(
|
||
time(22, 0), date(2026, 3, 26),
|
||
time(7, 0), date(2026, 3, 27)
|
||
)
|
||
assert any("11h" in w for w in warnings)
|
||
|
||
|
||
def test_rest_period_ok():
|
||
from app.services.time_service import _check_rest_period
|
||
from datetime import date, time
|
||
warnings = _check_rest_period(
|
||
time(17, 0), date(2026, 3, 26),
|
||
time(8, 0), date(2026, 3, 27)
|
||
)
|
||
assert len(warnings) == 0
|