"""Attendance data is tenant-owned: isolation, IDOR, module gating and mandatory test 10.""" import datetime from django.core.management import call_command from django.db import IntegrityError, transaction from apps.modules import services as modules from apps.organizations.models import OrganizationMembership from domain.value_objects import JalaliDate from infrastructure.models import Attendance, Holiday, LeaveRequest, WorkSchedule from tests.base import CoreTestCase TODAY = datetime.date.today() ISO = TODAY.isoformat() JALALI = JalaliDate.from_gregorian(TODAY) REPORT = f"?year={JALALI.year}&month={JALALI.month}" V1 = "/api/v1/attendance" def row_for_today(response): return next(r for r in response.data["rows"] if r["date"] == ISO) class AttendanceDataIsolationTests(CoreTestCase): def test_manager_own_attendance_is_selectable_and_matches_daily_and_monthly_reports(self): Attendance.objects.create( organization=self.org_a, user=self.manager_a, date=TODAY, check_in=datetime.time(8, 15), check_out=datetime.time(16, 30), ) client = self.api(self.manager_a, self.org_a) subjects = client.get(f"{V1}/subjects/") self.assertIn(self.manager_a.pk, {user["id"] for user in subjects.data}) daily = client.get(f"{V1}/team/today/") own = next(row for row in daily.data["rows"] if row["user"]["id"] == self.manager_a.pk) monthly = client.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.manager_a.pk}") self.assertEqual(monthly.status_code, 200) for row in (own["attendance"], row_for_today(monthly)): self.assertEqual(row["check_in"], "08:15:00") self.assertEqual(row["check_out"], "16:30:00") def test_same_person_has_separate_records_per_organization(self): Attendance.objects.create(organization=self.org_a, user=self.multi, date=TODAY, check_in=datetime.time(8, 0)) Attendance.objects.create(organization=self.org_b, user=self.multi, date=TODAY, check_in=datetime.time(13, 30)) in_a = row_for_today(self.api(self.multi, self.org_a).get(f"{V1}/reports/monthly/{REPORT}")) in_b = row_for_today(self.api(self.multi, self.org_b).get(f"{V1}/reports/monthly/{REPORT}")) self.assertEqual(in_a["check_in"], "08:00:00") self.assertEqual(in_b["check_in"], "13:30:00") def test_duplicate_day_is_rejected_only_inside_one_organization(self): Attendance.objects.create(organization=self.org_a, user=self.multi, date=TODAY) Attendance.objects.create(organization=self.org_b, user=self.multi, date=TODAY) # other tenant: fine with self.assertRaises(IntegrityError), transaction.atomic(): Attendance.objects.create(organization=self.org_a, user=self.multi, date=TODAY) def test_manager_cannot_read_another_organizations_member_report(self): Attendance.objects.create(organization=self.org_b, user=self.employee_b, date=TODAY, check_in=datetime.time(9, 0)) denied = self.api(self.manager_a, self.org_a).get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_b.pk}") self.assertEqual(denied.status_code, 404) allowed = self.api(self.manager_b, self.org_b).get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_b.pk}") self.assertEqual(allowed.status_code, 200) self.assertEqual(row_for_today(allowed)["check_in"], "09:00:00") def test_team_overview_lists_only_the_selected_organizations_members(self): Attendance.objects.create(organization=self.org_b, user=self.employee_b, date=TODAY, check_in=datetime.time(9, 0)) rows = self.api(self.manager_a, self.org_a).get(f"{V1}/team/today/").data["rows"] names = {r["user"]["personnel_code"] for r in rows} self.assertIn("employee-a", names) self.assertNotIn("employee-b", names) self.assertTrue(all(r["attendance"] is None for r in rows)) # B's check-in does not leak into A def test_manual_edit_cannot_cross_organizations(self): url = f"{V1}/records/{self.employee_b.pk}/{ISO}/" response = self.api(self.manager_a, self.org_a).put(url, {"check_in": "09:00", "check_out": "17:00"}, format="json") self.assertEqual(response.status_code, 404) self.assertFalse(Attendance.objects.filter(user=self.employee_b).exists()) own = self.api(self.manager_a, self.org_a).put( f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00", "check_out": "17:00"}, format="json" ) self.assertEqual(own.status_code, 200) self.assertEqual(Attendance.objects.get(user=self.employee_a).organization, self.org_a) def test_leave_review_is_scoped_and_foreign_ids_are_invisible(self): leave_b = LeaveRequest.objects.create( organization=self.org_b, user=self.employee_b, date=TODAY, type="daily", category="annual" ) listing = self.api(self.manager_a, self.org_a).get(f"{V1}/leave-requests/review/?status=all") self.assertEqual(listing.data, []) for action in ("approve", "reject"): response = self.api(self.manager_a, self.org_a).post(f"{V1}/leave-requests/{leave_b.pk}/{action}/") self.assertEqual(response.status_code, 404, action) leave_b.refresh_from_db() self.assertEqual(leave_b.status, "pending") self.assertEqual(self.api(self.manager_b, self.org_b).post(f"{V1}/leave-requests/{leave_b.pk}/approve/").status_code, 200) def test_suspended_and_former_members_drop_out_of_every_manager_view(self): for user, status in ((self.employee_a, "suspended"), (self.manager_a2, "left")): OrganizationMembership.objects.filter(user=user, organization=self.org_a).update(status=status) manager = self.api(self.manager_a, self.org_a) listed = {r["user"]["personnel_code"] for r in manager.get(f"{V1}/team/today/").data["rows"]} self.assertNotIn("employee-a", listed) self.assertNotIn("manager-a2", listed) subjects = {u["personnel_code"] for u in manager.get(f"{V1}/subjects/").data} self.assertEqual(subjects, {"manager-a", "multi"}) self.assertEqual(manager.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_a.pk}").status_code, 404) self.assertEqual(manager.put(f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00"}, format="json").status_code, 404) grant = manager.post( f"{V1}/leave-requests/grant/", {"user_id": self.employee_a.pk, "date": ISO, "category": "annual"}, format="json" ) self.assertEqual(grant.status_code, 400) self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{V1}/today/").status_code, 403) def test_granting_leave_to_a_member_of_another_organization_is_refused(self): response = self.api(self.manager_a, self.org_a).post( f"{V1}/leave-requests/grant/", {"user_id": self.employee_b.pk, "date": ISO, "category": "annual"}, format="json" ) self.assertEqual(response.status_code, 400) self.assertEqual(response.data["code"], "user_not_found") self.assertFalse(LeaveRequest.objects.exists()) def test_holidays_are_per_organization(self): body = {"date": ISO, "title": "Local holiday"} self.assertEqual(self.api(self.manager_a, self.org_a).post(f"{V1}/holidays/", body, format="json").status_code, 201) self.assertEqual(self.api(self.manager_b, self.org_b).post(f"{V1}/holidays/", body, format="json").status_code, 201) again = self.api(self.manager_a, self.org_a).post(f"{V1}/holidays/", body, format="json") self.assertEqual(again.status_code, 400) # duplicate inside A only holiday_b = Holiday.objects.get(organization=self.org_b) self.assertEqual(self.api(self.manager_a, self.org_a).delete(f"{V1}/holidays/{holiday_b.pk}/").status_code, 204) self.assertTrue(Holiday.objects.filter(pk=holiday_b.pk).exists()) # A could not delete B's holiday self.assertEqual(len(self.api(self.manager_b, self.org_b).get(f"{V1}/holidays/").data), 1) def test_holiday_and_schedule_affect_only_their_own_organizations_report(self): Holiday.objects.create(organization=self.org_a, date=TODAY, title="A only") self.api(self.manager_a, self.org_a).put( f"{V1}/work-schedule/default/", {"standard_check_in": "08:00", "standard_check_out": "16:00"}, format="json" ) a_row = row_for_today(self.api(self.manager_a, self.org_a).get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_a.pk}")) b_row = row_for_today(self.api(self.manager_b, self.org_b).get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_b.pk}")) self.assertTrue(a_row["is_holiday"]) self.assertEqual(a_row["standard_check_in"], "08:00:00") self.assertFalse(b_row["is_holiday"]) self.assertIsNone(b_row["standard_check_in"]) self.assertEqual(WorkSchedule.objects.filter(organization=self.org_b).count(), 0) def test_work_schedule_management_is_isolated(self): self.api(self.manager_a, self.org_a).put( f"{V1}/work-schedule/{ISO}/", {"standard_check_in": "09:00", "standard_check_out": "17:00"}, format="json" ) self.assertEqual(self.api(self.manager_b, self.org_b).get(f"{V1}/work-schedule/").data, []) self.assertEqual(self.api(self.manager_b, self.org_b).delete(f"{V1}/work-schedule/{ISO}/").status_code, 204) self.assertEqual(WorkSchedule.objects.filter(organization=self.org_a).count(), 1) class ModuleGatingOnAttendanceTests(CoreTestCase): ENDPOINTS = ( ("get", f"{V1}/today/"), ("post", f"{V1}/check-in/"), ("get", f"{V1}/reports/monthly/{REPORT}"), ("get", f"{V1}/leave-requests/"), ("get", f"{V1}/team/today/"), ("get", f"{V1}/holidays/"), ("get", f"{V1}/work-schedule/default/"), ("get", f"{V1}/leave-requests/review/"), ) def test_every_attendance_endpoint_is_blocked_when_the_module_is_disabled(self): modules.disable_module(actor=self.root, organization=self.org_a, module_code="attendance") for user in (self.employee_a, self.manager_a): for method, url in self.ENDPOINTS: response = getattr(self.api(user, self.org_a), method)(url) self.assertEqual(response.status_code, 403, (user.personnel_code, url)) self.assertEqual(response.data["code"], "module_disabled", url) # legacy paths are gated the same way self.assertEqual(self.api(self.employee_a).get("/api/attendance/today/").data["code"], "module_disabled") # another organization keeps working self.assertEqual(self.api(self.employee_b, self.org_b).get(f"{V1}/today/").status_code, 200) def test_10_disabling_the_module_keeps_all_data_and_enabling_restores_access(self): """Mandatory test 10 (now with organization-owned attendance data).""" Attendance.objects.create(organization=self.org_a, user=self.employee_a, date=TODAY, check_in=datetime.time(8, 0)) Holiday.objects.create(organization=self.org_a, date=TODAY + datetime.timedelta(days=1), title="H") leave = LeaveRequest.objects.create( organization=self.org_a, user=self.employee_a, date=TODAY, type="daily", category="annual" ) WorkSchedule.objects.create( organization=self.org_a, date=None, standard_check_in=datetime.time(8, 0), standard_check_out=datetime.time(16, 0) ) before = {m.__name__: m.objects.filter(organization=self.org_a).count() for m in (Attendance, Holiday, LeaveRequest, WorkSchedule)} self.assertTrue(all(count == 1 for count in before.values())) modules.disable_module(actor=self.root, organization=self.org_a, module_code="attendance") after = {m.__name__: m.objects.filter(organization=self.org_a).count() for m in (Attendance, Holiday, LeaveRequest, WorkSchedule)} self.assertEqual(after, before) self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{V1}/today/").status_code, 403) modules.enable_module(actor=self.root, organization=self.org_a, module_code="attendance") today = self.api(self.employee_a, self.org_a).get(f"{V1}/today/") self.assertEqual(today.status_code, 200) self.assertEqual(today.data["check_in"], "08:00:00") # the data came back untouched self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{V1}/leave-requests/").data[0]["id"], leave.pk) def test_inactive_organization_blocks_attendance_for_its_members(self): self.org_a.is_active = False self.org_a.save() self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{V1}/today/").status_code, 403) class DefaultCheckoutCommandTests(CoreTestCase): def test_command_processes_only_organizations_with_the_module_enabled(self): yesterday = TODAY - datetime.timedelta(days=1) open_a = Attendance.objects.create(organization=self.org_a, user=self.employee_a, date=yesterday, check_in=datetime.time(9, 0)) open_b = Attendance.objects.create(organization=self.org_b, user=self.employee_b, date=yesterday, check_in=datetime.time(9, 0)) modules.disable_module(actor=self.root, organization=self.org_b, module_code="attendance") call_command("apply_default_checkout", verbosity=0) open_a.refresh_from_db() open_b.refresh_from_db() self.assertEqual(open_a.check_out, datetime.time(17, 30)) self.assertTrue(open_a.is_default_checkout) self.assertIsNone(open_b.check_out) # module disabled: nothing is processed def test_command_never_crosses_tenants_in_a_single_run(self): yesterday = TODAY - datetime.timedelta(days=1) row_b = Attendance.objects.create(organization=self.org_b, user=self.employee_b, date=yesterday, check_in=datetime.time(9, 0)) call_command("apply_default_checkout", verbosity=0) row_b.refresh_from_db() self.assertEqual(row_b.organization, self.org_b) self.assertEqual(row_b.check_out, datetime.time(17, 30))