import datetime from apps.audit.models import AuditLog from apps.modules import services as modules from apps.payroll.models import PayrollItem from apps.personnel.models import Contract from infrastructure.models import Attendance, LeaveRequest, WorkSchedule from tests.base import CoreTestCase RUNS = "/api/v1/payroll/runs/" Y, M = 1403, 1 # 2024-03-20 .. 2024-04-19 class PayrollTests(CoreTestCase): def setUp(self): super().setUp() for code in ("personnel", "payroll"): modules.enable_module(actor=self.root, organization=self.org_a, module_code=code) self.m_emp = self.membership(self.employee_a, self.org_a) self.m_mgr = self.membership(self.manager_a, self.org_a) Contract.objects.create( organization=self.org_a, membership=self.m_emp, title="c", start_date=datetime.date(2023, 1, 1), pay_amount=22_000_000, ) WorkSchedule.objects.create( organization=self.org_a, date=None, standard_check_in=datetime.time(8), standard_check_out=datetime.time(16) ) Attendance.objects.create( organization=self.org_a, user=self.employee_a, date=datetime.date(2024, 3, 25), check_in=datetime.time(9, 0), check_out=datetime.time(17, 30), is_default_checkout=False, ) self.mgr = self.api(self.manager_a, self.org_a) def run_id(self): r = self.mgr.post(RUNS, {"year": Y, "month": M}, format="json") self.assertEqual(r.status_code, 201, r.data) return r.data["id"] def calculated(self): pk = self.run_id() self.assertEqual(self.mgr.post(f"{RUNS}{pk}/calculate/").status_code, 200) return pk def item(self, pk): return self.mgr.get(f"{RUNS}{pk}/").data["items"][0] def test_calculation_from_contract_and_attendance(self): pk = self.calculated() it = self.item(pk) self.assertEqual(it["late_minutes"], 60) self.assertEqual(it["overtime_minutes"], 90) self.assertEqual(it["overtime_pay"], 210_000) # 100000/h x 1.5h x 1.4 self.assertEqual(it["late_deduction"], 100_000) self.assertEqual(it["net"], 22_110_000) def test_members_without_contract_are_skipped(self): pk = self.calculated() self.assertEqual(len(self.mgr.get(f"{RUNS}{pk}/").data["items"]), 1) def test_manual_adjustments_survive_recalculation(self): pk = self.calculated() item_id = self.item(pk)["id"] r = self.mgr.patch(f"/api/v1/payroll/items/{item_id}/", {"additions": 500_000, "deductions": 200_000}, format="json") self.assertEqual(r.data["net"], 22_410_000) self.mgr.post(f"{RUNS}{pk}/calculate/") self.assertEqual(self.item(pk)["net"], 22_410_000) def test_settings_are_frozen_on_the_run(self): pk = self.run_id() self.assertEqual(self.mgr.put("/api/v1/payroll/settings/", {"overtime_multiplier": "2.00", "deduct_lateness": False}, format="json").status_code, 200) self.mgr.post(f"{RUNS}{pk}/calculate/") self.assertEqual(self.item(pk)["overtime_pay"], 210_000) # run keeps 1.4 / deduct self.assertEqual(self.item(pk)["late_deduction"], 100_000) def test_approve_locks_the_run(self): pk = self.calculated() item_id = self.item(pk)["id"] self.assertEqual(self.mgr.post(f"{RUNS}{pk}/approve/").status_code, 200) self.assertEqual(self.mgr.post(f"{RUNS}{pk}/approve/").status_code, 400) self.assertEqual(self.mgr.post(f"{RUNS}{pk}/calculate/").status_code, 400) self.assertEqual(self.mgr.patch(f"/api/v1/payroll/items/{item_id}/", {"additions": 1}, format="json").status_code, 400) self.assertEqual(self.mgr.delete(f"{RUNS}{pk}/").status_code, 400) self.assertEqual(PayrollItem.objects.count(), 1) def test_cannot_approve_empty_run_or_duplicate_month(self): pk = self.run_id() self.assertEqual(self.mgr.post(f"{RUNS}{pk}/approve/").status_code, 400) self.assertEqual(self.mgr.post(RUNS, {"year": Y, "month": M}, format="json").status_code, 400) self.assertEqual(self.mgr.post(RUNS, {"year": Y, "month": 13}, format="json").status_code, 400) def test_employee_sees_only_own_payslip_of_approved_runs(self): pk = self.calculated() emp = self.api(self.employee_a, self.org_a) self.assertEqual(emp.get(RUNS).data, []) # draft is hidden self.assertEqual(emp.get(f"{RUNS}{pk}/").status_code, 404) self.mgr.post(f"{RUNS}{pk}/approve/") self.assertEqual(len(emp.get(RUNS).data), 1) items = emp.get(f"{RUNS}{pk}/").data["items"] self.assertEqual([i["user"]["id"] for i in items], [self.employee_a.pk]) def test_other_employee_sees_nothing_of_someone_elses_pay(self): pk = self.calculated() self.mgr.post(f"{RUNS}{pk}/approve/") other = self.make_user("employee-a2") self.join(other, self.org_a, "employee") self.assertEqual(self.api(other, self.org_a).get(f"{RUNS}{pk}/").data["items"], []) def test_employee_cannot_manage(self): emp = self.api(self.employee_a, self.org_a) self.assertEqual(emp.post(RUNS, {"year": Y, "month": M}, format="json").status_code, 403) pk = self.calculated() self.assertEqual(emp.post(f"{RUNS}{pk}/calculate/").status_code, 403) self.assertEqual(emp.post(f"{RUNS}{pk}/approve/").status_code, 403) self.assertEqual(emp.get("/api/v1/payroll/settings/").status_code, 403) def test_tenant_isolation_and_module_gate(self): pk = self.calculated() self.assertEqual(self.api(self.manager_b, self.org_b).get(RUNS).status_code, 403) # module off in B for code in ("personnel", "payroll"): modules.enable_module(actor=self.root, organization=self.org_b, module_code=code) other = self.api(self.manager_b, self.org_b) self.assertEqual(other.get(f"{RUNS}{pk}/").status_code, 404) self.assertEqual(other.post(f"{RUNS}{pk}/calculate/").status_code, 404) item_id = self.item(pk)["id"] self.assertEqual(other.patch(f"/api/v1/payroll/items/{item_id}/", {"additions": 1}, format="json").status_code, 404) def test_payroll_needs_its_dependencies(self): from domain.exceptions import DomainError modules.disable_module(actor=self.root, organization=self.org_b, module_code="personnel") with self.assertRaises(DomainError): modules.enable_module(actor=self.root, organization=self.org_b, module_code="payroll") def test_audit_trail(self): pk = self.calculated() self.mgr.post(f"{RUNS}{pk}/approve/") actions = set(AuditLog.objects.filter(organization=self.org_a).values_list("action", flat=True)) self.assertTrue({"payroll.run_create", "payroll.calculate", "payroll.approve"} <= actions) # ---- daily / hourly pay types def _attendance_week(self): Attendance.objects.create( organization=self.org_a, user=self.employee_a, date=datetime.date(2024, 3, 26), check_in=datetime.time(8), check_out=datetime.time(16), is_default_checkout=False, ) for day, category in ((27, "annual"), (28, "unpaid")): LeaveRequest.objects.create( organization=self.org_a, user=self.employee_a, date=datetime.date(2024, 3, day), type="daily", category=category, status="approved", ) def _set_contract(self, pay_type, amount): Contract.objects.filter(membership=self.m_emp).update(pay_type=pay_type, pay_amount=amount) def test_daily_pay_counts_worked_days_and_paid_leave(self): self._attendance_week() self._set_contract("daily", 1_000_000) it = self.item(self.calculated()) self.assertEqual((it["pay_type"], it["unit_rate"], it["paid_days"]), ("daily", 1_000_000, 3)) # 2 worked + annual leave self.assertEqual(it["base_salary"], 3_000_000) self.assertEqual(it["overtime_pay"], 262_500) # 1.5h x (1,000,000 / 8h) x 1.4 self.assertEqual(it["late_deduction"], 125_000) self.assertEqual(it["net"], 3_137_500) def test_hourly_pay_is_worked_hours_without_late_deduction(self): self._attendance_week() self._set_contract("hourly", 100_000) it = self.item(self.calculated()) self.assertEqual(it["base_salary"], 1_500_000) # (990 - 90 overtime) min = 15 h self.assertEqual(it["overtime_pay"], 210_000) self.assertEqual(it["late_deduction"], 0) self.assertEqual(it["net"], 1_710_000) def test_daily_pay_follows_the_working_day_length(self): self._attendance_week() self._set_contract("daily", 1_000_000) WorkSchedule.objects.filter(organization=self.org_a, date=None).update( standard_check_in=datetime.time(8), standard_check_out=datetime.time(14) ) it = self.item(self.calculated()) # 6-hour day: hourly 166,667 -> late 1h, overtime counted from 14:00 (3.5h + 2h) self.assertEqual(it["late_deduction"], 166_667) def test_changing_pay_type_changes_the_result_on_recalculation(self): self._attendance_week() pk = self.calculated() self.assertEqual(self.item(pk)["pay_type"], "monthly") self._set_contract("hourly", 100_000) self.mgr.post(f"{RUNS}{pk}/calculate/") self.assertEqual(self.item(pk)["pay_type"], "hourly") def test_contract_pay_type_validation_and_default(self): url = f"/api/v1/personnel/{self.m_emp.pk}/contracts/" bad = {"title": "x", "start_date": "2024-01-01", "pay_type": "weekly", "pay_amount": 5} self.assertEqual(self.mgr.post(url, bad, format="json").status_code, 400) ok = self.mgr.post(url, {"title": "x", "start_date": "2024-01-01", "pay_amount": 5}, format="json") self.assertEqual(ok.data["pay_type"], "monthly")