import datetime from django.test import override_settings from apps.audit.models import AuditLog from apps.workplace import services from apps.workplace.models import LocationPolicy, PunchLog from infrastructure.models import Attendance from tests.base import CoreTestCase V1 = "/api/v1/attendance" OFFICE = (35.6892, 51.3890) # Tehran NEAR = (35.6895, 51.3893) # ~40 m away FAR = (35.7500, 51.4500) # ~8 km away def pos(point, accuracy=30): return {"latitude": point[0], "longitude": point[1], "accuracy": accuracy} class LocationPolicyTests(CoreTestCase): def setUp(self): super().setUp() self.mgr = self.api(self.manager_a, self.org_a) self.emp = self.api(self.employee_a, self.org_a) self.mgr.put(f"{V1}/location/policy/", {"enabled": True}, format="json") self.mgr.post(f"{V1}/location/places/", {"name": "HQ", "latitude": OFFICE[0], "longitude": OFFICE[1], "radius_m": 150}, format="json") def check_in(self, client=None, **extra): return (client or self.emp).post(f"{V1}/check-in/", extra.pop("body", {}), format="json", **extra) # ---- the decision def test_inside_the_work_place_is_accepted(self): r = self.check_in(body=pos(NEAR)) self.assertEqual(r.status_code, 200, r.data) self.assertEqual((r.data["location"]["status"], r.data["location"]["verdict"]), ("approved", "verified")) self.assertIsNotNone(r.data["check_in"]) def test_outside_is_recorded_but_waits_for_a_manager(self): r = self.check_in(body=pos(FAR)) self.assertEqual((r.status_code, r.data["location"]["status"], r.data["location"]["verdict"]), (200, "pending", "outside")) self.assertIsNotNone(r.data["check_in"]) # recorded, flagged self.assertEqual(PunchLog.objects.get().status, "pending") def test_no_location_and_no_network_is_refused_and_nothing_is_recorded(self): r = self.check_in() self.assertEqual(r.status_code, 400) self.assertEqual(r.data["code"], "location_required") self.assertFalse(Attendance.objects.filter(user=self.employee_a).exists()) def test_poor_accuracy_cannot_prove_presence(self): r = self.check_in(body=pos(NEAR, accuracy=900)) self.assertEqual((r.data["location"]["status"], r.data["location"]["verdict"]), ("pending", "low_accuracy")) r2 = self.api(self.manager_a2, self.org_a).post(f"{V1}/check-in/", {**pos(NEAR), "accuracy": None}, format="json") self.assertEqual(r2.data["location"]["verdict"], "low_accuracy") # unknown accuracy counts as poor def test_policy_off_changes_nothing(self): self.mgr.put(f"{V1}/location/policy/", {"enabled": False}, format="json") r = self.check_in() self.assertEqual((r.status_code, r.data["location"]["required"]), (200, False)) self.assertEqual(PunchLog.objects.count(), 0) def test_checkout_is_checked_too_and_can_be_excluded(self): self.check_in(body=pos(NEAR)) bad = self.emp.post(f"{V1}/checkout/", {"tasks_note": "done work", **pos(FAR)}, format="json") self.assertEqual((bad.status_code, bad.data["location"]["status"]), (200, "pending")) self.assertEqual(self.emp.post(f"{V1}/checkout/", {"tasks_note": "again"}, format="json").status_code, 400) # already out other = self.api(self.manager_a2, self.org_a) other.post(f"{V1}/check-in/", pos(NEAR), format="json") self.assertEqual(other.post(f"{V1}/checkout/", {"tasks_note": "no position"}, format="json").status_code, 400) self.mgr.put(f"{V1}/location/policy/", {"apply_to_checkout": False}, format="json") self.assertEqual(other.post(f"{V1}/checkout/", {"tasks_note": "no position"}, format="json").status_code, 200) def test_second_check_in_the_same_day_is_a_noop(self): self.check_in(body=pos(NEAR)) self.check_in(body=pos(FAR)) self.assertEqual(PunchLog.objects.count(), 1) def test_rejected_check_in_can_be_retried_and_is_verified_again(self): self.check_in(body=pos(FAR)) original = PunchLog.objects.get() rejected = self.mgr.post(f"{V1}/location/punches/{original.pk}/reject/", {"note": "outside"}, format="json") self.assertEqual(rejected.status_code, 200) row = Attendance.objects.get(user=self.employee_a) self.assertIsNone(row.check_in) self.assertEqual(self.check_in().status_code, 400) row.refresh_from_db() self.assertIsNone(row.check_in) self.mgr.post(f"{V1}/location/networks/", {"label": "Office", "cidr": "203.0.113.9"}, format="json") retried = self.check_in(REMOTE_ADDR="203.0.113.9") self.assertEqual(retried.status_code, 200) self.assertIsNotNone(retried.data["check_in"]) self.assertEqual(retried.data["location"]["verdict"], "network") self.assertEqual(Attendance.objects.filter(user=self.employee_a).count(), 1) self.assertEqual(PunchLog.objects.count(), 2) original.refresh_from_db() self.assertEqual(original.status, "rejected") repeated = self.check_in(REMOTE_ADDR="203.0.113.9") self.assertEqual(repeated.data["check_in"], retried.data["check_in"]) self.assertEqual(PunchLog.objects.count(), 2) def test_distance_is_computed_on_the_server(self): self.assertAlmostEqual(services.distance_m(*OFFICE, *OFFICE), 0, places=3) self.assertTrue(30 < services.distance_m(*OFFICE, *NEAR) < 60) self.assertTrue(7000 < services.distance_m(*OFFICE, *FAR) < 12000) # ---- network def test_allowed_office_network_needs_no_location(self): self.mgr.post(f"{V1}/location/networks/", {"label": "Office", "cidr": "203.0.113.0/24"}, format="json") r = self.check_in(REMOTE_ADDR="203.0.113.9") self.assertEqual((r.status_code, r.data["location"]["verdict"]), (200, "network")) self.assertEqual(self.api(self.manager_a2, self.org_a).post(f"{V1}/check-in/", {}, format="json", REMOTE_ADDR="198.51.100.4").status_code, 400) def test_a_forged_forwarded_header_is_ignored_by_default(self): self.mgr.post(f"{V1}/location/networks/", {"label": "Office", "cidr": "203.0.113.5"}, format="json") r = self.check_in(REMOTE_ADDR="198.51.100.4", HTTP_X_FORWARDED_FOR="203.0.113.5") self.assertEqual(r.status_code, 400) # still "no location" @override_settings(ATTENDANCE_TRUSTED_PROXY_HOPS=1) def test_behind_a_trusted_proxy_the_proxy_appended_address_is_used(self): self.mgr.post(f"{V1}/location/networks/", {"label": "Office", "cidr": "203.0.113.5"}, format="json") ok = self.check_in(REMOTE_ADDR="10.0.0.1", HTTP_X_FORWARDED_FOR="9.9.9.9, 203.0.113.5") self.assertEqual(ok.data["location"]["verdict"], "network") spoof = self.api(self.manager_a2, self.org_a).post( f"{V1}/check-in/", {}, format="json", REMOTE_ADDR="10.0.0.1", HTTP_X_FORWARDED_FOR="203.0.113.5, 9.9.9.9") self.assertEqual(spoof.status_code, 400) # the client-written first entry is not trusted def test_network_validation(self): for bad in ("not-an-ip", "300.1.1.1", "10.0.0.0/8"): self.assertEqual(self.mgr.post(f"{V1}/location/networks/", {"label": "x", "cidr": bad}, format="json").status_code, 400, bad) # ---- exemptions def test_exempt_people_can_punch_from_anywhere(self): self.mgr.post(f"{V1}/location/exemptions/", {"user": self.employee_a.pk, "reason": "field sales"}, format="json") r = self.check_in() self.assertEqual((r.status_code, r.data["location"]["verdict"]), (200, "exempt")) def test_exemption_for_other_dates_does_not_apply_today(self): far = (datetime.date.today() + datetime.timedelta(days=5)).isoformat() self.mgr.post(f"{V1}/location/exemptions/", {"user": self.employee_a.pk, "start_date": far, "end_date": far}, format="json") self.assertEqual(self.check_in().status_code, 400) def test_exemption_rules(self): self.assertEqual(self.mgr.post(f"{V1}/location/exemptions/", {"user": self.employee_b.pk}, format="json").status_code, 400) # other org body = {"user": self.employee_a.pk, "start_date": "2025-02-01", "end_date": "2025-01-01"} self.assertEqual(self.mgr.post(f"{V1}/location/exemptions/", body, format="json").status_code, 400) # ---- review def pending(self): self.check_in(body=pos(FAR)) return PunchLog.objects.get() def test_rejecting_removes_the_time_and_needs_a_reason(self): punch = self.pending() self.assertEqual(self.mgr.post(f"{V1}/location/punches/{punch.pk}/reject/", {}, format="json").status_code, 400) r = self.mgr.post(f"{V1}/location/punches/{punch.pk}/reject/", {"note": "was at home"}, format="json") self.assertEqual((r.status_code, r.data["status"], r.data["reviewed_by_name"]), (200, "rejected", "manager-a")) self.assertIsNone(Attendance.objects.get(user=self.employee_a).check_in) self.assertEqual(self.mgr.post(f"{V1}/location/punches/{punch.pk}/approve/").status_code, 400) # only once def test_approving_keeps_the_time(self): punch = self.pending() r = self.mgr.post(f"{V1}/location/punches/{punch.pk}/approve/", {"note": "was on a visit"}, format="json") self.assertEqual(r.data["status"], "approved") self.assertIsNotNone(Attendance.objects.get(user=self.employee_a).check_in) def test_rejecting_a_checkout_clears_only_the_checkout(self): self.check_in(body=pos(NEAR)) self.emp.post(f"{V1}/checkout/", {"tasks_note": "finished", **pos(FAR)}, format="json") punch = PunchLog.objects.get(kind="out") self.mgr.post(f"{V1}/location/punches/{punch.pk}/reject/", {"note": "not at the office"}, format="json") row = Attendance.objects.get(user=self.employee_a) self.assertIsNotNone(row.check_in) self.assertIsNone(row.check_out) def test_review_list_and_filters(self): self.pending() listing = self.mgr.get(f"{V1}/location/punches/").data self.assertEqual((len(listing), listing[0]["user_name"], listing[0]["verdict"]), (1, "employee-a", "outside")) self.assertEqual(self.mgr.get(f"{V1}/location/punches/", {"status": "reviewed"}).data, []) # ---- access, tenants, audit def test_only_managers_configure_and_review(self): punch = self.pending() for method, url in (("get", "policy/"), ("get", "places/"), ("get", "networks/"), ("get", "exemptions/"), ("get", "punches/"), ("get", "my-ip/")): self.assertEqual(getattr(self.emp, method)(f"{V1}/location/{url}").status_code, 403, url) self.assertEqual(self.emp.put(f"{V1}/location/policy/", {"enabled": False}, format="json").status_code, 403) self.assertEqual(self.emp.post(f"{V1}/location/punches/{punch.pk}/approve/").status_code, 403) def test_status_endpoint_tells_the_client_whether_to_ask_for_a_location(self): s = self.emp.get(f"{V1}/location/status/", REMOTE_ADDR="198.51.100.4").data self.assertEqual((s["enabled"], s["needs_location"], s["network_ok"]), (True, True, False)) self.mgr.post(f"{V1}/location/networks/", {"label": "O", "cidr": "198.51.100.4"}, format="json") s = self.emp.get(f"{V1}/location/status/", REMOTE_ADDR="198.51.100.4").data self.assertEqual((s["needs_location"], s["network_ok"]), (False, True)) def test_tenant_isolation(self): punch = self.pending() other = self.api(self.manager_b, self.org_b) self.assertEqual(other.get(f"{V1}/location/places/").data, []) self.assertEqual(other.post(f"{V1}/location/punches/{punch.pk}/approve/").status_code, 404) self.assertEqual(other.get(f"{V1}/location/punches/").data, []) place_id = self.mgr.get(f"{V1}/location/places/").data[0]["id"] self.assertEqual(other.delete(f"{V1}/location/places/{place_id}/").status_code, 404) # organization B has no policy: its people punch freely self.assertEqual(self.api(self.employee_b, self.org_b).post(f"{V1}/check-in/", {}, format="json").data["location"]["required"], False) def test_place_validation_and_audit(self): bad = lambda **kw: self.mgr.post(f"{V1}/location/places/", {"name": "x", "latitude": 35, "longitude": 51, **kw}, format="json").status_code self.assertEqual(bad(latitude=120), 400) self.assertEqual(bad(longitude=200), 400) self.assertEqual(bad(radius_m=5), 400) self.assertEqual(bad(name=" "), 400) self.assertTrue(AuditLog.objects.filter(action="location.policy_update", organization=self.org_a).exists()) def test_legacy_sign_in_does_not_bypass_the_policy(self): r = self.api().post("/api/auth/login/", {"personnel_code": "employee-a", "password": "Str0ng-pass!"}, format="json") self.assertEqual(r.status_code, 200) self.assertFalse(Attendance.objects.filter(user=self.employee_a).exists()) LocationPolicy.objects.filter(organization=self.org_a).update(enabled=False) self.api().post("/api/auth/login/", {"personnel_code": "employee-a", "password": "Str0ng-pass!"}, format="json") self.assertTrue(Attendance.objects.filter(user=self.employee_a).exists())