"""Who may do what in the attendance module, data scopes, and the legacy client contract.""" import datetime from apps.access import services as access from apps.access.models import MembershipRole from apps.organizations.models import OrganizationMembership from domain.value_objects import JalaliDate from infrastructure.models import Attendance, LeaveRequest, User from tests.base import PASSWORD, CoreTestCase TODAY = datetime.date.today() ISO = TODAY.isoformat() J = JalaliDate.from_gregorian(TODAY) REPORT = f"?year={J.year}&month={J.month}" V1 = "/api/v1/attendance" class EmployeePermissionTests(CoreTestCase): def setUp(self): super().setUp() self.client_e = self.api(self.employee_a, self.org_a) def test_employee_self_service_flow(self): self.assertEqual(self.client_e.get(f"{V1}/today/").data["check_in"], None) checked_in = self.client_e.post(f"{V1}/check-in/") self.assertEqual(checked_in.status_code, 200) self.assertIsNotNone(checked_in.data["check_in"]) again = self.client_e.post(f"{V1}/check-in/") self.assertEqual(again.data["check_in"], checked_in.data["check_in"]) # idempotent out = self.client_e.post(f"{V1}/checkout/", {"tasks_note": "finished the report"}, format="json") self.assertEqual(out.status_code, 200) self.assertEqual(self.client_e.post(f"{V1}/checkout/", {"tasks_note": "again"}, format="json").status_code, 400) report = self.client_e.get(f"{V1}/reports/monthly/{REPORT}") self.assertEqual(report.status_code, 200) leave = self.client_e.post(f"{V1}/leave-requests/", {"date": ISO, "type": "daily", "category": "annual"}, format="json") self.assertEqual(leave.status_code, 201) self.assertEqual(len(self.client_e.get(f"{V1}/leave-requests/").data), 1) self.assertEqual(Attendance.objects.get(user=self.employee_a).organization, self.org_a) self.assertEqual(LeaveRequest.objects.get(user=self.employee_a).organization, self.org_a) def test_employee_cannot_use_manager_endpoints(self): forbidden = ( ("get", f"{V1}/team/today/", None), ("get", f"{V1}/holidays/", None), ("post", f"{V1}/holidays/", {"date": ISO, "title": "x"}), ("get", f"{V1}/work-schedule/default/", None), ("put", f"{V1}/work-schedule/default/", {"standard_check_in": "08:00", "standard_check_out": "16:00"}), ("get", f"{V1}/leave-requests/review/", None), ("post", f"{V1}/leave-requests/grant/", {"user_id": self.employee_a.pk, "date": ISO, "category": "annual"}), ("put", f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00"}), ("get", f"{V1}/reports/monthly/{REPORT}&user_id={self.manager_a.pk}", None), ) for method, url, body in forbidden: response = getattr(self.client_e, method)(url, body, format="json") if body else getattr(self.client_e, method)(url) self.assertEqual(response.status_code, 403, url) self.assertEqual(response.data["code"], "permission_denied", url) self.assertFalse(Attendance.objects.exists()) # the forbidden edit did not write anything def test_employee_own_report_with_explicit_user_id_is_allowed(self): response = self.client_e.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_a.pk}") self.assertEqual(response.status_code, 200) def test_input_validation(self): self.assertEqual(self.client_e.get(f"{V1}/reports/monthly/").status_code, 400) self.assertEqual(self.client_e.get(f"{V1}/reports/monthly/{REPORT}&user_id=abc").status_code, 400) self.assertEqual(self.client_e.post(f"{V1}/checkout/", {"tasks_note": "x"}, format="json").status_code, 400) # too short manager = self.api(self.manager_a, self.org_a) self.assertEqual(manager.put(f"{V1}/records/{self.employee_a.pk}/not-a-date/", {}, format="json").status_code, 400) self.assertEqual( manager.put(f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "17:00", "check_out": "09:00"}, format="json").status_code, 400, ) class SubjectsAndContextTests(CoreTestCase): def test_subjects_list_is_scoped_permissioned_and_includes_managers(self): subjects = self.api(self.manager_a, self.org_a).get(f"{V1}/subjects/") self.assertEqual(subjects.status_code, 200) self.assertEqual({u["personnel_code"] for u in subjects.data}, {"employee-a", "manager-a", "manager-a2", "multi"}) self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{V1}/subjects/").status_code, 403) self.assertEqual(self.api(self.manager_a, self.org_b).get(f"{V1}/subjects/").status_code, 403) def test_me_context_tells_whether_the_caller_is_a_member_and_has_a_name(self): own = self.api(self.employee_a, self.org_a).get("/api/v1/me/").data self.assertTrue(own["context"]["is_member"]) self.assertTrue(own["user"]["has_name"]) staff = self.api(self.root, self.org_a).get("/api/v1/me/").data self.assertFalse(staff["context"]["is_member"]) # reaching in through the platform role only self.assertTrue(staff["context"]["via_platform"]) nameless = User.objects.create_user(personnel_code="nameless", password=PASSWORD) self.join(nameless, self.org_a, "employee") self.assertFalse(self.api(nameless, self.org_a).get("/api/v1/me/").data["user"]["has_name"]) class ManagerPermissionTests(CoreTestCase): def test_manager_can_manage_everything_in_their_organization(self): manager = self.api(self.manager_a, self.org_a) self.assertEqual(manager.get(f"{V1}/team/today/").status_code, 200) self.assertEqual(manager.post(f"{V1}/holidays/", {"date": ISO, "title": "Holiday"}, format="json").status_code, 201) self.assertEqual( manager.put(f"{V1}/work-schedule/default/", {"standard_check_in": "08:00", "standard_check_out": "16:00"}, format="json").status_code, 200, ) leave = LeaveRequest.objects.create(organization=self.org_a, user=self.employee_a, date=TODAY, type="daily", category="annual") self.assertEqual(manager.post(f"{V1}/leave-requests/{leave.pk}/approve/").status_code, 200) self.assertEqual(manager.post(f"{V1}/leave-requests/{leave.pk}/approve/").status_code, 400) # no double review edit = manager.put(f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00", "check_out": "17:00"}, format="json") self.assertEqual(edit.status_code, 200) report = manager.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_a.pk}") self.assertEqual(next(r for r in report.data["rows"] if r["date"] == ISO)["worked"], "8:00") def test_a_deny_override_removes_a_single_manager_capability(self): target = self.membership(self.manager_a2, self.org_a) self.api(self.manager_a, self.org_a).put( f"/api/v1/members/{target.pk}/overrides/", {"permission": "attendance.update", "effect": "deny"}, format="json" ) client = self.api(self.manager_a2, self.org_a) self.assertEqual(client.put(f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00"}, format="json").status_code, 403) self.assertEqual(client.get(f"{V1}/team/today/").status_code, 200) def test_deny_on_create_stops_self_service_but_not_reading(self): target = self.membership(self.employee_a, self.org_a) self.api(self.manager_a, self.org_a).put( f"/api/v1/members/{target.pk}/overrides/", {"permission": "attendance.create", "effect": "deny"}, format="json" ) client = self.api(self.employee_a, self.org_a) self.assertEqual(client.post(f"{V1}/check-in/").status_code, 403) self.assertEqual(client.post(f"{V1}/leave-requests/", {"date": ISO, "type": "daily", "category": "annual"}, format="json").status_code, 403) self.assertEqual(client.get(f"{V1}/today/").status_code, 200) class DataScopeTests(CoreTestCase): def _role(self, code, grants): root_ctx = self.ctx(self.root, self.org_a) role = access.create_role(actor=root_ctx, code=code, name=code) access.set_role_permissions(actor=root_ctx, role=role, grants=grants) return role def test_report_permission_at_self_scope_does_not_reach_other_people(self): role = self._role("self-reporter", [{"code": "attendance.report", "scope": "self"}]) MembershipRole.objects.create(membership=self.membership(self.employee_a, self.org_a), role=role) client = self.api(self.employee_a, self.org_a) self.assertEqual(client.get(f"{V1}/team/today/").status_code, 403) self.assertEqual(client.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.manager_a.pk}").status_code, 403) def test_organization_scope_reaches_everyone_in_the_organization_only(self): role = self._role("org-reporter", [{"code": "attendance.report", "scope": "organization"}, {"code": "attendance.view", "scope": "self"}]) MembershipRole.objects.create(membership=self.membership(self.employee_a, self.org_a), role=role) client = self.api(self.employee_a, self.org_a) self.assertEqual(client.get(f"{V1}/team/today/").status_code, 200) self.assertEqual(client.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.manager_a.pk}").status_code, 200) self.assertEqual(client.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_b.pk}").status_code, 404) # reading is not editing self.assertEqual(client.put(f"{V1}/records/{self.manager_a.pk}/{ISO}/", {"check_in": "09:00"}, format="json").status_code, 403) def test_custom_role_combines_only_what_it_lists(self): role = self._role("approver", [{"code": "attendance.approve", "scope": "organization"}]) MembershipRole.objects.create(membership=self.join(self.stranger, self.org_a), role=role) client = self.api(self.stranger, self.org_a) self.assertEqual(client.get(f"{V1}/leave-requests/review/").status_code, 200) self.assertEqual(client.get(f"{V1}/team/today/").status_code, 403) self.assertEqual(client.get(f"{V1}/today/").status_code, 403) # no attendance.view either class PlatformStaffAttendanceTests(CoreTestCase): def setUp(self): super().setUp() self.exec_manager = self.make_user("exec-1") self.join(self.exec_manager, self.platform, "exec_manager") self.assign(self.exec_manager, self.org_a) def test_assigned_staff_can_read_reports_but_have_no_attendance_of_their_own(self): client = self.api(self.exec_manager, self.org_a) self.assertEqual(client.get(f"{V1}/team/today/").status_code, 200) report = client.get(f"{V1}/reports/monthly/{REPORT}&user_id={self.employee_a.pk}") self.assertEqual(report.status_code, 200) own = client.get(f"{V1}/today/") self.assertEqual(own.status_code, 403) self.assertEqual(own.data["code"], "not_a_member") self.assertEqual(client.post(f"{V1}/check-in/").status_code, 403) # staff role has no attendance.create def test_staff_cannot_write_or_reach_unassigned_organizations(self): client = self.api(self.exec_manager, self.org_a) self.assertEqual(client.put(f"{V1}/records/{self.employee_a.pk}/{ISO}/", {"check_in": "09:00"}, format="json").status_code, 403) self.assertEqual(self.api(self.exec_manager, self.org_b).get(f"{V1}/team/today/").status_code, 403) def test_super_admin_has_no_attendance_record_unless_a_member(self): client = self.api(self.root, self.org_a) self.assertEqual(client.post(f"{V1}/check-in/").data["code"], "not_a_member") self.assertEqual(client.get(f"{V1}/team/today/").status_code, 200) self.assertFalse(Attendance.objects.filter(user=self.root).exists()) class LegacyClientCompatibilityTests(CoreTestCase): """The current frontend sends no X-Organization header and routes on `is_admin`.""" def login(self, code, org=None): client = self.api() if org is not None: client.credentials(HTTP_X_ORGANIZATION=org.slug) return client.post("/api/auth/login/", {"personnel_code": code, "password": PASSWORD}, format="json") def test_single_organization_users_need_no_header(self): employee = self.login("employee-a") self.assertEqual(employee.status_code, 200) self.assertFalse(employee.data["user"]["is_admin"]) self.assertTrue(Attendance.objects.filter(user=self.employee_a, organization=self.org_a, date=TODAY).exists()) # old login check-in manager = self.login("manager-a") self.assertTrue(manager.data["user"]["is_admin"]) # derived from attendance.report, not from the stored flag self.assertFalse(Attendance.objects.filter(user=self.manager_a).exists()) # managers are not auto checked in token = employee.data["access"] client = self.api() client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}") self.assertEqual(client.get("/api/attendance/today/").status_code, 200) self.assertEqual(client.get("/api/me/").data["is_admin"], False) self.assertEqual(client.get(f"/api/reports/monthly/{REPORT}").status_code, 200) def test_stored_is_admin_flag_no_longer_matters(self): self.employee_a.is_admin = True self.employee_a.save() self.assertFalse(self.login("employee-a").data["user"]["is_admin"]) self.manager_a.is_admin = False self.manager_a.save() self.assertTrue(self.login("manager-a").data["user"]["is_admin"]) def test_multi_organization_users_must_choose_an_organization(self): no_header = self.login("multi") self.assertEqual(no_header.status_code, 200) self.assertFalse(Attendance.objects.filter(user=self.multi).exists()) # ambiguous: no automatic check-in client = self.api(self.multi) self.assertEqual(client.get("/api/attendance/today/").status_code, 403) self.assertEqual(client.get("/api/attendance/today/").data["code"], "organization_required") as_employee_in_b = self.login("multi", self.org_b) self.assertFalse(as_employee_in_b.data["user"]["is_admin"]) self.assertEqual(Attendance.objects.get(user=self.multi).organization, self.org_b) as_manager_in_a = self.login("multi", self.org_a) self.assertTrue(as_manager_in_a.data["user"]["is_admin"]) self.assertEqual(Attendance.objects.filter(user=self.multi).count(), 1) def test_legacy_personnel_list_includes_managers(self): listing = self.api(self.manager_a).get("/api/admin/personnel/") self.assertEqual(listing.status_code, 200) self.assertEqual({u["personnel_code"] for u in listing.data}, {"employee-a", "manager-a", "manager-a2", "multi"}) def test_legacy_create_and_delete_personnel_use_the_core_member_flow(self): manager = self.api(self.manager_a) created = manager.post( "/api/admin/personnel/", {"personnel_code": "legacy-hire", "name": "New", "password": PASSWORD}, format="json" ) self.assertEqual(created.status_code, 201, created.data) member = OrganizationMembership.objects.get(user__personnel_code="legacy-hire", organization=self.org_a) self.assertEqual(list(member.membership_roles.values_list("role__code", flat=True)), ["employee"]) self.assertEqual(self.login("legacy-hire").status_code, 200) removed = manager.delete(f"/api/admin/personnel/{created.data['id']}/") self.assertEqual(removed.status_code, 204) member.refresh_from_db() self.assertEqual(member.status, "left") self.assertTrue(User.objects.filter(pk=created.data["id"]).exists()) # the account is kept self.assertEqual(manager.delete(f"/api/admin/personnel/{self.employee_b.pk}/").status_code, 404) # other tenant def test_legacy_endpoints_still_enforce_permissions(self): employee = self.api(self.employee_a) for method, url in (("get", "/api/admin/personnel/"), ("get", "/api/admin/holidays/"), ("get", "/api/admin/leave-requests/"), ("get", "/api/admin/attendance/today/"), ("get", "/api/admin/work-schedule/default/")): self.assertEqual(getattr(employee, method)(url).status_code, 403, url) self.assertEqual(self.api().get("/api/attendance/today/").status_code, 401) def test_new_check_in_endpoint_exists_on_legacy_paths_too(self): response = self.api(self.employee_a).post("/api/attendance/check-in/") self.assertEqual(response.status_code, 200) self.assertEqual(Attendance.objects.filter(user=self.employee_a).count(), 1)