Initial commit – TimeMaster Zeiterfassung & HR-Tool
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>
This commit is contained in:
sysops
@@ -0,0 +1,309 @@
|
||||
"""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}"}
|
||||
|
||||
|
||||
# ── 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):
|
||||
"""Manager genehmigt einen Eintrag."""
|
||||
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=manager_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
|
||||
Reference in New Issue
Block a user