"""NoTeam staff: assigned-organization scope, super admin reach, and audited cross-tenant access.""" from apps.access.models import RolePermission from apps.audit.models import AuditLog from infrastructure.models import User from tests.base import PASSWORD, CoreTestCase class PlatformStaffScopeTests(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_executive_reaches_only_assigned_organizations(self): in_a = self.api(self.exec_manager, self.org_a).get("/api/v1/members/") self.assertEqual(in_a.status_code, 200) self.assertEqual(self.api(self.exec_manager, self.org_b).get("/api/v1/members/").status_code, 403) def test_executive_only_holds_the_permissions_of_their_role(self): client = self.api(self.exec_manager, self.org_a) self.assertEqual(client.get("/api/v1/audit-logs/").status_code, 200) self.assertEqual(client.get("/api/v1/roles/").status_code, 403) self.assertEqual( client.post("/api/v1/members/", {"username": "x-person", "password": PASSWORD}, format="json").status_code, 403 ) self.assertEqual( self.api(self.exec_manager).post(f"/api/v1/organizations/{self.org_a.pk}/modules/attendance/disable/").status_code, 403 ) def test_organization_list_is_limited_to_assignments(self): listed = self.api(self.exec_manager).get("/api/v1/organizations/").data self.assertEqual({o["slug"] for o in listed}, {"org-a"}) everything = self.api(self.root).get("/api/v1/organizations/").data self.assertEqual({o["slug"] for o in everything}, {"noteam", "org-a", "org-b"}) def test_me_lists_only_reachable_organizations(self): me = self.api(self.exec_manager).get("/api/v1/me/").data self.assertEqual({o["slug"] for o in me["organizations"]}, {"noteam", "org-a"}) root_me = self.api(self.root).get("/api/v1/me/").data self.assertEqual({o["slug"] for o in root_me["organizations"]}, {"noteam", "org-a", "org-b"}) def test_super_admin_has_full_reach_and_context_marks_platform_access(self): response = self.api(self.root, self.org_b).get("/api/v1/me/") self.assertEqual(response.status_code, 200) self.assertTrue(response.data["context"]["via_platform"]) self.assertIn("modules.manage", response.data["context"]["permissions"]) def test_platform_staff_access_to_a_customer_is_audited(self): self.api(self.exec_manager, self.org_a).get("/api/v1/members/") entry = AuditLog.objects.filter(action="tenant.platform_access", organization=self.org_a).first() self.assertIsNotNone(entry) self.assertEqual(entry.actor, self.exec_manager) def test_assignments_are_managed_by_super_admin_only(self): membership = self.membership(self.exec_manager, self.platform) url = f"/api/v1/staff/{membership.pk}/assignments/" self.assertEqual(self.api(self.manager_a).put(url, {"organization_ids": [self.org_b.pk]}, format="json").status_code, 403) self.assertEqual(self.api(self.exec_manager).put(url, {"organization_ids": [self.org_b.pk]}, format="json").status_code, 403) ok = self.api(self.root).put(url, {"organization_ids": [self.org_a.pk, self.org_b.pk]}, format="json") self.assertEqual(ok.status_code, 200) self.assertEqual(self.api(self.exec_manager, self.org_b).get("/api/v1/members/").status_code, 200) def test_suspended_staff_member_loses_customer_access(self): membership = self.membership(self.exec_manager, self.platform) membership.status = "suspended" membership.save() self.assertEqual(self.api(self.exec_manager, self.org_a).get("/api/v1/members/").status_code, 403) def test_super_admin_creates_a_customer_organization_with_seeded_roles(self): created = self.api(self.root).post("/api/v1/organizations/", {"name": "Org C", "slug": "org-c"}, format="json") self.assertEqual(created.status_code, 201) roles = self.api(self.root, self.org_a).get("/api/v1/roles/").data self.assertEqual({r["code"] for r in roles}, {"org_manager", "employee"}) duplicate = self.api(self.root).post("/api/v1/organizations/", {"name": "dup", "slug": "org-c"}, format="json") self.assertEqual(duplicate.status_code, 400) def test_new_organization_starts_with_attendance_and_tasks_only(self): self.api(self.root).post("/api/v1/organizations/", {"name": "Org C", "slug": "org-c"}, format="json") from apps.organizations.models import Organization org_c = Organization.objects.get(slug="org-c") from apps.modules.services import enabled_module_codes self.assertEqual(enabled_module_codes(org_c), {"core", "attendance", "tasks", "personnel"}) def test_organization_can_be_created_with_its_first_manager(self): r = self.api(self.root).post( "/api/v1/organizations/", {"name": "Org D", "slug": "org-d", "manager": {"username": "boss-d", "password": "Str0ng-pass-77", "name": "Boss"}}, format="json", ) self.assertEqual(r.status_code, 201, r.data) login = self.api().post("/api/v1/auth/login/", {"identifier": "boss-d", "password": "Str0ng-pass-77"}, format="json") self.assertEqual(login.status_code, 200) from apps.organizations.models import Organization org_d = Organization.objects.get(slug="org-d") boss = User.objects.get(username="boss-d") ctx = self.api(boss, org_d).get("/api/v1/me/").data["context"] self.assertIn("tasks.view", ctx["permissions"]) self.assertIn("members.create", ctx["permissions"]) def test_super_admin_can_appoint_manager_in_org_with_few_modules(self): from apps.organizations.models import Organization self.api(self.root).post("/api/v1/organizations/", {"name": "Org G", "slug": "org-g"}, format="json") org_g = Organization.objects.get(slug="org-g") manager_role = self.role(org_g, "org_manager") r = self.api(self.root, org_g).post( "/api/v1/members/", {"username": "boss-g", "password": "Str0ng-pass-77", "role_ids": [manager_role.pk]}, format="json", ) self.assertEqual(r.status_code, 201, r.data) def test_super_admin_directory_groups_users_by_organization(self): groups = {g["organization"]["slug"]: g for g in self.api(self.root).get("/api/v1/platform/directory/").data} self.assertEqual({"noteam", "org-a", "org-b"}, set(groups)) names_a = {m["user"]["username"] for m in groups["org-a"]["members"]} self.assertIn("employee-a", names_a) self.assertNotIn("employee-b", names_a) self.assertTrue(all(m["is_manager"] for m in groups["org-a"]["members"][:3])) # managers come first self.assertFalse(groups["org-a"]["members"][-1]["is_manager"]) self.assertEqual({m["username"] for m in groups["org-a"]["managers"]}, {"manager-a", "manager-a2", "multi"}) self.assertEqual(groups["org-b"]["managers"][0]["username"], "manager-b") found = self.api(self.root).get("/api/v1/platform/directory/", {"search": "employee-a"}).data self.assertEqual([g["organization"]["slug"] for g in found], ["org-a"]) self.assertEqual(self.api(self.manager_a, self.org_a).get("/api/v1/platform/directory/").status_code, 403) self.assertEqual(self.api(self.manager_a).get("/api/v1/platform/directory/").status_code, 403) def test_failed_manager_leaves_no_organization(self): r = self.api(self.root).post( "/api/v1/organizations/", {"name": "Org E", "slug": "org-e", "manager": {"username": "ab", "password": "x"}}, format="json", ) self.assertEqual(r.status_code, 400) from apps.organizations.models import Organization self.assertFalse(Organization.objects.filter(slug="org-e").exists()) def test_super_admin_sees_tasks_in_a_new_organization(self): # a super admin role seeded before the tasks module existed has no tasks permissions yet RolePermission.objects.filter(role__code="super_admin", permission__module_id="tasks").delete() self.api(self.root).post("/api/v1/organizations/", {"name": "Org F", "slug": "org-f"}, format="json") from apps.organizations.models import Organization ctx = self.api(self.root, Organization.objects.get(slug="org-f")).get("/api/v1/me/").data["context"] self.assertIn("tasks", {i["key"] for i in ctx["navigation"]}) def test_organization_can_be_deactivated_and_reactivated_by_staff(self): url = f"/api/v1/organizations/{self.org_a.pk}/" self.assertEqual(self.api(self.root).patch(url, {"is_active": False}, format="json").status_code, 200) self.assertEqual(self.api(self.manager_a, self.org_a).get("/api/v1/members/").status_code, 403) self.assertEqual(self.api(self.root, self.org_a).get("/api/v1/members/").status_code, 200) self.assertEqual(self.api(self.root).patch(url, {"is_active": True}, format="json").status_code, 200) self.assertEqual(self.api(self.manager_a, self.org_a).get("/api/v1/members/").status_code, 200) def test_super_admin_who_is_also_a_member_keeps_platform_powers(self): """Regression: a person can be NoTeam super admin AND a manager of a customer org.""" self.join(self.root, self.org_a, "org_manager") response = self.api(self.root).post(f"/api/v1/organizations/{self.org_a.pk}/modules/attendance/disable/") self.assertEqual(response.status_code, 200, response.data) context = self.api(self.root, self.org_a).get("/api/v1/me/").data["context"] self.assertTrue(context["via_platform"]) self.assertIn("modules.manage", context["permissions"]) self.assertIn("members.create", context["permissions"]) def test_membership_and_staff_access_are_merged_not_exclusive(self): """An executive assigned to A who is only an employee in B: staff power in A, employee-only in B.""" self.join(self.exec_manager, self.org_b, "employee") in_b = self.api(self.exec_manager, self.org_b).get("/api/v1/me/").data["context"] self.assertFalse(in_b["via_platform"]) self.assertEqual( set(in_b["permissions"]), {"attendance.view", "attendance.create", "tasks.view", "tasks.create", "tasks.update", "personnel.view"}, ) self.assertEqual(self.api(self.exec_manager, self.org_b).get("/api/v1/members/").status_code, 403) self.join(self.exec_manager, self.org_a, "employee") in_a = self.api(self.exec_manager, self.org_a).get("/api/v1/me/").data["context"] self.assertTrue(in_a["via_platform"]) self.assertIn("attendance.create", in_a["permissions"]) # from the employee membership self.assertIn("members.view", in_a["permissions"]) # from the staff role def test_suspended_organization_membership_does_not_remove_staff_access(self): membership = self.join(self.root, self.org_a, "org_manager") membership.status = "suspended" membership.save() self.assertEqual(self.api(self.root, self.org_a).get("/api/v1/members/").status_code, 200) def test_platform_organization_cannot_be_deactivated(self): response = self.api(self.root).patch(f"/api/v1/organizations/{self.platform.pk}/", {"is_active": False}, format="json") self.assertEqual(response.status_code, 400)