"""Member management and sign-in identifiers (username, verified mobile, legacy personnel code).""" import datetime from django.utils import timezone from apps.audit.models import AuditLog, AuditLogImmutableError from infrastructure.models import User from tests.base import PASSWORD, CoreTestCase class SignInIdentifierTests(CoreTestCase): def login(self, identifier, password=PASSWORD): return self.api().post("/api/v1/auth/login/", {"identifier": identifier, "password": password}, format="json") def test_username_is_case_insensitive(self): self.make_user("Ali.K") self.assertEqual(self.login("ali.k").status_code, 200) self.assertEqual(self.login("ALI.K").status_code, 200) def test_legacy_personnel_code_still_signs_in(self): User.objects.create_user(personnel_code="1500", password=PASSWORD, username="someone-else") self.assertEqual(self.login("1500").status_code, 200) old = self.api().post("/api/auth/login/", {"personnel_code": "1500", "password": PASSWORD}, format="json") self.assertEqual(old.status_code, 200) self.assertIn("access", old.data) def test_mobile_signs_in_only_when_verified(self): user = self.make_user("mobile-user", mobile="09123456789") self.assertEqual(self.login("09123456789").status_code, 401) user.mobile_verified_at = timezone.now() user.save() for variant in ("09123456789", "+98 912 345 6789", "۰۹۱۲۳۴۵۶۷۸۹"): self.assertEqual(self.login(variant).status_code, 200, variant) def test_wrong_password_and_unknown_identifier_look_the_same(self): self.make_user("real-user") wrong = self.login("real-user", "nope") unknown = self.login("ghost-user", "nope") self.assertEqual((wrong.status_code, wrong.data), (unknown.status_code, unknown.data)) self.assertEqual(wrong.data["code"], "invalid_credentials") def test_inactive_account_cannot_sign_in(self): user = self.make_user("gone") user.is_active = False user.save() self.assertEqual(self.login("gone").status_code, 401) def test_login_returns_reachable_organizations(self): response = self.login("multi") self.assertEqual({o["slug"] for o in response.data["organizations"]}, {"org-a", "org-b"}) self.assertEqual(response.data["user"]["username"], "multi") class MemberManagementTests(CoreTestCase): def test_manager_adds_a_member_with_roles(self): client = self.api(self.manager_a, self.org_a) response = client.post( "/api/v1/members/", {"username": "New.Hire", "password": PASSWORD, "name": "تازه‌وارد", "mobile": "0912 111 2233", "personnel_code": "E-100", "role_ids": [self.role(self.org_a, "employee").pk]}, format="json", ) self.assertEqual(response.status_code, 201, response.data) self.assertEqual(response.data["user"]["username"], "new.hire") self.assertEqual(response.data["user"]["mobile"], "09121112233") self.assertFalse(response.data["user"]["mobile_verified"]) self.assertEqual([r["code"] for r in response.data["roles"]], ["employee"]) self.assertEqual(self.api().post( "/api/v1/auth/login/", {"identifier": "new.hire", "password": PASSWORD}, format="json").status_code, 200) def test_username_collisions_do_not_reveal_other_organizations(self): client = self.api(self.manager_a, self.org_a) taken_elsewhere = client.post("/api/v1/members/", {"username": "employee-b", "password": PASSWORD}, format="json") taken_here = client.post("/api/v1/members/", {"username": "employee-a", "password": PASSWORD}, format="json") self.assertEqual(taken_elsewhere.status_code, 400) self.assertEqual(taken_elsewhere.data, taken_here.data) def test_member_validation(self): client = self.api(self.manager_a, self.org_a) weak = client.post("/api/v1/members/", {"username": "weakling", "password": "1"}, format="json") bad_mobile = client.post("/api/v1/members/", {"username": "badmobile", "password": PASSWORD, "mobile": "123"}, format="json") short_name = client.post("/api/v1/members/", {"username": "ab", "password": PASSWORD}, format="json") for response in (weak, bad_mobile, short_name): self.assertEqual(response.status_code, 400) self.assertFalse(User.objects.filter(username__in=["weakling", "badmobile", "ab"]).exists()) def test_personnel_code_is_unique_per_organization_only(self): client_a = self.api(self.manager_a, self.org_a) client_b = self.api(self.manager_b, self.org_b) body = {"password": PASSWORD, "personnel_code": "E-1"} self.assertEqual(client_a.post("/api/v1/members/", {**body, "username": "user-a1"}, format="json").status_code, 201) self.assertEqual(client_b.post("/api/v1/members/", {**body, "username": "user-b1"}, format="json").status_code, 201) self.assertEqual(client_a.post("/api/v1/members/", {**body, "username": "user-a2"}, format="json").status_code, 400) def test_only_platform_staff_can_link_an_existing_account(self): body = {"username": "employee-b"} self.assertEqual(self.api(self.manager_a, self.org_a).post("/api/v1/members/link/", body, format="json").status_code, 403) linked = self.api(self.root, self.org_a).post("/api/v1/members/link/", body, format="json") self.assertEqual(linked.status_code, 201, linked.data) again = self.api(self.root, self.org_a).post("/api/v1/members/link/", body, format="json") self.assertEqual(again.status_code, 400) def test_member_status_changes_and_suspension_blocks_access(self): member = self.membership(self.employee_a, self.org_a) response = self.api(self.manager_a, self.org_a).patch(f"/api/v1/members/{member.pk}/", {"status": "suspended"}, format="json") self.assertEqual(response.status_code, 200) self.assertEqual(self.api(self.employee_a, self.org_a).get("/probe/").status_code, 403) own = self.membership(self.manager_a, self.org_a) self_change = self.api(self.manager_a, self.org_a).patch(f"/api/v1/members/{own.pk}/", {"status": "left"}, format="json") self.assertEqual(self_change.status_code, 400) def test_member_without_organization_scope_sees_only_themselves(self): from apps.access import services as access # give the employee members.view but only at "self" scope root_ctx = self.ctx(self.root, self.org_a) role = access.create_role(actor=root_ctx, code="viewer", name="Viewer") access.set_role_permissions(actor=root_ctx, role=role, grants=[{"code": "members.view", "scope": "self"}]) from apps.access.models import MembershipRole MembershipRole.objects.create(membership=self.membership(self.employee_a, self.org_a), role=role) listing = self.api(self.employee_a, self.org_a).get("/api/v1/members/") self.assertEqual(listing.status_code, 200) self.assertEqual({m["user"]["username"] for m in listing.data["results"]}, {"employee-a"}) other = self.membership(self.manager_a, self.org_a) self.assertEqual(self.api(self.employee_a, self.org_a).get(f"/api/v1/members/{other.pk}/").status_code, 404) def test_member_search_and_pagination(self): client = self.api(self.manager_a, self.org_a) found = client.get("/api/v1/members/?search=employee") self.assertEqual({m["user"]["username"] for m in found.data["results"]}, {"employee-a"}) page = client.get("/api/v1/members/?page_size=2") self.assertEqual(len(page.data["results"]), 2) self.assertEqual(page.data["count"], 4) self.assertIsNotNone(page.data["next"]) def test_navigation_is_built_from_effective_permissions(self): employee = self.api(self.employee_a, self.org_a).get("/api/v1/me/").data["context"]["navigation"] keys_employee = {i["key"] for i in employee} self.assertEqual(keys_employee, {"dashboard", "settings", "attendance_leave", "tasks", "personnel"}) manager = self.api(self.manager_a, self.org_a).get("/api/v1/me/").data["context"]["navigation"] keys_manager = {i["key"] for i in manager} self.assertIn("members", keys_manager) self.assertIn("attendance_report", keys_manager) self.assertNotIn("organizations", keys_manager) # platform-only entry self.assertNotIn("modules", {k for k in keys_manager if k == "organizations"}) class AuditLogTests(CoreTestCase): def test_audit_rows_are_append_only(self): entry = AuditLog.objects.create(action="test.created") entry.action = "tampered" with self.assertRaises(AuditLogImmutableError): entry.save() with self.assertRaises(AuditLogImmutableError): entry.delete() entry.refresh_from_db() self.assertEqual(entry.action, "test.created") def test_filters(self): client = self.api(self.manager_a, self.org_a) self.api(self.manager_a, self.org_a).post("/api/v1/roles/", {"code": "r1", "name": "R1"}, format="json") by_action = client.get("/api/v1/audit-logs/?action=role.created").data self.assertEqual(by_action["count"], 1) self.assertEqual(by_action["results"][0]["actor"]["username"], "manager-a") none = client.get("/api/v1/audit-logs/?action=does.not.exist").data self.assertEqual(none["count"], 0) def test_only_audit_viewers_can_read(self): self.assertEqual(self.api(self.employee_a, self.org_a).get("/api/v1/audit-logs/").status_code, 403) def test_dates_are_timezone_consistent(self): entry = AuditLog.objects.create(action="x") self.assertLess(abs((timezone.now() - entry.created_at).total_seconds()), datetime.timedelta(minutes=1).total_seconds())