"""S4.3 — Time & Attendance backend tests. Covers: - ShiftDefinition services (create, update, delete with assignment guard) - EmployeeShift assignment - AttendanceRecord services: - check_in / check_out with work_minutes computation - manual create / update / delete - finalize (overtime calculation) - finalized-edit guard - Monthly summary selector - Registered permissions (4 new attendance/shift codenames) - API endpoints: - GET/POST /hr/shifts/ + PATCH/DELETE /hr/shifts/{id}/ - GET/POST /hr/employee-shifts/ - GET/POST /hr/attendance/ - PATCH/DELETE /hr/attendance/{public_id}/ - POST /hr/attendance/check-in/ - POST /hr/attendance/check-out/ - GET /hr/attendance/summary/ """ from __future__ import annotations from datetime import date, time import pytest from simorgh.apps.hr.models import ( AttendanceRecord, AttendanceSource, EmployeeShift, ShiftDefinition, ) from simorgh.apps.hr import services from simorgh.apps.hr.permissions import ( PERM_ATTENDANCE_MANAGE, PERM_ATTENDANCE_VIEW_OWN, PERM_ATTENDANCE_VIEW_TEAM, PERM_SHIFT_MANAGE, ) from simorgh.apps.iam.registry import sync_registry_to_db from simorgh.apps.iam.models import Permission, Role from simorgh.apps.memberships.models import Membership from rest_framework.test import APIClient # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _api(user=None): client = APIClient() if user: client.force_login(user) return client def _with_tenant(client, tenant): client.credentials(HTTP_X_TENANT=tenant.slug) return client # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def perms_synced(db): sync_registry_to_db() return {p.codename: p for p in Permission.objects.all()} @pytest.fixture def role_att_manager(tenant_acme, perms_synced): """Role with full attendance + shift management.""" role = Role.objects.create(tenant=tenant_acme, code="att_mgr", name="Attendance Manager") role.permissions.set([ perms_synced[PERM_ATTENDANCE_MANAGE], perms_synced[PERM_ATTENDANCE_VIEW_OWN], perms_synced[PERM_ATTENDANCE_VIEW_TEAM], perms_synced[PERM_SHIFT_MANAGE], ]) return role @pytest.fixture def role_att_viewer(tenant_acme, perms_synced): """ESS: can only view own attendance.""" role = Role.objects.create(tenant=tenant_acme, code="att_view", name="Attendance Viewer") role.permissions.set([perms_synced[PERM_ATTENDANCE_VIEW_OWN]]) return role @pytest.fixture def att_manager(alice, tenant_acme, acme_tree, role_att_manager): m = Membership.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], role=role_att_manager, ) m.users.add(alice) return alice @pytest.fixture def standard_shift(tenant_acme): """Standard 09:00-17:00, Mon-Fri, 60 min break.""" return services.create_shift( tenant=tenant_acme, name="Standard", start_time=time(9, 0), end_time=time(17, 0), work_days=[0, 1, 2, 3, 4], break_minutes=60, ) @pytest.fixture def employee_alice(alice, tenant_acme, acme_tree): from simorgh.apps.hr import services as hr_svc emp = hr_svc.create_employee( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, first_name="Alice", last_name="Test", hire_date=date(2020, 1, 1), employment_type="full_time", work_email="alice.att@acme.example", ) # link user so ESS check-in can find this employee emp.user = alice emp.save() return emp @pytest.fixture def shift_management_enabled(db, tenant_acme, acme_tree): """Enable hr.shift_management feature for tenant_acme via TenantFeatureOverride.""" from simorgh.apps.modules.models import TenantFeatureOverride return TenantFeatureOverride.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], feature_key="hr.shift_management", name="shift_management", enabled=True, ) # =========================================================================== # ShiftDefinition service tests # =========================================================================== class TestShiftServices: def test_create_shift(self, db, tenant_acme): shift = services.create_shift( tenant=tenant_acme, name="Night Shift", start_time=time(22, 0), end_time=time(6, 0), work_days=[0, 1, 2], break_minutes=30, ) assert shift.pk is not None assert shift.name == "Night Shift" assert shift.break_minutes == 30 def test_update_shift(self, db, standard_shift): updated = services.update_shift(standard_shift, name="Morning Shift", break_minutes=45) assert updated.name == "Morning Shift" assert updated.break_minutes == 45 def test_delete_shift_no_assignments(self, db, standard_shift): pk = standard_shift.pk services.delete_shift(standard_shift) assert not ShiftDefinition.objects.filter(pk=pk).exists() def test_delete_shift_with_assignments_raises(self, db, standard_shift, employee_alice, tenant_acme): services.assign_shift_to_employee( employee=employee_alice, shift=standard_shift, effective_from=date(2024, 1, 1), ) with pytest.raises(ValueError, match="active employee assignments"): services.delete_shift(standard_shift) def test_unique_shift_name_per_tenant(self, db, standard_shift, tenant_acme): from django.db import IntegrityError with pytest.raises(Exception): services.create_shift( tenant=tenant_acme, name="Standard", start_time=time(8, 0), end_time=time(16, 0), work_days=[0, 1, 2, 3, 4], ) def test_expected_work_minutes(self, db, standard_shift): # 17:00 - 09:00 = 480 min - 60 break = 420 assert standard_shift.expected_work_minutes == 420 # =========================================================================== # EmployeeShift assignment tests # =========================================================================== class TestEmployeeShiftAssignment: def test_assign_shift(self, db, employee_alice, standard_shift): ea = services.assign_shift_to_employee( employee=employee_alice, shift=standard_shift, effective_from=date(2024, 1, 1), effective_to=date(2024, 12, 31), ) assert ea.pk is not None assert ea.effective_from == date(2024, 1, 1) assert ea.effective_to == date(2024, 12, 31) def test_assign_shift_open_ended(self, db, employee_alice, standard_shift): ea = services.assign_shift_to_employee( employee=employee_alice, shift=standard_shift, effective_from=date(2024, 6, 1), ) assert ea.effective_to is None # =========================================================================== # AttendanceRecord service tests # =========================================================================== class TestCheckInOut: def test_check_in_creates_record(self, db, tenant_acme, employee_alice): record = services.check_in_employee( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(9, 5), ) assert record.check_in == time(9, 5) assert record.check_out is None assert record.work_minutes is None def test_check_out_computes_work_minutes(self, db, tenant_acme, employee_alice, standard_shift): # assign shift so break_minutes = 60 services.assign_shift_to_employee( employee=employee_alice, shift=standard_shift, effective_from=date(2024, 1, 1), ) services.check_in_employee( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(9, 0), ) record = services.check_out_employee( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(17, 0), ) # 480 min total - 60 break = 420 assert record.work_minutes == 420 def test_check_out_without_check_in_raises(self, db, tenant_acme, employee_alice): with pytest.raises(ValueError, match="No check-in"): services.check_out_employee( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(17, 0), ) def test_check_in_idempotent_update(self, db, tenant_acme, employee_alice): services.check_in_employee(tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(9, 0)) record = services.check_in_employee(tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), time=time(9, 10)) assert record.check_in == time(9, 10) assert AttendanceRecord.objects.filter(tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15)).count() == 1 class TestManualAttendance: def test_create_manual_record(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 20), check_in=time(8, 30), check_out=time(16, 30), source="manual", break_minutes=30, ) assert record.source == "manual" # 480 - 30 = 450 assert record.work_minutes == 450 def test_update_record(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 21), check_in=time(9, 0), check_out=time(17, 0), ) updated = services.update_attendance_record(record, note="corrected") assert updated.note == "corrected" def test_update_finalized_raises(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 22), check_in=time(9, 0), check_out=time(17, 0), ) services.finalize_attendance_record(record) with pytest.raises(ValueError, match="finalized"): services.update_attendance_record(record, note="X") def test_delete_record(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 23), ) pk = record.pk services.delete_attendance_record(record) assert not AttendanceRecord.objects.filter(pk=pk).exists() def test_delete_finalized_raises(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 24), check_in=time(9, 0), check_out=time(17, 0), ) services.finalize_attendance_record(record) with pytest.raises(ValueError, match="Finalized"): services.delete_attendance_record(record) class TestFinalizeAttendance: def test_finalize_sets_flag(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 4, 1), check_in=time(9, 0), check_out=time(17, 0), break_minutes=60, ) # work_minutes = 420, standard shift expected = 420 → no overtime result = services.finalize_attendance_record(record) assert result.is_finalized is True def test_finalize_computes_overtime(self, db, tenant_acme, employee_alice, standard_shift): services.assign_shift_to_employee( employee=employee_alice, shift=standard_shift, effective_from=date(2024, 1, 1) ) # check-in 09:00 check-out 18:30 = 570 min, break 60 = 510 work_min # expected 420, excess = 90 ≥ 30 threshold → overtime = 90 record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 4, 2), check_in=time(9, 0), check_out=time(18, 30), break_minutes=60, ) result = services.finalize_attendance_record(record) assert result.overtime_minutes == 90 def test_double_finalize_raises(self, db, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 4, 3), check_in=time(9, 0), check_out=time(17, 0), ) services.finalize_attendance_record(record) record.refresh_from_db() with pytest.raises(ValueError, match="already finalized"): services.finalize_attendance_record(record) class TestMonthlySummary: def test_summary_aggregates_correctly(self, db, tenant_acme, employee_alice): for day in range(1, 4): services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 5, day), check_in=time(9, 0), check_out=time(17, 0), break_minutes=60, ) summary = services.get_monthly_summary(tenant_acme, employee_alice, 2024, 5) assert summary["total_days"] == 3 assert summary["total_work_minutes"] == 3 * 420 assert summary["total_overtime_minutes"] == 0 assert summary["finalized_days"] == 0 def test_summary_empty_month(self, db, tenant_acme, employee_alice): summary = services.get_monthly_summary(tenant_acme, employee_alice, 2024, 6) assert summary["total_days"] == 0 assert summary["total_work_minutes"] == 0 # =========================================================================== # Permission registration tests # =========================================================================== class TestAttendancePermissionsRegistered: def test_attendance_permissions_exist(self, db): sync_registry_to_db() codenames = set(Permission.objects.values_list("codename", flat=True)) assert PERM_ATTENDANCE_VIEW_OWN in codenames assert PERM_ATTENDANCE_VIEW_TEAM in codenames assert PERM_ATTENDANCE_MANAGE in codenames assert PERM_SHIFT_MANAGE in codenames # =========================================================================== # Shift API tests # =========================================================================== @pytest.mark.django_db class TestShiftAPI: def test_list_shifts_requires_auth(self, tenant_acme): client = _api() _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/shifts/") assert res.status_code in (401, 403) def test_create_shift(self, att_manager, tenant_acme, shift_management_enabled): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/shifts/", { "name": "Evening", "start_time": "14:00:00", "end_time": "22:00:00", "work_days": [0, 1, 2, 3, 4], "break_minutes": 30, }, format="json") assert res.status_code == 201 assert res.data["name"] == "Evening" def test_list_shifts(self, att_manager, tenant_acme, standard_shift, shift_management_enabled): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/shifts/") assert res.status_code == 200 assert len(res.data) >= 1 def test_update_shift(self, att_manager, tenant_acme, standard_shift, shift_management_enabled): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.patch(f"/api/v1/hr/shifts/{standard_shift.pk}/", {"break_minutes": 45}, format="json") assert res.status_code == 200 assert res.data["break_minutes"] == 45 def test_delete_shift(self, att_manager, tenant_acme, shift_management_enabled): shift = services.create_shift( tenant=tenant_acme, name="Temp", start_time=time(8, 0), end_time=time(16, 0), work_days=[0], break_minutes=0, ) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.delete(f"/api/v1/hr/shifts/{shift.pk}/") assert res.status_code == 204 # =========================================================================== # EmployeeShift API tests # =========================================================================== @pytest.mark.django_db class TestEmployeeShiftAPI: def test_list_employee_shifts(self, att_manager, tenant_acme, employee_alice, standard_shift, shift_management_enabled): services.assign_shift_to_employee(employee_alice, standard_shift, date(2024, 1, 1)) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/employee-shifts/") assert res.status_code == 200 assert len(res.data) >= 1 def test_create_employee_shift(self, att_manager, tenant_acme, employee_alice, standard_shift, shift_management_enabled): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/employee-shifts/", { "employee_id": employee_alice.pk, "shift_id": standard_shift.pk, "effective_from": "2024-07-01", }, format="json") assert res.status_code == 201 assert res.data["effective_to"] is None # =========================================================================== # Attendance record API tests # =========================================================================== @pytest.mark.django_db class TestAttendanceListCreateAPI: def test_list_requires_auth(self, tenant_acme): client = _api() _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/attendance/") assert res.status_code in (401, 403) def test_admin_list_all_attendance(self, att_manager, tenant_acme, employee_alice): services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 1), check_in=time(9, 0), check_out=time(17, 0), ) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/attendance/") assert res.status_code == 200 assert len(res.data) >= 1 def test_admin_create_manual_record(self, att_manager, tenant_acme, employee_alice): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/attendance/", { "employee_id": employee_alice.pk, "date": "2024-03-10", "check_in": "09:00:00", "check_out": "17:00:00", "source": "manual", "break_minutes": 60, }, format="json") assert res.status_code == 201 assert res.data["work_minutes"] == 420 def test_list_filter_by_date(self, att_manager, tenant_acme, employee_alice): services.create_attendance_record(tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 5)) services.create_attendance_record(tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 10)) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/attendance/?date_from=2024-03-08&date_to=2024-03-12") assert res.status_code == 200 assert len(res.data) == 1 @pytest.mark.django_db class TestAttendanceDetailAPI: def test_patch_attendance(self, att_manager, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 15), ) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.patch(f"/api/v1/hr/attendance/{record.public_id}/", {"note": "corrected"}, format="json") assert res.status_code == 200 assert res.data["note"] == "corrected" def test_delete_attendance(self, att_manager, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 16), ) pk = record.pk client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.delete(f"/api/v1/hr/attendance/{record.public_id}/") assert res.status_code == 204 assert not AttendanceRecord.objects.filter(pk=pk).exists() def test_patch_finalized_returns_400(self, att_manager, tenant_acme, employee_alice): record = services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 3, 17), check_in=time(9, 0), check_out=time(17, 0), ) services.finalize_attendance_record(record) record.refresh_from_db() client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.patch(f"/api/v1/hr/attendance/{record.public_id}/", {"note": "X"}, format="json") assert res.status_code == 400 # =========================================================================== # Check-in / Check-out API tests # =========================================================================== @pytest.mark.django_db class TestCheckInOutAPI: def test_check_in_as_admin(self, att_manager, tenant_acme, employee_alice): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/attendance/check-in/", { "employee_id": employee_alice.pk, "date": "2024-04-10", "time": "09:00:00", "source": "biometric", }, format="json") assert res.status_code == 200 assert res.data["check_in"] == "09:00:00" assert res.data["source"] == "biometric" def test_check_out_after_check_in(self, att_manager, tenant_acme, employee_alice): services.check_in_employee( tenant=tenant_acme, employee=employee_alice, date=date(2024, 4, 11), time=time(9, 0), ) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/attendance/check-out/", { "employee_id": employee_alice.pk, "date": "2024-04-11", "time": "17:00:00", }, format="json") assert res.status_code == 200 assert res.data["check_out"] == "17:00:00" def test_check_out_no_check_in_returns_400(self, att_manager, tenant_acme, employee_alice): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/attendance/check-out/", { "employee_id": employee_alice.pk, "date": "2024-04-12", "time": "17:00:00", }, format="json") assert res.status_code == 400 def test_check_in_requires_auth(self, tenant_acme): client = _api() _with_tenant(client, tenant_acme) res = client.post("/api/v1/hr/attendance/check-in/", { "date": "2024-04-13", "time": "09:00:00", }, format="json") assert res.status_code in (401, 403) # =========================================================================== # Monthly summary API tests # =========================================================================== @pytest.mark.django_db class TestMonthlySummaryAPI: def test_summary_requires_auth(self, tenant_acme): client = _api() _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/attendance/summary/?year=2024&month=5") assert res.status_code in (401, 403) def test_summary_for_admin(self, att_manager, tenant_acme, employee_alice): for d in range(1, 4): services.create_attendance_record( tenant=tenant_acme, employee=employee_alice, date=date(2024, 8, d), check_in=time(9, 0), check_out=time(17, 0), break_minutes=60, ) client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get(f"/api/v1/hr/attendance/summary/?employee={employee_alice.pk}&year=2024&month=8") assert res.status_code == 200 assert res.data["total_days"] == 3 assert res.data["total_work_minutes"] == 3 * 420 def test_summary_missing_params_returns_400(self, att_manager, tenant_acme): client = _api(att_manager) _with_tenant(client, tenant_acme) res = client.get("/api/v1/hr/attendance/summary/") assert res.status_code == 400