mirror of
https://github.com/ansible/awx.git
synced 2026-10-29 22:03:20 -02:30
[Devel] Fix Migration From Old system_auditor to New platform_auditor (#16582)
* Fixed migration from old to new rbac for sys auditor * Added first migration * Added testing for migrations and fix * Removed attempts to remove platform auditor assignments * Fixed migration test failures
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
import logging
|
||||
|
||||
from django.db import migrations
|
||||
|
||||
logger = logging.getLogger('awx.main.migrations')
|
||||
|
||||
|
||||
def fix_system_auditor_assignments(apps, schema_editor):
|
||||
"""
|
||||
Corrective migration for a bug in 0192 (migrate_to_new_rbac) where a stale
|
||||
loop variable caused old system_auditor members to not receive Platform Auditor.
|
||||
Only adds missing assignments — does not remove any, since we cannot safely
|
||||
distinguish bug-granted assignments from legitimate ones (e.g. from 0202 or admin actions).
|
||||
"""
|
||||
Role = apps.get_model('main', 'Role')
|
||||
RoleDefinition = apps.get_model('dab_rbac', 'RoleDefinition')
|
||||
RoleUserAssignment = apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
|
||||
platform_auditor = RoleDefinition.objects.filter(name='Platform Auditor').first()
|
||||
if not platform_auditor:
|
||||
return
|
||||
|
||||
old_system_auditor = Role.objects.filter(singleton_name='system_auditor').first()
|
||||
if not old_system_auditor:
|
||||
return
|
||||
|
||||
expected_user_ids = set(old_system_auditor.members.values_list('id', flat=True))
|
||||
current_user_ids = set(RoleUserAssignment.objects.filter(role_definition=platform_auditor).values_list('user_id', flat=True))
|
||||
missing_user_ids = expected_user_ids - current_user_ids
|
||||
|
||||
for user_id in missing_user_ids:
|
||||
RoleUserAssignment.objects.create(user_id=user_id, role_definition=platform_auditor)
|
||||
|
||||
if missing_user_ids:
|
||||
logger.info(f'Fixed {len(missing_user_ids)} missing Platform Auditor assignments for old system_auditor members')
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
dependencies = [
|
||||
('main', '0207_alter_skip_tags_to_textfield'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RunPython(fix_system_auditor_assignments, migrations.RunPython.noop),
|
||||
]
|
||||
@@ -268,7 +268,7 @@ def migrate_to_new_rbac(apps, schema_editor):
|
||||
if old_system_auditor:
|
||||
# if the system auditor role is not present, this is a new install and no users should exist
|
||||
ct = 0
|
||||
for user in role.members.all():
|
||||
for user in old_system_auditor.members.all():
|
||||
RoleUserAssignment.objects.create(user=user, role_definition=new_system_auditor)
|
||||
ct += 1
|
||||
if ct:
|
||||
|
||||
@@ -224,3 +224,84 @@ class TestGithubAppBug:
|
||||
assert CredentialType.objects.filter(
|
||||
namespace='github_app_lookup', name='Legacy GitHub App Credential'
|
||||
).exists(), "CredentialType should be updated to 'github_app_lookup' and retain its name."
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestSystemAuditorMigration:
|
||||
def test_system_auditor_migrated_to_platform_auditor(self, migrator):
|
||||
"""
|
||||
The bug in 0192 used a stale loop variable `role` instead of
|
||||
`old_system_auditor`, so system_auditor members were not granted
|
||||
Platform Auditor. With the fix applied, they should be.
|
||||
"""
|
||||
old_state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more'))
|
||||
User = old_state.apps.get_model('auth', 'User')
|
||||
Role = old_state.apps.get_model('main', 'Role')
|
||||
|
||||
user = User.objects.create(username='sys-auditor-user')
|
||||
system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'})
|
||||
system_auditor_role.members.add(user)
|
||||
|
||||
new_state = migrator.apply_tested_migration(('main', '0192_custom_roles'))
|
||||
RoleUserAssignment = new_state.apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
|
||||
assert RoleUserAssignment.objects.filter(
|
||||
user=user.id, role_definition__name='Platform Auditor'
|
||||
).exists(), "system_auditor member should have been granted Platform Auditor"
|
||||
|
||||
def test_fix_system_auditor_migration_backfills_missing(self, migrator):
|
||||
"""
|
||||
Migration 0208 should backfill Platform Auditor for any
|
||||
system_auditor members who were missed by the original migration.
|
||||
"""
|
||||
state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more'))
|
||||
User = state.apps.get_model('auth', 'User')
|
||||
Role = state.apps.get_model('main', 'Role')
|
||||
|
||||
user = User.objects.create(username='missed-auditor')
|
||||
system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'})
|
||||
system_auditor_role.members.add(user)
|
||||
|
||||
state = migrator.apply_tested_migration(('main', '0192_custom_roles'))
|
||||
RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
RoleDefinition = state.apps.get_model('dab_rbac', 'RoleDefinition')
|
||||
|
||||
# Simulate the pre-fix state: remove the Platform Auditor assignment
|
||||
platform_auditor = RoleDefinition.objects.get(name='Platform Auditor')
|
||||
RoleUserAssignment.objects.filter(user=user.id, role_definition=platform_auditor).delete()
|
||||
assert not RoleUserAssignment.objects.filter(
|
||||
user=user.id, role_definition=platform_auditor
|
||||
).exists(), "Setup: user should not have Platform Auditor before corrective migration"
|
||||
|
||||
state = migrator.apply_tested_migration(('main', '0208_fix_system_auditor_migration'))
|
||||
RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
|
||||
assert RoleUserAssignment.objects.filter(
|
||||
user=user.id, role_definition__name='Platform Auditor'
|
||||
).exists(), "Corrective migration should have backfilled Platform Auditor"
|
||||
|
||||
def test_fix_system_auditor_migration_no_duplicates(self, migrator):
|
||||
"""
|
||||
Migration 0208 must be idempotent — users who already have
|
||||
Platform Auditor should not receive a duplicate assignment.
|
||||
"""
|
||||
state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more'))
|
||||
User = state.apps.get_model('auth', 'User')
|
||||
Role = state.apps.get_model('main', 'Role')
|
||||
|
||||
user = User.objects.create(username='already-auditor')
|
||||
system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'})
|
||||
system_auditor_role.members.add(user)
|
||||
|
||||
state = migrator.apply_tested_migration(('main', '0192_custom_roles'))
|
||||
RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
|
||||
# Confirm the user already has the assignment (from the fixed 0192)
|
||||
assert RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Platform Auditor').exists()
|
||||
|
||||
state = migrator.apply_tested_migration(('main', '0208_fix_system_auditor_migration'))
|
||||
RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment')
|
||||
|
||||
assert (
|
||||
RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Platform Auditor').count() == 1
|
||||
), "Should have exactly one Platform Auditor assignment, not duplicates"
|
||||
|
||||
Reference in New Issue
Block a user