External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-06-21 06:37:45 -02:30
Code Issues Packages Projects Releases Wiki Activity
Files
c1bd2eb3387948e11dc638258ba33c7fd1b5d19b
awx/awx/main/tests/functional/api/test_organizations.py
Dirk Julich c1bd2eb338 [AAP-72817] Fix cartesian product in organization user/admin count queries (#16501) 
* Fix cartesian product in organization user/admin count queries

The organizations list and detail endpoints annotated each org with user and admin counts using two Count() calls that traverse the Role.members M2M. Django generated two LEFT JOINs on the same through table, crossing every member row with every admin row before COUNT(DISTINCT) reduced the product.

At scale (2,617 members × 46,233 admins) this produced 120M intermediate rows and 96-second query times, causing 504 timeouts.

Replace with independent Subquery expressions that each query main_rbac_roles_members separately - no cross product.

Fixes: AAP-72817
Fixes: AAP-72480

* Fix variable names which do not meet coding standards

* Fix formatting inconsistency in organization detail subquery annotation

Break the long .annotate() line across multiple lines to match the style used in mixin.py.

* Rewrite org count subqueries to use DAB RBAC models

Replace old RBAC Role.members.through subqueries with
RoleUserAssignment-based correlated subqueries, querying
managed RoleDefinitions ('Organization Member' / 'Organization Admin')
directly. This aligns with the DAB RBAC migration direction and
eliminates dependency on the deprecated ImplicitRoleField M2M tables
for these counts.

Update test fixtures to use RoleDefinition.give_permission() and
add setup_managed_roles where needed.

* Fix collection tests: set up managed role definitions

The DAB RBAC migration to use RoleUserAssignment subqueries in
organization views requires managed role definitions (Organization
Member, Organization Admin) to exist in the test database.

Add an autouse fixture to the collection test conftest that calls
setup_managed_role_definitions() before each test.

* Add setup_managed_roles fixture to functional tests hitting org views

Tests that hit organization list/detail views now require the
setup_managed_roles fixture to pre-create the Organization Member
and Organization Admin RoleDefinition objects used by the DAB RBAC
subqueries.

* Revert setup_managed_roles from ext_auditor tests

The setup_managed_roles fixture conflicts with the ext_auditor_rd
fixture by deleting the Alien Auditor role definition. These tests
don't need it — the defensive view code handles missing role
definitions gracefully.

* Handle missing Organization Member/Admin role definitions gracefully

Use filter().first() instead of get() for RoleDefinition lookups in
organization list and detail views. Returns 0 for user/admin counts
when role definitions are not yet created, preventing 500 errors in
environments where post_migrate signals haven't run.

* Cast OuterRef('pk') to TextField for RoleUserAssignment.object_id comparison

RoleUserAssignment.object_id is a TextField, but OuterRef('pk') on
Organization produces an integer. PostgreSQL strictly rejects text = integer
comparisons. Use Cast() to explicitly convert the PK to text.

---------
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-06-18 18:35:22 +02:00

350 lines
15 KiB
Python
Raw Blame History

# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.
import pytest
# AWX
from awx.main.models import ProjectUpdate, CredentialType, Credential
from awx.api.versioning import reverse
@pytest.fixture
def create_job_factory(job_factory, project):
def fn(status='running'):
j = job_factory()
j.status = status
j.project = project
j.save()
return j
return fn
@pytest.fixture
def create_project_update_factory(organization, project):
def fn(status='running'):
pu = ProjectUpdate(project=project)
pu.status = status
pu.organization = organization
pu.save()
return pu
return fn
@pytest.fixture
def organization_jobs_successful(create_job_factory, create_project_update_factory):
return [create_job_factory(status='successful') for i in range(0, 2)] + [create_project_update_factory(status='successful') for i in range(0, 2)]
@pytest.fixture
def organization_jobs_running(create_job_factory, create_project_update_factory):
return [create_job_factory(status='running') for i in range(0, 2)] + [create_project_update_factory(status='running') for i in range(0, 2)]
@pytest.mark.django_db
def test_organization_list_access_tests(options, head, get, admin, alice):
options(reverse('api:organization_list'), user=admin, expect=200)
head(reverse('api:organization_list'), user=admin, expect=200)
get(reverse('api:organization_list'), user=admin, expect=200)
options(reverse('api:organization_list'), user=alice, expect=200)
head(reverse('api:organization_list'), user=alice, expect=200)
get(reverse('api:organization_list'), user=alice, expect=200)
options(reverse('api:organization_list'), user=None, expect=401)
head(reverse('api:organization_list'), user=None, expect=401)
get(reverse('api:organization_list'), user=None, expect=401)
@pytest.mark.django_db
def test_organization_access_tests(organization, get, admin, alice, bob, setup_managed_roles):
organization.member_role.members.add(alice)
get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=admin, expect=200)
get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=alice, expect=200)
get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=bob, expect=403)
get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=None, expect=401)
@pytest.mark.django_db
def test_organization_list_integrity(organization, get, admin, alice, setup_managed_roles):
res = get(reverse('api:organization_list'), user=admin)
for field in ['id', 'url', 'name', 'description', 'created']:
assert field in res.data['results'][0]
@pytest.mark.django_db
def test_organization_list_order_integrity(organizations, get, admin, setup_managed_roles):
# check that the order of the organization list retains integrity.
orgs = organizations(4)
org_ids = {}
for x in range(3):
res = get(reverse('api:organization_list'), user=admin).data['results']
org_ids[x] = [org['id'] for org in res]
assert org_ids[0] == org_ids[1] == org_ids[2] == [orgs[0].id, orgs[1].id, orgs[2].id, orgs[3].id]
@pytest.mark.django_db
def test_organization_list_visibility(organizations, get, admin, alice, setup_managed_roles):
orgs = organizations(2)
res = get(reverse('api:organization_list'), user=admin)
assert res.data['count'] == 2
assert len(res.data['results']) == 2
res = get(reverse('api:organization_list'), user=alice)
assert res.data['count'] == 0
orgs[1].member_role.members.add(alice)
res = get(reverse('api:organization_list'), user=alice)
assert res.data['count'] == 1
assert len(res.data['results']) == 1
assert res.data['results'][0]['id'] == orgs[1].id
@pytest.mark.django_db
def test_organization_project_list(organization, project_factory, get, alice, bob, rando):
prj1 = project_factory('project-one')
project_factory('project-two')
organization.admin_role.members.add(alice)
organization.member_role.members.add(bob)
prj1.use_role.members.add(bob)
assert get(reverse('api:organization_projects_list', kwargs={'pk': organization.id}), user=alice).data['count'] == 2
assert get(reverse('api:organization_projects_list', kwargs={'pk': organization.id}), user=bob).data['count'] == 1
assert get(reverse('api:organization_projects_list', kwargs={'pk': organization.id}), user=rando).status_code == 403
@pytest.mark.django_db
def test_organization_user_list(organization, get, admin, alice, bob):
organization.admin_role.members.add(alice)
organization.member_role.members.add(alice)
organization.member_role.members.add(bob)
assert get(reverse('api:organization_users_list', kwargs={'pk': organization.id}), user=admin).data['count'] == 2
assert get(reverse('api:organization_users_list', kwargs={'pk': organization.id}), user=alice).data['count'] == 2
assert get(reverse('api:organization_users_list', kwargs={'pk': organization.id}), user=bob).data['count'] == 2
assert get(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), user=admin).data['count'] == 1
assert get(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), user=alice).data['count'] == 1
assert get(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), user=bob).data['count'] == 1
@pytest.mark.django_db
def test_organization_inventory_list(organization, inventory_factory, get, alice, bob, rando):
inv1 = inventory_factory('inventory-one')
inventory_factory('inventory-two')
organization.admin_role.members.add(alice)
organization.member_role.members.add(bob)
inv1.use_role.members.add(bob)
assert get(reverse('api:organization_inventories_list', kwargs={'pk': organization.id}), user=alice).data['count'] == 2
assert get(reverse('api:organization_inventories_list', kwargs={'pk': organization.id}), user=bob).data['count'] == 1
get(reverse('api:organization_inventories_list', kwargs={'pk': organization.id}), user=rando, expect=403)
@pytest.mark.django_db
def test_organization_inventory_list_order_integrity(organization, admin, inventory_factory, get):
inv1 = inventory_factory('inventory')
inv2 = inventory_factory('inventory2')
inv3 = inventory_factory('inventory3')
inv_ids = {}
for x in range(3):
res = get(reverse('api:organization_inventories_list', kwargs={'pk': organization.id}), user=admin).data['results']
inv_ids[x] = [inv['id'] for inv in res]
assert inv_ids[0] == inv_ids[1] == inv_ids[2] == [inv1.id, inv2.id, inv3.id]
@pytest.mark.django_db
def test_create_organization(post, admin, alice, setup_managed_roles):
new_org = {'name': 'new org', 'description': 'my description'}
res = post(reverse('api:organization_list'), new_org, user=admin, expect=201)
assert res.data['name'] == new_org['name']
res = post(reverse('api:organization_list'), new_org, user=admin, expect=400)
@pytest.mark.django_db
def test_create_organization_xfail(post, alice):
new_org = {'name': 'new org', 'description': 'my description'}
post(reverse('api:organization_list'), new_org, user=alice, expect=403)
@pytest.mark.django_db
def test_add_user_to_organization(post, organization, alice, bob):
organization.admin_role.members.add(alice)
post(reverse('api:organization_users_list', kwargs={'pk': organization.id}), {'id': bob.id}, user=alice, expect=204)
assert bob in organization.member_role
post(reverse('api:organization_users_list', kwargs={'pk': organization.id}), {'id': bob.id, 'disassociate': True}, user=alice, expect=204)
assert bob not in organization.member_role
@pytest.mark.django_db
def test_add_user_to_organization_xfail(post, organization, alice, bob):
organization.member_role.members.add(alice)
post(reverse('api:organization_users_list', kwargs={'pk': organization.id}), {'id': bob.id}, user=alice, expect=403)
@pytest.mark.django_db
def test_add_admin_to_organization(post, organization, alice, bob):
organization.admin_role.members.add(alice)
post(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), {'id': bob.id}, user=alice, expect=204)
assert bob in organization.admin_role
assert bob in organization.member_role
post(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), {'id': bob.id, 'disassociate': True}, user=alice, expect=204)
assert bob not in organization.admin_role
assert bob not in organization.member_role
@pytest.mark.django_db
def test_add_admin_to_organization_xfail(post, organization, alice, bob):
organization.member_role.members.add(alice)
post(reverse('api:organization_admins_list', kwargs={'pk': organization.id}), {'id': bob.id}, user=alice, expect=403)
@pytest.mark.django_db
def test_update_organization(get, put, organization, alice, bob, setup_managed_roles):
organization.admin_role.members.add(alice)
data = get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=alice, expect=200).data
data['description'] = 'hi'
put(reverse('api:organization_detail', kwargs={'pk': organization.id}), data, user=alice, expect=200)
organization.refresh_from_db()
assert organization.description == 'hi'
data['description'] = 'bye'
put(reverse('api:organization_detail', kwargs={'pk': organization.id}), data, user=bob, expect=403)
@pytest.mark.django_db
def test_update_organization_max_hosts(get, put, organization, admin, alice, bob, setup_managed_roles):
# Admin users can get and update max_hosts
data = get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=admin, expect=200).data
assert organization.max_hosts == 0
data['max_hosts'] = 3
put(reverse('api:organization_detail', kwargs={'pk': organization.id}), data, user=admin, expect=200)
organization.refresh_from_db()
assert organization.max_hosts == 3
# Organization admins can get the data and can update other fields, but not max_hosts
organization.admin_role.members.add(alice)
data = get(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=alice, expect=200).data
data['max_hosts'] = 5
put(reverse('api:organization_detail', kwargs={'pk': organization.id}), data, user=alice, expect=400)
organization.refresh_from_db()
assert organization.max_hosts == 3
# Ordinary users shouldn't be able to update either.
put(reverse('api:organization_detail', kwargs={'pk': organization.id}), data, user=bob, expect=403)
organization.refresh_from_db()
assert organization.max_hosts == 3
@pytest.mark.django_db
def test_delete_organization(delete, organization, admin):
delete(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=admin, expect=204)
@pytest.mark.django_db
def test_delete_organization2(delete, organization, alice):
organization.admin_role.members.add(alice)
delete(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=alice, expect=204)
@pytest.mark.django_db
def test_delete_organization_xfail1(delete, organization, alice):
organization.member_role.members.add(alice)
delete(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=alice, expect=403)
@pytest.mark.django_db
def test_delete_organization_xfail2(delete, organization):
delete(reverse('api:organization_detail', kwargs={'pk': organization.id}), user=None, expect=401)
@pytest.mark.django_db
def test_organization_delete(delete, admin, organization, organization_jobs_successful):
url = reverse('api:organization_detail', kwargs={'pk': organization.id})
delete(url, None, user=admin, expect=204)
@pytest.mark.django_db
def test_organization_delete_with_active_jobs(delete, admin, organization, organization_jobs_running):
def sort_keys(x):
return (x['type'], str(x['id']))
url = reverse('api:organization_detail', kwargs={'pk': organization.id})
resp = delete(url, None, user=admin, expect=409)
expect_transformed = [dict(id=j.id, type=j.model_to_str()) for j in organization_jobs_running]
resp_sorted = sorted(resp.data['active_jobs'], key=sort_keys)
expect_sorted = sorted(expect_transformed, key=sort_keys)
assert resp.data['error'] == u"Resource is being used by running jobs."
assert resp_sorted == expect_sorted
@pytest.mark.django_db
def test_galaxy_credential_association_forbidden(alice, organization, post):
galaxy = CredentialType.defaults['galaxy_api_token']()
galaxy.save()
cred = Credential.objects.create(credential_type=galaxy, name='Public Galaxy', organization=organization, inputs={'url': 'https://galaxy.ansible.com/'})
url = reverse('api:organization_galaxy_credentials_list', kwargs={'pk': organization.id})
post(url, {'associate': True, 'id': cred.pk}, user=alice, expect=403)
@pytest.mark.django_db
def test_galaxy_credential_type_enforcement(admin, organization, post):
ssh = CredentialType.defaults['ssh']()
ssh.save()
cred = Credential.objects.create(
credential_type=ssh,
name='SSH Credential',
organization=organization,
)
url = reverse('api:organization_galaxy_credentials_list', kwargs={'pk': organization.id})
resp = post(url, {'associate': True, 'id': cred.pk}, user=admin, expect=400)
assert resp.data['msg'] == 'Credential must be a Galaxy credential, not Machine.'
@pytest.mark.django_db
def test_galaxy_credential_association(alice, admin, organization, post, get):
galaxy = CredentialType.defaults['galaxy_api_token']()
galaxy.save()
for i in range(5):
cred = Credential.objects.create(
credential_type=galaxy, name=f'Public Galaxy {i + 1}', organization=organization, inputs={'url': 'https://galaxy.ansible.com/'}
)
url = reverse('api:organization_galaxy_credentials_list', kwargs={'pk': organization.id})
post(url, {'associate': True, 'id': cred.pk}, user=admin, expect=204)
resp = get(url, user=admin)
assert [cred['name'] for cred in resp.data['results']] == [
'Public Galaxy 1',
'Public Galaxy 2',
'Public Galaxy 3',
'Public Galaxy 4',
'Public Galaxy 5',
]
post(url, {'disassociate': True, 'id': Credential.objects.get(name='Public Galaxy 3').pk}, user=admin, expect=204)
resp = get(url, user=admin)
assert [cred['name'] for cred in resp.data['results']] == [
'Public Galaxy 1',
'Public Galaxy 2',
'Public Galaxy 4',
'Public Galaxy 5',
]
@pytest.mark.django_db
def test_org_admin_credential_count(org_admin, admin, organization, post, get):
galaxy = CredentialType.defaults['galaxy_api_token']()
galaxy.save()
for i in range(3):
cred = Credential.objects.create(credential_type=galaxy, name=f'test_{i}', inputs={'url': 'https://galaxy.ansible.com/'})
url = reverse('api:organization_galaxy_credentials_list', kwargs={'pk': organization.pk})
post(url, {'associate': True, 'id': cred.pk}, user=admin, expect=204)
# org admin should see all associated galaxy credentials
resp = get(url, user=org_admin)
assert resp.data['count'] == 3
# removing one to validate new count
post(url, {'disassociate': True, 'id': Credential.objects.get(name='test_1').pk}, user=admin, expect=204)
resp_new = get(url, user=org_admin)
assert resp_new.data['count'] == 2
Reference in New Issue View Git Blame Copy Permalink