From 4996c91d12c51d52f27cc43fb38e8fe5657afa03 Mon Sep 17 00:00:00 2001 From: Alan Rominger Date: Mon, 3 Aug 2026 14:39:55 -0400 Subject: [PATCH] AAP-85067 Old RBAC sync regression tests for bulk claims (#16565) * test: add regression tests for old RBAC sync after bulk claims Tests that save_user_claims (which uses bulk_create, skipping signals) correctly syncs old Role.members when run through AwxJWTAuthentication. Covers add, remove, and multi-org/team scenarios. * test: call process_permissions instead of internal _sync_old_rbac Tests now exercise the public AwxJWTAuthentication.process_permissions() API with the JWT layer mocked, rather than calling the private _sync_old_rbac method directly. The test_bulk_claims_skips_old_rbac_signals test asserts that save_user_claims does NOT populate old Role.members via signals. On current DAB, save_user_claims uses the serial give_permission path which fires signals normally, so old Role.members IS populated. Co-Authored-By: Claude Opus 4.6 --- .../dab_rbac/test_claims_old_rbac_sync.py | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 awx/main/tests/functional/dab_rbac/test_claims_old_rbac_sync.py diff --git a/awx/main/tests/functional/dab_rbac/test_claims_old_rbac_sync.py b/awx/main/tests/functional/dab_rbac/test_claims_old_rbac_sync.py new file mode 100644 index 0000000000..965f8f8aae --- /dev/null +++ b/awx/main/tests/functional/dab_rbac/test_claims_old_rbac_sync.py @@ -0,0 +1,107 @@ +""" +Tests that save_user_claims (which uses bulk_create, skipping signals) +correctly syncs old Role.members when run through AwxJWTAuthentication. +""" + +from unittest import mock + +import pytest + +from ansible_base.jwt_consumer.awx.auth import AwxJWTAuthentication +from ansible_base.jwt_consumer.common.auth import JWTAuthentication +from ansible_base.rbac.claims import save_user_claims +from awx.main.models import Organization, Team + + +@pytest.mark.django_db +class TestClaimsOldRbacSync: + + def _build_claims(self, orgs, teams): + """Build a claims dict from org/team model instances.""" + objects = {"organization": [], "team": []} + object_roles = {} + + org_indexes = [] + for i, org in enumerate(orgs): + objects["organization"].append( + { + "ansible_id": str(org.resource.ansible_id), + "name": org.name, + } + ) + org_indexes.append(i) + + team_indexes = [] + for i, team in enumerate(teams): + org_idx = next(j for j, o in enumerate(orgs) if o.pk == team.organization_id) + objects["team"].append( + { + "ansible_id": str(team.resource.ansible_id), + "name": team.name, + "org": org_idx, + } + ) + team_indexes.append(i) + + if org_indexes: + object_roles["Organization Admin"] = {"content_type": "organization", "objects": org_indexes} + if team_indexes: + object_roles["Team Member"] = {"content_type": "team", "objects": team_indexes} + + return {"objects": objects, "object_roles": object_roles, "global_roles": []} + + def _call_process_permissions(self, auth, user, claims): + """Call process_permissions with claims pre-loaded, mocking the JWT layer.""" + save_user_claims(user, **claims) + auth.common_auth.user = user + auth.common_auth._saved_claims = (claims["objects"], claims["object_roles"], claims["global_roles"]) + with mock.patch.object(JWTAuthentication, 'process_permissions'): + auth.process_permissions() + + def test_process_permissions_populates_old_rbac(self, bob, organization, team, setup_managed_roles): + """Verify that process_permissions populates old Role.members after bulk claims.""" + claims = self._build_claims([organization], [team]) + + auth = AwxJWTAuthentication() + self._call_process_permissions(auth, bob, claims) + + assert bob in organization.admin_role.members.all() + assert bob in team.member_role.members.all() + + def test_process_permissions_removes_stale_old_rbac(self, bob, organization, team, setup_managed_roles): + """Verify that process_permissions removes old Role.members when claims shrink.""" + auth = AwxJWTAuthentication() + + # First: give bob both org admin and team member + claims_full = self._build_claims([organization], [team]) + self._call_process_permissions(auth, bob, claims_full) + + assert bob in organization.admin_role.members.all() + assert bob in team.member_role.members.all() + + # Second: claims shrink to just org admin (no team member) + claims_reduced = self._build_claims([organization], []) + self._call_process_permissions(auth, bob, claims_reduced) + + assert bob in organization.admin_role.members.all() + assert bob not in team.member_role.members.all() + + def test_process_permissions_multiple_orgs_and_teams(self, bob, setup_managed_roles): + """Test sync at small scale with multiple orgs and teams.""" + orgs = [Organization.objects.create(name=f"sync-org-{i}") for i in range(3)] + teams = [] + for org in orgs: + teams.append(Team.objects.create(name=f"sync-team-{org.name}", organization=org)) + + claims = self._build_claims(orgs, teams) + + auth = AwxJWTAuthentication() + self._call_process_permissions(auth, bob, claims) + + for org in orgs: + org.refresh_from_db() + assert bob in org.admin_role.members.all(), f"bob not in {org.name}.admin_role" + + for team in teams: + team.refresh_from_db() + assert bob in team.member_role.members.all(), f"bob not in {team.name}.member_role"