Update quay.io/aap-ci/tekton-catalog/pipeline/test/aap-api-tests:0.1 Docker digest to 7a49ac8

Signed-off-by: red-hat-konflux-kflux-prd-rh02 <190377777+red-hat-konflux-kflux-prd-rh02[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -260,7 +260,7 @@ jobs:
         continue-on-error: true
         run: |
           set +e
-          timeout 15m bash -elc '
+          timeout 20m bash -elc '
             python -m pip install -r molecule/requirements.txt
             python -m pip install PyYAML # for awx/tools/scripts/rewrite-awx-operator-requirements.py
             $(realpath ../awx/tools/scripts/rewrite-awx-operator-requirements.py) molecule/requirements.yml $(realpath ../awx)

.tekton/run-atf-tests-pull-request.yaml

@@ -27,7 +27,7 @@ spec:
       - name: name
         value: aap-api-tests
       - name: bundle
-        value: quay.io/aap-ci/tekton-catalog/pipeline/test/aap-api-tests:0.1@sha256:0c1621395487e9305fb7652feb6d65071018953a199b991dcf520bd50c0b05ef
+        value: quay.io/aap-ci/tekton-catalog/pipeline/test/aap-api-tests:0.1@sha256:7a49ac8f1b6178775345ad369e04a6d51947a63c4cf6e472e5be9f920aaae038
       - name: kind
         value: pipeline
       - name: secret

awx/api/pagination.py

@@ -6,12 +6,15 @@ from collections import OrderedDict
 # Django REST Framework
 from django.conf import settings
 from django.core.paginator import Paginator as DjangoPaginator
+from django.utils.functional import cached_property
 from rest_framework import pagination
 from rest_framework.response import Response
 from rest_framework.utils.urls import replace_query_param
 from rest_framework.settings import api_settings
 from django.utils.translation import gettext_lazy as _
 
+from awx.main.models import ActivityStream, UnifiedJob
+
 
 class DisabledPaginator(DjangoPaginator):
     @property
@@ -23,6 +26,22 @@ class DisabledPaginator(DjangoPaginator):
         return 200
 
 
+class ActivityStreamPaginator(DjangoPaginator):
+    """Use unfiltered table count for activity stream pagination (AAP-83773)."""
+
+    @cached_property
+    def count(self):
+        return ActivityStream.objects.count()
+
+
+class UnifiedJobPaginator(DjangoPaginator):
+    """Use unfiltered table count for unified job pagination."""
+
+    @cached_property
+    def count(self):
+        return UnifiedJob.objects.count()
+
+
 class Pagination(pagination.PageNumberPagination):
     page_size_query_param = 'page_size'
     max_page_size = settings.MAX_PAGE_SIZE
@@ -57,12 +76,13 @@ class Pagination(pagination.PageNumberPagination):
 
     def paginate_queryset(self, queryset, request, **kwargs):
         self.count_disabled = 'count_disabled' in request.query_params
+        original_paginator = self.django_paginator_class
         try:
             if self.count_disabled:
                 self.django_paginator_class = DisabledPaginator
             return super(Pagination, self).paginate_queryset(queryset, request, **kwargs)
         finally:
-            self.django_paginator_class = DjangoPaginator
+            self.django_paginator_class = original_paginator
 
     def get_paginated_response(self, data):
         if self.count_disabled:
@@ -70,6 +90,14 @@ class Pagination(pagination.PageNumberPagination):
         return super(Pagination, self).get_paginated_response(data)
 
 
+class ActivityStreamPagination(Pagination):
+    django_paginator_class = ActivityStreamPaginator
+
+
+class UnifiedJobPagination(Pagination):
+    django_paginator_class = UnifiedJobPaginator
+
+
 class LimitPagination(pagination.BasePagination):
     default_limit = api_settings.PAGE_SIZE
     limit_query_param = 'limit'

awx/api/serializers.py

@@ -961,14 +961,27 @@ class UnifiedJobSerializer(BaseSerializer):
 
 
 class UnifiedJobListSerializer(UnifiedJobSerializer):
+    OPTIONAL_EXCLUDE_FIELDS = frozenset({'artifacts', 'extra_vars'})
+
+    _ALWAYS_STRIPPED_FIELDS = frozenset({'job_args', 'job_cwd', 'job_env', 'result_traceback', 'event_processing_finished'})
+
     class Meta:
-        fields = ('*', '-job_args', '-job_cwd', '-job_env', '-result_traceback', '-event_processing_finished', '-artifacts')
+        fields = ('*', '-job_args', '-job_cwd', '-job_env', '-result_traceback', '-event_processing_finished')
+
+    def _requested_excludes(self):
+        request = self.context.get('request')
+        if request is None:
+            return frozenset()
+        raw = request.query_params.get('exclude', '')
+        requested = {name.strip() for name in raw.split(',') if name.strip()}
+        return frozenset(requested) & self.OPTIONAL_EXCLUDE_FIELDS
 
     def get_field_names(self, declared_fields, info):
         field_names = super(UnifiedJobListSerializer, self).get_field_names(declared_fields, info)
         # Meta multiple inheritance and -field_name options don't seem to be
         # taking effect above, so remove the undesired fields here.
-        return tuple(x for x in field_names if x not in ('job_args', 'job_cwd', 'job_env', 'result_traceback', 'event_processing_finished', 'artifacts'))
+        strip = self._ALWAYS_STRIPPED_FIELDS | self._requested_excludes()
+        return tuple(x for x in field_names if x not in strip)
 
     def get_types(self):
         if type(self) is UnifiedJobListSerializer:

awx/api/views/__init__.py

@@ -127,8 +127,9 @@ from awx.api.views.mixin import (
     RelatedJobsPreventDeleteMixin,
     UnifiedJobDeletionMixin,
     NoTruncateMixin,
+    UnifiedJobExcludeMixin,
 )
-from awx.api.pagination import UnifiedJobEventPagination
+from awx.api.pagination import ActivityStreamPagination, UnifiedJobEventPagination, UnifiedJobPagination
 from awx.main.utils import set_environ
 
 logger = logging.getLogger('awx.api.views')
@@ -1926,7 +1927,8 @@ class HostList(HostRelatedSearchMixin, ListCreateAPIView):
         if filter_string:
             filter_qs = SmartFilter.query_from_string(filter_string)
             qs &= filter_qs
-        return qs.distinct().with_latest_summary_id()
+            qs = qs.distinct()
+        return qs.with_latest_summary_id()
 
     def list(self, *args, **kwargs):
         try:
@@ -3850,7 +3852,7 @@ class SystemJobTemplateNotificationTemplatesSuccessList(SystemJobTemplateNotific
     resource_purpose = 'notification templates triggered on system job success'
 
 
-class JobList(ListAPIView):
+class JobList(UnifiedJobExcludeMixin, ListAPIView):
     model = models.Job
     serializer_class = serializers.JobListSerializer
     resource_purpose = 'jobs'
@@ -4567,10 +4569,11 @@ class UnifiedJobTemplateList(ListAPIView):
     resource_purpose = 'unified job templates'
 
 
-class UnifiedJobList(ListAPIView):
+class UnifiedJobList(UnifiedJobExcludeMixin, ListAPIView):
     model = models.UnifiedJob
     serializer_class = serializers.UnifiedJobListSerializer
     search_fields = ('description', 'name', 'job__playbook')
+    pagination_class = UnifiedJobPagination
     resource_purpose = 'unified jobs'
 
 
@@ -4816,6 +4819,7 @@ class ActivityStreamList(SimpleListAPIView):
     model = models.ActivityStream
     serializer_class = serializers.ActivityStreamSerializer
     search_fields = ('changes',)
+    pagination_class = ActivityStreamPagination
     resource_purpose = 'audit trail entries for tracking system changes'
 
     @extend_schema_if_available(

awx/api/views/mixin.py

@@ -4,8 +4,8 @@
 import dateutil
 import logging
 
-from django.db.models import Count, OuterRef, Subquery, TextField
-from django.db.models.functions import Cast, Coalesce
+from django.db.models import Count, Q, TextField
+from django.db.models.functions import Cast
 from django.db import transaction
 from django.shortcuts import get_object_or_404
 from django.utils.timezone import now
@@ -179,48 +179,37 @@ class OrganizationCountsMixin(object):
 
         db_results['projects'] = project_qs.values('organization').annotate(Count('organization')).order_by('organization')
 
-        member_rd = RoleDefinition.objects.filter(name='Organization Member').first()
-        admin_rd = RoleDefinition.objects.filter(name='Organization Admin').first()
-
-        if member_rd and admin_rd:
-
-            def assignment_count(rd):
-                return Coalesce(
-                    Subquery(
-                        RoleUserAssignment.objects.filter(
-                            object_id=Cast(OuterRef('pk'), output_field=TextField()),
-                            role_definition=rd,
-                        )
-                        .values('role_definition')
-                        .annotate(c=Count('pk'))
-                        .values('c')
-                    ),
-                    0,
-                )
-
-            db_results['users'] = org_qs.annotate(
-                users=assignment_count(member_rd),
-                admins=assignment_count(admin_rd),
-            ).values('id', 'users', 'admins')
-
         count_context = {}
         for org in org_id_list:
             org_id = org['id']
             count_context[org_id] = {'inventories': 0, 'teams': 0, 'users': 0, 'job_templates': 0, 'admins': 0, 'projects': 0}
 
         for res, count_qs in db_results.items():
-            if res == 'users':
-                org_reference = 'id'
-            else:
-                org_reference = 'organization'
             for entry in count_qs:
-                org_id = entry[org_reference]
+                org_id = entry['organization']
                 if org_id in count_context:
-                    if res == 'users':
-                        count_context[org_id]['admins'] = entry['admins']
-                        count_context[org_id]['users'] = entry['users']
-                        continue
-                    count_context[org_id][res] = entry['%s__count' % org_reference]
+                    count_context[org_id][res] = entry['organization__count']
+
+        member_rd = RoleDefinition.objects.filter(name='Organization Member').first()
+        admin_rd = RoleDefinition.objects.filter(name='Organization Admin').first()
+
+        if member_rd and admin_rd:
+            user_admin_counts = (
+                RoleUserAssignment.objects.filter(
+                    role_definition__in=[member_rd, admin_rd],
+                    object_id__in=org_qs.annotate(text_pk=Cast('pk', TextField())).values('text_pk'),
+                )
+                .values('object_id')
+                .annotate(
+                    users=Count('pk', filter=Q(role_definition=member_rd)),
+                    admins=Count('pk', filter=Q(role_definition=admin_rd)),
+                )
+            )
+            for entry in user_admin_counts:
+                org_id = int(entry['object_id'])
+                if org_id in count_context:
+                    count_context[org_id]['users'] = entry['users']
+                    count_context[org_id]['admins'] = entry['admins']
 
         full_context['related_field_counts'] = count_context
 
@@ -233,3 +222,9 @@ class NoTruncateMixin(object):
         if self.request.query_params.get('no_truncate'):
             context.update(no_truncate=True)
         return context
+
+
+class UnifiedJobExcludeMixin(object):
+    # Reserve the name 'exclude' so we can use it as a query param. Otherwise, the rest-filters backend
+    # would treat it as a model field lookup.
+    rest_filters_reserved_names = ('exclude',)

awx/api/views/organization.py

@@ -5,8 +5,7 @@
 import logging
 
 # Django
-from django.db.models import Count, OuterRef, Subquery, TextField
-from django.db.models.functions import Cast, Coalesce
+from django.db.models import Count, Q
 from django.contrib.contenttypes.models import ContentType
 from django.utils.translation import gettext_lazy as _
 
@@ -83,37 +82,17 @@ class OrganizationDetail(RelatedJobsPreventDeleteMixin, RetrieveUpdateDestroyAPI
         admin_rd = RoleDefinition.objects.filter(name='Organization Admin').first()
 
         if member_rd and admin_rd:
-
-            def assignment_count(rd):
-                return Coalesce(
-                    Subquery(
-                        RoleUserAssignment.objects.filter(
-                            object_id=Cast(OuterRef('pk'), output_field=TextField()),
-                            role_definition=rd,
-                        )
-                        .values('role_definition')
-                        .annotate(c=Count('pk'))
-                        .values('c')
-                    ),
-                    0,
-                )
-
-            direct_counts = (
-                Organization.objects.filter(id=org_id)
-                .annotate(
-                    users=assignment_count(member_rd),
-                    admins=assignment_count(admin_rd),
-                )
-                .values('users', 'admins')
+            counts = RoleUserAssignment.objects.filter(
+                role_definition__in=[member_rd, admin_rd],
+                object_id=str(org_id),
+            ).aggregate(
+                users=Count('pk', filter=Q(role_definition=member_rd)),
+                admins=Count('pk', filter=Q(role_definition=admin_rd)),
             )
-
-            if direct_counts:
-                org_counts = direct_counts[0]
+            org_counts.update(counts)
         else:
-            org_counts = {'users': 0, 'admins': 0}
+            org_counts.update({'users': 0, 'admins': 0})
 
-        if not org_counts:
-            return full_context
         org_counts['inventories'] = Inventory.accessible_objects(**access_kwargs).filter(organization__id=org_id).count()
         org_counts['teams'] = Team.accessible_objects(**access_kwargs).filter(organization__id=org_id).count()
         org_counts['projects'] = Project.accessible_objects(**access_kwargs).filter(organization__id=org_id).count()

awx/main/access.py

@@ -20,6 +20,7 @@ from rest_framework.exceptions import ParseError, PermissionDenied
 # django-ansible-base
 from ansible_base.lib.utils.validation import to_python_boolean
 from ansible_base.rbac.models import RoleEvaluation
+from ansible_base.rbac.policies import visible_users
 from ansible_base.rbac import permission_registry
 
 # AWX
@@ -643,6 +644,8 @@ class UserAccess(BaseAccess):
             Organization.access_qs(self.user, 'change').exists() or Organization.access_qs(self.user, 'audit').exists()
         ):
             qs = User.objects.all()
+        elif settings.ANSIBLE_BASE_ROLE_SYSTEM_ACTIVATED:
+            qs = visible_users(self.user)
         else:
             qs = (
                 User.objects.filter(pk__in=Organization.access_qs(self.user, 'view').values('member_role__members'))
@@ -706,12 +709,13 @@ class UserAccess(BaseAccess):
                 # in these cases only superusers can modify orphan users
                 return False
             if settings.ANSIBLE_BASE_ROLE_SYSTEM_ACTIVATED:
-                # Permission granted if the user has all permissions that the target user has
                 target_perms = set(
-                    RoleEvaluation.objects.filter(role__in=obj.has_roles.all()).values_list('object_id', 'content_type_id', 'codename').distinct()
+                    RoleEvaluation.objects.filter(**RoleEvaluation._actor_role_filter(obj)).values_list('object_id', 'content_type_id', 'codename').distinct()
                 )
                 user_perms = set(
-                    RoleEvaluation.objects.filter(role__in=self.user.has_roles.all()).values_list('object_id', 'content_type_id', 'codename').distinct()
+                    RoleEvaluation.objects.filter(**RoleEvaluation._actor_role_filter(self.user))
+                    .values_list('object_id', 'content_type_id', 'codename')
+                    .distinct()
                 )
                 return not (target_perms - user_perms)
             return not obj.roles.all().exclude(ancestors__in=self.user.roles.all()).exists()
@@ -1228,9 +1232,11 @@ class TeamAccess(BaseAccess):
             Organization.access_qs(self.user, 'change').exists() or Organization.access_qs(self.user, 'audit').exists()
         ):
             return self.model.objects.all()
-        return self.model.objects.filter(
-            Q(organization__in=Organization.accessible_pk_qs(self.user, 'member_role')) | Q(pk__in=self.model.accessible_pk_qs(self.user, 'read_role'))
+        org_member_teams = (
+            self.model.objects.filter(organization__in=Organization.accessible_pk_qs(self.user, 'member_role')).order_by().values_list('pk', flat=True)
         )
+        direct_read_teams = self.model.objects.filter(pk__in=self.model.accessible_pk_qs(self.user, 'read_role')).order_by().values_list('pk', flat=True)
+        return self.model.objects.filter(pk__in=org_member_teams.union(direct_read_teams))
 
     @check_superuser
     def can_add(self, data):
@@ -1665,11 +1671,11 @@ class JobAccess(BaseAccess):
     def filtered_queryset(self):
         qs = self.model.objects
 
-        qs_jt = qs.filter(job_template__in=JobTemplate.access_qs(self.user, 'view'))
-
-        org_access_qs = Organization.objects.filter(Q(admin_role__members=self.user) | Q(auditor_role__members=self.user))
+        org_access_qs = Organization.objects.filter(
+            Q(pk__in=Organization.access_ids_qs(self.user, 'change')) | Q(pk__in=Organization.access_ids_qs(self.user, 'audit_organization'))
+        )
         if not org_access_qs.exists():
-            return qs_jt
+            return qs.filter(job_template__in=JobTemplate.access_qs(self.user, 'view'))
 
         return qs.filter(Q(job_template__in=JobTemplate.access_qs(self.user, 'view')) | Q(organization__in=org_access_qs)).distinct()
 
@@ -2309,7 +2315,7 @@ class JobHostSummaryAccess(BaseAccess):
 
 class JobEventAccess(BaseAccess):
     """
-    I can see job event records whenever I can read both job and host.
+    I can see job event records whenever I can read the job or the host.
     """
 
     model = JobEvent
@@ -2320,8 +2326,8 @@ class JobEventAccess(BaseAccess):
 
     def filtered_queryset(self):
         return self.model.objects.filter(
-            Q(host__inventory__in=Inventory.accessible_pk_qs(self.user, 'read_role'))
-            | Q(job__job_template__in=JobTemplate.accessible_pk_qs(self.user, 'read_role'))
+            Q(host_id__in=Host.objects.filter(inventory__in=Inventory.access_ids_qs(self.user, 'view')).values('pk'))
+            | Q(job_id__in=Job.objects.filter(job_template__in=JobTemplate.access_ids_qs(self.user, 'view')).values('pk'))
         )
 
     def can_add(self, data):
@@ -2451,7 +2457,11 @@ class UnifiedJobTemplateAccess(BaseAccess):
     def filtered_queryset(self):
         return self.model.objects.filter(
             Q(pk__in=self.model.accessible_pk_qs(self.user, 'read_role'))
-            | Q(inventorysource__inventory__id__in=Inventory._accessible_pk_qs(Inventory, self.user, 'read_role'))
+            | Q(
+                pk__in=InventorySource.objects.filter(
+                    inventory__id__in=Inventory.access_ids_qs(self.user, 'view'),
+                ).values('unifiedjobtemplate_ptr_id')
+            )
         )
 
     def can_start(self, obj, validate_license=True):
@@ -2497,14 +2507,39 @@ class UnifiedJobAccess(BaseAccess):
     # )
 
     def filtered_queryset(self):
-        inv_pk_qs = Inventory._accessible_pk_qs(Inventory, self.user, 'read_role')
-        qs = self.model.objects.filter(
-            Q(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
-            | Q(inventoryupdate__inventory_source__inventory__id__in=inv_pk_qs)
-            | Q(adhoccommand__inventory__id__in=inv_pk_qs)
-            | Q(organization__in=Organization.accessible_pk_qs(self.user, 'auditor_role'))
+        inv_pk_qs = Inventory.access_ids_qs(self.user, 'view')
+
+        by_template = (
+            self.model.objects.filter(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
+            .order_by()
+            .values_list('pk', flat=True)
         )
-        return qs
+
+        by_inventory_update = (
+            InventoryUpdate.objects.filter(
+                inventory_source__inventory__id__in=inv_pk_qs,
+            )
+            .order_by()
+            .values_list('pk', flat=True)
+        )
+
+        by_adhoc = (
+            AdHocCommand.objects.filter(
+                inventory__id__in=inv_pk_qs,
+            )
+            .order_by()
+            .values_list('pk', flat=True)
+        )
+
+        by_org_auditor = (
+            self.model.objects.filter(
+                organization__in=Organization.access_ids_qs(self.user, 'audit_organization'),
+            )
+            .order_by()
+            .values_list('pk', flat=True)
+        )
+
+        return self.model.objects.filter(pk__in=by_template.union(by_inventory_update, by_adhoc, by_org_auditor))
 
     def get_queryset(self):
         return super(UnifiedJobAccess, self).get_queryset().filter(workflowapproval__isnull=True)
@@ -2622,9 +2657,13 @@ class LabelAccess(BaseAccess):
 
     def filtered_queryset(self):
         return self.model.objects.filter(
-            Q(organization__in=Organization.accessible_pk_qs(self.user, 'read_role'))
-            | Q(unifiedjobtemplate_labels__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
-        ).distinct()
+            Q(organization__in=Organization.access_ids_qs(self.user, 'view'))
+            | Q(
+                pk__in=UnifiedJobTemplate.labels.through.objects.filter(
+                    unifiedjobtemplate_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'),
+                ).values('label_id')
+            )
+        )
 
     @check_superuser
     def can_add(self, data):
@@ -2698,54 +2737,73 @@ class ActivityStreamAccess(BaseAccess):
         # 'job_template', 'job', 'project', 'project_update', 'workflow_job',
         # 'inventory_source', 'workflow_job_template'
 
-        q = Q(user=self.user)
-        inventory_set = Inventory.accessible_pk_qs(self.user, 'read_role')
-        if inventory_set:
+        AS = ActivityStream
+
+        q = Q(pk__in=AS.user.through.objects.filter(user=self.user).values('activitystream_id'))
+
+        inventory_set = Inventory.access_ids_qs(self.user, 'view')
+        if inventory_set.exists():
             q |= (
-                Q(ad_hoc_command__inventory__in=inventory_set)
-                | Q(inventory__in=inventory_set)
-                | Q(host__inventory__in=inventory_set)
-                | Q(group__inventory__in=inventory_set)
-                | Q(inventory_source__inventory__in=inventory_set)
-                | Q(inventory_update__inventory_source__inventory__in=inventory_set)
+                Q(pk__in=AS.ad_hoc_command.through.objects.filter(adhoccommand__inventory__in=inventory_set).values('activitystream_id'))
+                | Q(pk__in=AS.inventory.through.objects.filter(inventory__in=inventory_set).values('activitystream_id'))
+                | Q(pk__in=AS.host.through.objects.filter(host__inventory__in=inventory_set).values('activitystream_id'))
+                | Q(pk__in=AS.group.through.objects.filter(group__inventory__in=inventory_set).values('activitystream_id'))
+                | Q(pk__in=AS.inventory_source.through.objects.filter(inventorysource__inventory__in=inventory_set).values('activitystream_id'))
+                | Q(
+                    pk__in=AS.inventory_update.through.objects.filter(inventoryupdate__inventory_source__inventory__in=inventory_set).values(
+                        'activitystream_id'
+                    )
+                )
             )
 
-        credential_set = Credential.accessible_pk_qs(self.user, 'read_role')
-        if credential_set:
-            q |= Q(credential__in=credential_set)
+        credential_set = Credential.access_ids_qs(self.user, 'view')
+        if credential_set.exists():
+            q |= Q(pk__in=AS.credential.through.objects.filter(credential__in=credential_set).values('activitystream_id'))
 
         auditing_orgs = (Organization.access_qs(self.user, 'change') | Organization.access_qs(self.user, 'audit')).distinct().values_list('id', flat=True)
-        if auditing_orgs:
+        if auditing_orgs.exists():
             q |= (
-                Q(user__in=auditing_orgs.values('member_role__members'))
-                | Q(organization__in=auditing_orgs)
-                | Q(notification_template__organization__in=auditing_orgs)
-                | Q(notification__notification_template__organization__in=auditing_orgs)
-                | Q(label__organization__in=auditing_orgs)
-                | Q(role__in=Role.visible_roles(self.user) if auditing_orgs else [])
+                Q(pk__in=AS.user.through.objects.filter(user__in=auditing_orgs.values('member_role__members')).values('activitystream_id'))
+                | Q(pk__in=AS.organization.through.objects.filter(organization__in=auditing_orgs).values('activitystream_id'))
+                | Q(pk__in=AS.notification_template.through.objects.filter(notificationtemplate__organization__in=auditing_orgs).values('activitystream_id'))
+                | Q(
+                    pk__in=AS.notification.through.objects.filter(notification__notification_template__organization__in=auditing_orgs).values(
+                        'activitystream_id'
+                    )
+                )
+                | Q(pk__in=AS.label.through.objects.filter(label__organization__in=auditing_orgs).values('activitystream_id'))
+                | Q(pk__in=AS.role.through.objects.filter(role__in=Role.visible_roles(self.user)).values('activitystream_id'))
             )
 
-        project_set = Project.accessible_pk_qs(self.user, 'read_role')
-        if project_set:
-            q |= Q(project__in=project_set) | Q(project_update__project__in=project_set)
-
-        jt_set = JobTemplate.accessible_pk_qs(self.user, 'read_role')
-        if jt_set:
-            q |= Q(job_template__in=jt_set) | Q(job__job_template__in=jt_set)
-
-        wfjt_set = WorkflowJobTemplate.accessible_pk_qs(self.user, 'read_role')
-        if wfjt_set:
-            q |= (
-                Q(workflow_job_template__in=wfjt_set)
-                | Q(workflow_job_template_node__workflow_job_template__in=wfjt_set)
-                | Q(workflow_job__workflow_job_template__in=wfjt_set)
+        project_set = Project.access_ids_qs(self.user, 'view')
+        if project_set.exists():
+            q |= Q(pk__in=AS.project.through.objects.filter(project__in=project_set).values('activitystream_id')) | Q(
+                pk__in=AS.project_update.through.objects.filter(projectupdate__project__in=project_set).values('activitystream_id')
             )
 
-        team_set = Team.accessible_pk_qs(self.user, 'read_role')
-        if team_set:
-            q |= Q(team__in=team_set)
+        jt_set = JobTemplate.access_ids_qs(self.user, 'view')
+        if jt_set.exists():
+            q |= Q(pk__in=AS.job_template.through.objects.filter(jobtemplate__in=jt_set).values('activitystream_id')) | Q(
+                pk__in=AS.job.through.objects.filter(job__job_template__in=jt_set).values('activitystream_id')
+            )
 
-        return qs.filter(q).distinct()
+        wfjt_set = WorkflowJobTemplate.access_ids_qs(self.user, 'view')
+        if wfjt_set.exists():
+            q |= (
+                Q(pk__in=AS.workflow_job_template.through.objects.filter(workflowjobtemplate__in=wfjt_set).values('activitystream_id'))
+                | Q(
+                    pk__in=AS.workflow_job_template_node.through.objects.filter(workflowjobtemplatenode__workflow_job_template__in=wfjt_set).values(
+                        'activitystream_id'
+                    )
+                )
+                | Q(pk__in=AS.workflow_job.through.objects.filter(workflowjob__workflow_job_template__in=wfjt_set).values('activitystream_id'))
+            )
+
+        team_set = Team.access_ids_qs(self.user, 'view')
+        if team_set.exists():
+            q |= Q(pk__in=AS.team.through.objects.filter(team__in=team_set).values('activitystream_id'))
+
+        return qs.filter(q)
 
     def can_add(self, data):
         return False

awx/main/apps.py

@@ -67,5 +67,28 @@ class MainConfig(AppConfig):
         super().ready()
 
         self.configure_dispatcherd()
+
+        from ansible_base.rbac.triggers import dab_post_migrate
+
+        dab_post_migrate.connect(self._sync_managed_role_definitions, dispatch_uid='awx-sync-managed-role-definitions')
+
         self.load_named_url_feature()
         pre_migrate.connect(self.check_db_requirement, sender=self)
+
+    @staticmethod
+    def _sync_managed_role_definitions(sender, **kwargs):
+        from django.apps import apps as global_apps
+
+        from ansible_base.resource_registry.signals.handlers import no_reverse_sync
+
+        # NOTE: setup_managed_role_definitions lives in the migrations module because
+        # it is also called from migration 0192. Ideally this would be extracted to a
+        # shared non-migration module, but doing so requires updating the migration
+        # import, which is a broader refactor (see also models/rbac.py imports).
+        from awx.main.migrations._dab_rbac import setup_managed_role_definitions
+
+        # During post-migrate the resource server (gateway) may not be ready
+        # (e.g. migrate_service_data still holds a 423 lock).  Disable reverse
+        # sync for this call — gateway reconciles via migrate_service_data.
+        with no_reverse_sync():
+            setup_managed_role_definitions(global_apps, None)

awx/main/db/profiled_pg/base.py

@@ -68,7 +68,7 @@ class RecordedQueryLog(object):
                     progname = match
                     break
             else:
-                progname = os.path.basename(sys.argv[0])
+                progname = 'unknown'
             filepath = os.path.join(self.dest, '{}.sqlite'.format(progname))
             version = _get_version('awx')
             log = sqlite3.connect(filepath, timeout=3)

awx/main/management/commands/cleanup_jobs.py

@@ -49,6 +49,41 @@ def dt_to_partition_name(tbl_name, dt):
     return f"{tbl_name}_{dt.strftime('%Y%m%d_%H')}"
 
 
+JHS_CHUNK_SIZE = 1000
+
+
+def _pre_delete_job_host_summaries(job_pks, logger=None):
+    """Pre-delete JobHostSummary rows and clear Host FK references in batches.
+
+    Django's cascade collector materializes all JHS IDs into a single
+    UPDATE ... IN (...) to SET_NULL on Host.last_job_host_summary.
+    With many jobs x hosts this exceeds PostgreSQL's 1GB alloc limit.
+    Doing it in chunks with raw SQL avoids that.
+    """
+    if not job_pks:
+        return
+
+    # ANY(%s) is PostgreSQL-specific; AWX only supports PostgreSQL
+    with connection.cursor() as cursor:
+        for i in range(0, len(job_pks), JHS_CHUNK_SIZE):
+            chunk = list(job_pks[i : i + JHS_CHUNK_SIZE])
+
+            cursor.execute(
+                "UPDATE main_host SET last_job_host_summary_id = NULL"
+                " WHERE last_job_host_summary_id IN"
+                " (SELECT id FROM main_jobhostsummary WHERE job_id = ANY(%s))",
+                [chunk],
+            )
+
+            cursor.execute(
+                "DELETE FROM main_jobhostsummary WHERE job_id = ANY(%s)",
+                [chunk],
+            )
+
+            if logger:
+                logger.debug("Pre-deleted JobHostSummary chunk %d-%d of %d job PKs", i, i + len(chunk), len(job_pks))
+
+
 class DeleteMeta:
     def __init__(self, logger, job_class, cutoff, dry_run):
         self.logger = logger
@@ -91,6 +126,8 @@ class DeleteMeta:
 
     def delete_jobs(self):
         if not self.dry_run:
+            if self.job_class is Job:
+                _pre_delete_job_host_summaries(self.jobs_pk_list, self.logger)
             self.job_class.objects.filter(pk__in=self.jobs_pk_list).delete()
 
     def find_partitions_to_drop(self):
@@ -265,8 +302,9 @@ class Command(BaseCommand):
         if info['min'] is not None:
             for start in range(info['min'], info['max'] + 1, self.batch_size):
                 qs_batch = qs.filter(id__gte=start, id__lte=start + self.batch_size)
-                pk_list = qs_batch.values_list('id', flat=True)
+                pk_list = list(qs_batch.values_list('id', flat=True))
 
+                _pre_delete_job_host_summaries(pk_list, self.logger)
                 _, results = qs_batch.delete()
                 deleted += results['main.Job']
                 # Avoid dropping the job event table in case we have interacted with it already

awx/main/migrations/0206_jobhostsummary_host_id_idx.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('main', '0205_add_ordering_to_instancegroup_and_workflow_nodes'),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name='jobhostsummary',
+            index=models.Index(
+                fields=['host', '-id'],
+                name='main_jobhostsumm_host_id_desc',
+            ),
+        ),
+    ]

awx/main/migrations/0207_alter_skip_tags_to_textfield.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.8 on 2026-07-20 11:07
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0206_jobhostsummary_host_id_idx'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='job',
+            name='skip_tags',
+            field=models.TextField(blank=True, default=''),
+        ),
+        migrations.AlterField(
+            model_name='jobtemplate',
+            name='skip_tags',
+            field=models.TextField(blank=True, default=''),
+        ),
+    ]

awx/main/migrations/_dab_rbac.py

@@ -148,6 +148,12 @@ def get_permissions_for_role(role_field, children_map, apps):
     if role_field.name == 'auditor_role':
         perm_list.append(Permission.objects.get(codename='view_notificationtemplate'))
 
+    # organization child admin roles need member_organization for create operations
+    if role_field.model._meta.model_name == 'organization' and role_field.name.endswith('_admin_role') and role_field.name != 'admin_role':
+        member_perm = Permission.objects.get(codename='member_organization')
+        if member_perm not in perm_list:
+            perm_list.append(member_perm)
+
     return perm_list
 
 
@@ -339,6 +345,7 @@ def setup_managed_role_definitions(apps, schema_editor):
         if 'org_children' in to_create and (cls_name not in ('organization', 'instancegroup', 'team')):
             org_child_perms = object_perms.copy()
             org_child_perms.add(Permission.objects.get(codename='view_organization'))
+            org_child_perms.add(Permission.objects.get(codename='member_organization'))
 
             managed_role_definitions.append(
                 get_or_create_managed(

awx/main/models/jobs.py

@@ -132,8 +132,7 @@ class JobOptions(BaseModel):
         blank=True,
         default=False,
     )
-    skip_tags = models.CharField(
-        max_length=1024,
+    skip_tags = models.TextField(
         blank=True,
         default='',
     )
@@ -1092,6 +1091,9 @@ class JobHostSummary(CreatedModifiedModel):
         unique_together = [('job', 'host_name')]
         verbose_name_plural = _('job host summaries')
         ordering = ('-pk',)
+        indexes = [
+            models.Index(fields=['host', '-id'], name='main_jobhostsumm_host_id_desc'),
+        ]
 
     job = models.ForeignKey(
         'Job',

awx/main/models/projects.py

@@ -450,13 +450,14 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin, CustomVirtualEn
 
     @property
     def cache_id(self):
-        """This gives the folder name where collections and roles will be saved to so it does not re-download
+        # Prefer scm_revision as the cache key if available. This guarantees that project changes are tracked correctly
+        # even over multiple nodes. The scm_revision is a hex string and thus safe to be used as a directory name.
+        if self.scm_revision:
+            return self.scm_revision
 
-        Normally we want this to track with the last update, because every update should pull new content.
-        This does not count sync jobs, but sync jobs do not update last_job or current_job anyway.
-        If cleanup_jobs deletes the last jobs, then we can fallback to using any given heuristic related
-        to the last job ran.
-        """
+        # If no scm_revision is available (e.g. non-scm projects), use these. current_job_id and last_job_id are global
+        # IDs in the database. This means that when a project sync runs on one node, all other nodes become outdated,
+        # resulting in unnecessary re-syncs.
         if self.current_job_id:
             return str(self.current_job_id)
         elif self.last_job_id:
@@ -638,7 +639,7 @@ class ProjectUpdate(UnifiedJob, ProjectOptions, JobNotificationMixin, TaskManage
 
     @property
     def cache_id(self):
-        if self.branch_override or self.job_type == 'check' or (not self.project):
+        if self.branch_override or (not self.project):
             return str(self.id)  # causes it to not use the cache, basically
         return self.project.cache_id
 

awx/main/models/rbac.py

@@ -804,7 +804,17 @@ def _sync_assignments_to_old_rbac(instance, delete=True):
 
 @receiver(post_delete, sender=RoleUserAssignment)
 @receiver(post_delete, sender=RoleTeamAssignment)
-def sync_assignments_to_old_rbac_delete(instance, **kwargs):
+def sync_assignments_to_old_rbac_delete(instance, origin=None, **kwargs):
+    # Skip cascade deletes from non-assignment origins — sync is redundant:
+    #  - Model origin with app_label != dab_rbac: a parent object (e.g.
+    #    Organization) is being deleted and old Role M2M tables cascade from
+    #    the same parent.
+    #  - QuerySet of a different model (e.g. ObjectRole): bulk RBAC cleanup
+    #    such as defer_rbac_computations flush — parent objects already gone.
+    if isinstance(origin, models.Model) and origin._meta.app_label != 'dab_rbac':
+        return
+    if isinstance(origin, models.QuerySet) and origin.model is not type(instance):
+        return
     _sync_assignments_to_old_rbac(instance, delete=True)
 
 

awx/main/models/unified_jobs.py

@@ -20,6 +20,9 @@ from dispatcherd.factories import get_control_from_settings
 # Django
 from django.conf import settings
 from django.db import models, connection, transaction
+
+# psycopg
+from psycopg import sql
 from django.db.models.constraints import UniqueConstraint
 from django.core.exceptions import NON_FIELD_ERRORS
 from django.utils.translation import gettext_lazy as _
@@ -1179,17 +1182,23 @@ class UnifiedJob(
                         raise StdoutMaxBytesExceeded(total, max_supported)
 
                 tbl = self._meta.db_table + 'event'
-                created_by_cond = ''
+                where_parts = [
+                    sql.SQL('{} = {}').format(sql.Identifier(self.event_parent_key), sql.Literal(self.id)),
+                    sql.SQL("stdout != ''"),
+                ]
                 if self.has_unpartitioned_events:
-                    tbl = f'_unpartitioned_{tbl}'
+                    tbl = '_unpartitioned_' + tbl
                 else:
-                    created_by_cond = f"job_created='{self.created.isoformat()}' AND "
+                    where_parts.insert(0, sql.SQL('job_created = {}').format(sql.Literal(self.created)))
 
-                sql = f"copy (select stdout from {tbl} where {created_by_cond}{self.event_parent_key}={self.id} and stdout != '' order by start_line) to stdout"  # nosql
+                copy_sql = sql.SQL('COPY (SELECT stdout FROM {} WHERE {} ORDER BY start_line) TO STDOUT').format(
+                    sql.Identifier(tbl),
+                    sql.SQL(' AND ').join(where_parts),
+                )
                 # psycopg3's copy writes bytes, but callers of this
                 # function assume a str-based fd will be returned; decode
                 # .write() calls on the fly to maintain this interface
-                with cursor.copy(sql) as copy:
+                with cursor.copy(copy_sql) as copy:
                     while data := copy.read():
                         fd.write(smart_str(bytes(data)))
 

awx/main/scheduler/task_manager_models.py

@@ -146,12 +146,19 @@ class TaskManagerInstances:
             self.instances_by_hostname[instance.hostname] = TaskManagerInstance(instance, **kwargs)
 
     def consume_capacity(self, task):
+        """Subtract a task's capacity from its execution and control instances.
+
+        For the control instance, jobs_running is only incremented when the controller
+        differs from the execution node to avoid double-counting on hybrid nodes.
+        """
         control_instance = self.instances_by_hostname.get(task.controller_node, '')
         execution_instance = self.instances_by_hostname.get(task.execution_node, '')
         if execution_instance and execution_instance.node_type in ('hybrid', 'execution'):
             self.instances_by_hostname[task.execution_node].consume_capacity(task.task_impact, job_impact=True)
         if control_instance and control_instance.node_type in ('hybrid', 'control'):
-            self.instances_by_hostname[task.controller_node].consume_capacity(self.control_task_impact)
+            # Track jobs_running on the controller unless it was already counted as the execution node
+            count_as_job = control_instance != execution_instance
+            self.instances_by_hostname[task.controller_node].consume_capacity(self.control_task_impact, job_impact=count_as_job)
 
     def __getitem__(self, hostname):
         return self.instances_by_hostname.get(hostname)
@@ -207,6 +214,11 @@ class TaskManagerInstanceGroups:
         return self.instance_groups[group_name].instances
 
     def fit_task_to_most_remaining_capacity_instance(self, task, instance_group_name, impact=None, capacity_type=None, add_hybrid_control_cost=False):
+        """Select the instance with the most remaining capacity after absorbing the task.
+
+        When two or more instances would have equal remaining capacity, prefer the
+        instance with fewer jobs_running to balance controller load during bursts.
+        """
         impact = impact if impact else task.task_impact
         capacity_type = capacity_type if capacity_type else task.capacity_type
         instance_most_capacity = None
@@ -220,7 +232,11 @@ class TaskManagerInstanceGroups:
             # hybrid nodes _always_ control their own tasks
             if add_hybrid_control_cost and i.node_type == 'hybrid':
                 would_be_remaining -= self.control_task_impact
-            if would_be_remaining >= 0 and (instance_most_capacity is None or would_be_remaining > most_remaining_capacity):
+            if would_be_remaining >= 0 and (
+                instance_most_capacity is None
+                or would_be_remaining > most_remaining_capacity
+                or (would_be_remaining == most_remaining_capacity and i.jobs_running < instance_most_capacity.jobs_running)
+            ):
                 instance_most_capacity = i
                 most_remaining_capacity = would_be_remaining
         return instance_most_capacity

awx/main/tasks/jobs.py

@@ -885,7 +885,9 @@ class SourceControlMixin(BaseTask):
         # Determine whether or not this project sync needs to populate the cache for Ansible content, roles and collections
         has_cache = os.path.exists(os.path.join(project.get_cache_path(), project.cache_id))
         # Galaxy requirements are not supported for manual projects
-        if project.scm_type and ((not has_cache) or branch_override):
+        # If a source update is scheduled, always include roles/collections because
+        # the new revision may have different requirements.
+        if project.scm_type and ((not has_cache) or branch_override or source_update_tag in sync_needs):
             sync_needs.extend(['install_roles', 'install_collections'])
 
         return sync_needs

awx/main/tasks/system.py

@@ -1022,6 +1022,34 @@ def update_host_smart_inventory_memberships():
         smart_inventory.update_computed_fields()
 
 
+def _batched_delete_inventory(inventory, batch_size=500):
+    """Delete inventory hosts in batches to avoid high memory usage.
+
+    With ansible facts, loading thousands of hosts at once can use a lot of memory. To avoid
+    this, we delete them in batches (of 500).
+
+    Safe to retry after a crash because inventory.pending_deletion
+    is already set and each batch is its own transaction.
+    """
+    from awx.main.models.inventory import Host
+
+    # first delete all hosts in batches
+    total_deleted = 0
+    while True:
+        pks = list(Host.objects.filter(inventory_id=inventory.id).values_list('pk', flat=True)[:batch_size])
+        if not pks:
+            break
+        with transaction.atomic():
+            deleted_count, _ = Host.objects.filter(pk__in=pks).delete()
+            total_deleted += deleted_count
+        logger.debug('Batch-deleted %d hosts from inventory %d (%d total so far)', len(pks), inventory.id, total_deleted)
+
+    # then delete the inventory itself
+    inv_id = inventory.id
+    inventory.delete()
+    logger.info('Batched deletion of inventory %d complete (%d hosts removed)', inv_id, total_deleted)
+
+
 @task(queue=get_task_queuename, timeout=3600 * 5)
 def delete_inventory(inventory_id, user_id, retries=5):
     # Delete inventory as user
@@ -1034,11 +1062,12 @@ def delete_inventory(inventory_id, user_id, retries=5):
             user = None
     with ignore_inventory_computed_fields(), ignore_inventory_group_removal(), impersonate(user):
         try:
-            Inventory.objects.get(id=inventory_id).delete()
+            inv = Inventory.objects.get(id=inventory_id)
+            _batched_delete_inventory(inv)
             emit_channel_notification('inventories-status_changed', {'group_name': 'inventories', 'inventory_id': inventory_id, 'status': 'deleted'})
             logger.debug('Deleted inventory {} as user {}.'.format(inventory_id, user_id))
         except Inventory.DoesNotExist:
-            logger.exception("Delete Inventory failed due to missing inventory: " + str(inventory_id))
+            logger.warning("Delete Inventory failed due to missing inventory: " + str(inventory_id))
             return
         except DatabaseError:
             logger.exception('Database error deleting inventory {}, but will retry.'.format(inventory_id))

awx/main/tests/functional/api/test_activity_streams.py

@@ -139,6 +139,29 @@ def test_stream_queryset_hides_shows_items(
     assert access.can_read(activity_stream_entry)
 
 
+@pytest.mark.django_db
+def test_activity_stream_pagination_uses_unfiltered_count(get, organization, project, user, settings):
+    """The pagination count should reflect total activity stream rows, not
+    the RBAC-filtered subset.  The RBAC-filtered COUNT is catastrophically
+    slow on large tables (AAP-83773); an approximate over-count from an
+    unfiltered SELECT COUNT(*) is acceptable for pagination UI."""
+    settings.ACTIVITY_STREAM_ENABLED = True
+
+    no_access_user = user('no-access-user', False)
+
+    total_entries = ActivityStream.objects.count()
+    assert total_entries > 0
+
+    url = reverse('api:activity_stream_list')
+    response = get(url, no_access_user)
+
+    assert response.status_code == 200
+    visible_results = len(response.data['results'])
+    pagination_count = response.data['count']
+    assert pagination_count == total_entries
+    assert visible_results < pagination_count
+
+
 @pytest.mark.django_db
 def test_stream_user_direct_role_updates(get, post, organization_factory):
     objects = organization_factory('test_org', superusers=['admin'], users=['test'], inventories=['inv1'])

awx/main/tests/functional/api/test_job_runtime_params.py

@@ -191,6 +191,21 @@ def test_job_accept_empty_tags(job_template_prompts, post, admin_user, mocker):
     mock_job.signal_start.assert_called_once()
 
 
+@pytest.mark.django_db
+@pytest.mark.job_runtime_vars
+def test_job_accept_long_skip_tags(job_template_prompts, post, admin_user, mocker):
+    job_template = job_template_prompts(True)
+    long_skip_tags = ','.join(f'tag{i}' for i in range(500))
+    assert len(long_skip_tags) > 1024
+
+    mock_job = mocker.MagicMock(spec=Job, id=968)
+
+    mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job)
+    mocker.patch('awx.api.serializers.JobSerializer.to_representation')
+    post(reverse('api:job_template_launch', kwargs={'pk': job_template.pk}), {'skip_tags': long_skip_tags}, admin_user, expect=201)
+    JobTemplate.create_unified_job.assert_called_once_with(skip_tags=long_skip_tags)
+
+
 @pytest.mark.django_db
 @pytest.mark.job_runtime_vars
 def test_slice_timeout_forks_need_int(job_template_prompts, post, admin_user, mocker):

awx/main/tests/functional/api/test_organization_counts.py

@@ -1,5 +1,8 @@
 import pytest
 
+from django.test.utils import CaptureQueriesContext
+from django.db import connection
+
 from ansible_base.rbac.models import RoleDefinition
 from awx.api.versioning import reverse
 
@@ -213,3 +216,24 @@ def test_JT_not_double_counted(resourced_organization, user, get):
     assert 'hosts' in counts
     counts.pop('hosts')
     assert counts == counts_dict
+
+
+@pytest.mark.django_db
+def test_org_list_user_admin_query_count(organization_resource_creator, organizations, user, get):
+    """User/admin counts use O(1) queries against roleuserassignment, not O(N) correlated subqueries."""
+    admin_user = user('admin', True)
+    extra_orgs = organizations(4)
+    member_rd = RoleDefinition.objects.get(name='Organization Member')
+    admin_rd = RoleDefinition.objects.get(name='Organization Admin')
+    for org in extra_orgs:
+        for i in range(3):
+            member_rd.give_permission(user(f'member-{org.pk}-{i}'), org)
+        admin_rd.give_permission(user(f'admin-{org.pk}'), org)
+
+    with CaptureQueriesContext(connection) as ctx:
+        response = get(reverse('api:organization_list'), admin_user)
+
+    assert response.status_code == 200
+
+    rua_queries = [q for q in ctx.captured_queries if 'dab_rbac_roleuserassignment' in q['sql']]
+    assert len(rua_queries) <= 2, f"Expected at most 2 roleuserassignment queries, got {len(rua_queries)}"

awx/main/tests/functional/api/test_unified_jobs_view.py

@@ -145,3 +145,122 @@ def test_delete_ad_hoc_command_in_active_state(ad_hoc_command_factory, delete, a
     adhoc = ad_hoc_command_factory(initial_state=status)
     url = reverse('api:ad_hoc_command_detail', kwargs={'pk': adhoc.pk})
     delete(url, None, admin, expect=403)
+
+
+@pytest.fixture
+def job_with_heavy_fields(job_factory):
+    job = job_factory()
+    job.extra_vars = '{"some_var": "some_value"}'
+    job.artifacts = {"some_artifact": "some_value"}
+    job.save()
+    return job
+
+
+def _job_result(response, job_id):
+    for row in response.data['results']:
+        if row['id'] == job_id:
+            return row
+    raise AssertionError('job {} not found in {}'.format(job_id, [r['id'] for r in response.data['results']]))
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_includes_heavy_fields_by_default(get, admin, job_with_heavy_fields):
+    response = get(reverse('api:unified_job_list') + '?id={}'.format(job_with_heavy_fields.id), admin, expect=200)
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' in row
+    assert 'extra_vars' in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_artifacts(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:unified_job_list') + '?id={}&exclude=artifacts'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' not in row
+    assert 'extra_vars' in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_extra_vars(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:unified_job_list') + '?id={}&exclude=extra_vars'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'extra_vars' not in row
+    assert 'artifacts' in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_both(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:unified_job_list') + '?id={}&exclude=artifacts,extra_vars'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' not in row
+    assert 'extra_vars' not in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_tolerates_whitespace(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:unified_job_list') + '?id={}&exclude=%20artifacts%20,%20extra_vars%20'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' not in row
+    assert 'extra_vars' not in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_ignores_unknown(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:unified_job_list') + '?id={}&exclude=does_not_exist'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' in row
+    assert 'extra_vars' in row
+
+
+@pytest.mark.django_db
+def test_unified_jobs_list_exclude_does_not_honor_always_stripped(get, admin, job_with_heavy_fields):
+    # Always-stripped fields like event_processing_finished, job_args, result_traceback
+    # must remain stripped regardless of the ?exclude= param — they cannot be re-included.
+    response = get(
+        reverse('api:unified_job_list') + '?id={}'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'event_processing_finished' not in row
+    assert 'job_args' not in row
+    assert 'result_traceback' not in row
+
+
+@pytest.mark.django_db
+def test_jobs_list_includes_heavy_fields_by_default(get, admin, job_with_heavy_fields):
+    response = get(reverse('api:job_list') + '?id={}'.format(job_with_heavy_fields.id), admin, expect=200)
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'artifacts' in row
+    assert 'extra_vars' in row
+
+
+@pytest.mark.django_db
+def test_jobs_list_exclude_extra_vars(get, admin, job_with_heavy_fields):
+    response = get(
+        reverse('api:job_list') + '?id={}&exclude=extra_vars'.format(job_with_heavy_fields.id),
+        admin,
+        expect=200,
+    )
+    row = _job_result(response, job_with_heavy_fields.id)
+    assert 'extra_vars' not in row
+    assert 'artifacts' in row

awx/main/tests/functional/conftest.py

@@ -830,14 +830,13 @@ class MockCopy:
     events = []
     index = -1
 
-    def __init__(self, sql):
+    def __init__(self):
         self.events = []
-        parts = sql.split(' ')
-        tablename = parts[parts.index('from') + 1]
         for cls in (JobEvent, AdHocCommandEvent, ProjectUpdateEvent, InventoryUpdateEvent, SystemJobEvent):
-            if cls._meta.db_table == tablename:
-                for event in cls.objects.order_by('start_line').all():
-                    self.events.append(event.stdout)
+            events = list(cls.objects.order_by('start_line').values_list('stdout', flat=True))
+            if events:
+                self.events = events
+                break
 
     def read(self):
         self.index = self.index + 1
@@ -858,9 +857,8 @@ def sqlite_copy(request, mocker):
     # copy is postgres-specific, and SQLite doesn't support it; mock its
     # behavior to test that it writes a file that contains stdout from events
 
-    def write_stdout(self, sql):
-        mock_copy = MockCopy(sql)
-        return mock_copy
+    def write_stdout(self, sql, params=None):
+        return MockCopy()
 
     mocker.patch.object(SQLiteCursorWrapper, 'copy', write_stdout, create=True)
 

awx/main/tests/functional/dab_rbac/test_access_list.py

@@ -88,7 +88,7 @@ def test_access_list_organization_access(get, admin_user, inventory):
     assert len(by_username['u2']['summary_fields']['direct_access']) == 1
     assert len(by_username['u2']['summary_fields']['indirect_access']) == 0
     access_entry = by_username['u2']['summary_fields']['direct_access'][0]
-    assert sorted(access_entry['descendant_roles']) == sorted(['inventory_admin_role', 'read_role'])
+    assert sorted(access_entry['descendant_roles']) == sorted(['inventory_admin_role', 'member_role', 'read_role'])
 
 
 @pytest.mark.django_db

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"

awx/main/tests/functional/dab_rbac/test_dab_rbac_api.py

@@ -178,10 +178,13 @@ def test_adding_actor_to_platform_roles(setup_managed_roles, role_name, actor, o
     '''
     Allow user to be added to platform-level roles
     Exceptions:
-    - Team cannot be added to Organization Member or Admin role
+    - Team cannot be added to Organization Admin role (manages teams)
     - Team cannot be added to Team Admin or Team Member role
+    Note: Team CAN be added to Organization Member role because
+    ANSIBLE_BASE_ALLOW_TEAM_ORG_MEMBER is True (required for org child
+    admin roles like Project Admin to be assignable to teams).
     '''
-    if actor == 'team':
+    if actor == 'team' and role_name != 'Organization Member':
         expect = 400
     else:
         expect = 201
@@ -195,6 +198,6 @@ def test_adding_actor_to_platform_roles(setup_managed_roles, role_name, actor, o
     r = post(url, data=data, user=admin, expect=expect)
     if expect == 400:
         if 'Organization' in role_name:
-            assert 'Assigning organization member permission to teams is not allowed' in str(r.data)
+            assert 'Assigning organization permissions that manage other teams is not allowed' in str(r.data)
         if 'Team' in role_name:
             assert 'Assigning team permissions to other teams is not allowed' in str(r.data)

awx/main/tests/functional/dab_rbac/test_managed_roles.py

@@ -39,6 +39,16 @@ def test_org_child_add_permission(setup_managed_roles):
     assert not DABPermission.objects.filter(codename='add_jobtemplate').exists()
 
 
+@pytest.mark.django_db
+def test_org_child_admin_roles_include_member_organization(setup_managed_roles):
+    """All specialized Organization *Admin roles must include member_organization
+    so that users granted these roles can perform create operations (AAP-82221)."""
+    for model_name in ('Project', 'Credential', 'Inventory', 'NotificationTemplate', 'WorkflowJobTemplate', 'ExecutionEnvironment'):
+        rd = RoleDefinition.objects.get(name=f'Organization {model_name} Admin')
+        codenames = set(rd.permissions.values_list('codename', flat=True))
+        assert 'member_organization' in codenames, f'{rd.name} is missing member_organization permission'
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize('resource_name', ['Team', 'Organization'])
 @pytest.mark.parametrize('action', ['Member', 'Admin'])

awx/main/tests/functional/dab_rbac/test_translation_layer_new_to_old.py

@@ -1,4 +1,6 @@
-from ansible_base.rbac.models import RoleDefinition, RoleUserAssignment, RoleTeamAssignment
+from unittest import mock
+
+from ansible_base.rbac.models import ObjectRole, RoleDefinition, RoleUserAssignment, RoleTeamAssignment
 from ansible_base.lib.utils.response import get_relative_url
 import pytest
 
@@ -78,3 +80,76 @@ class TestNewToOld:
         url = get_relative_url('roleteamassignment-detail', kwargs={'pk': team_assignment.id})
         delete(url, user=admin, expect=204)
         assert team.member_role not in inventory.admin_role.parents.all()
+
+    def test_flush_rbac_cleanup_skips_sync(self, inventory, bob, setup_managed_roles):
+        """Simulate what defer_rbac_computations._flush_rbac does on exit:
+        it bulk-deletes ObjectRoles for deleted objects.  Those ObjectRole
+        deletions cascade to RoleUserAssignment via the object_role FK.
+
+        Django sets origin to the *initiating* QuerySet, so the cascaded
+        assignment post_delete receives origin=<QuerySet of ObjectRole>.
+        origin.model (ObjectRole) differs from type(instance) (RoleUserAssignment),
+        identifying this as a cascade from a parent model.  The sync handler
+        must skip this — the parent objects are already gone and old Role
+        M2M entries cascade-deleted from the same parent."""
+        from django.db.models import QuerySet
+        from django.db.models.signals import post_delete
+
+        rd = RoleDefinition.objects.get(name='Inventory Admin')
+        rd.give_permission(bob, inventory)
+        assert bob in inventory.admin_role.members.all()
+
+        # Capture the origin kwarg to verify its type empirically
+        captured_origins = []
+
+        def capture_origin(sender, instance, origin=None, **kwargs):
+            if sender is RoleUserAssignment:
+                captured_origins.append(origin)
+
+        post_delete.connect(capture_origin)
+        try:
+            with mock.patch('awx.main.models.rbac._sync_assignments_to_old_rbac') as mck:
+                # This is what cleanup_deleted_team_roles does:
+                ObjectRole.objects.filter(
+                    role_definition=rd,
+                    object_id=inventory.pk,
+                ).delete()
+        finally:
+            post_delete.disconnect(capture_origin)
+
+        # Verify origin is an ObjectRole QuerySet — a different model
+        # than the deleted RoleUserAssignment instance.
+        assert len(captured_origins) == 1
+        origin = captured_origins[0]
+        assert isinstance(origin, QuerySet)
+        assert origin.model is ObjectRole
+        assert origin.model is not RoleUserAssignment
+
+        # The handler should skip sync for cross-model QuerySet origins
+        mck.assert_not_called()
+
+    def test_cascade_from_non_rbac_model_skips_sync(self, organization, inventory, bob, setup_managed_roles):
+        """When a non-RBAC parent (Organization) is deleted, cascaded assignment
+        deletions should skip the old RBAC sync entirely."""
+        rd = RoleDefinition.objects.get(name='Inventory Admin')
+        rd.give_permission(bob, inventory)
+        assert bob in inventory.admin_role.members.all()
+
+        with mock.patch('awx.main.models.rbac._sync_assignments_to_old_rbac') as mck:
+            organization.delete()
+
+        mck.assert_not_called()
+
+    def test_cascade_team_assignment_from_non_rbac_model_skips_sync(self, organization, team, inventory, setup_managed_roles):
+        """When Organization is deleted, Team cascade-deletes via real FK,
+        which cascade-deletes RoleTeamAssignment.  Django's Collector sets
+        origin to the Organization instance (a Model with app_label != 'dab_rbac'),
+        so the sync handler must skip."""
+        rd = RoleDefinition.objects.get(name='Inventory Admin')
+        rd.give_permission(team, inventory)
+        assert RoleTeamAssignment.objects.filter(team=team, role_definition=rd, object_id=inventory.pk).exists()
+
+        with mock.patch('awx.main.models.rbac._sync_assignments_to_old_rbac') as mck:
+            organization.delete()
+
+        mck.assert_not_called()

awx/main/tests/functional/dab_rbac/test_unified_job_access.py

@@ -0,0 +1,112 @@
+import pytest
+
+from django.test.utils import CaptureQueriesContext
+from django.db import connection
+
+from ansible_base.rbac.models import RoleDefinition
+
+from awx.api.versioning import reverse
+from awx.main.models import (
+    AdHocCommand,
+    InventorySource,
+    InventoryUpdate,
+    JobTemplate,
+    Organization,
+    Project,
+    UnifiedJob,
+)
+
+
+@pytest.mark.django_db
+def test_unified_job_list_uses_union(user, organization, inventory, setup_managed_roles, get):
+    """The unified job list RBAC query uses UNION instead of OR to allow per-branch query planning."""
+    org_admin = user('uj-org-admin')
+    RoleDefinition.objects.get(name='Organization Admin').give_permission(org_admin, organization)
+
+    project = Project.objects.create(name='uj-test-project', organization=organization)
+    jt = JobTemplate.objects.create(name='uj-test-jt', project=project, inventory=inventory, organization=organization)
+    jt.create_unified_job()
+
+    inv_src = InventorySource.objects.create(name='uj-test-invsrc', inventory=inventory, source='ec2')
+    InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source)
+
+    AdHocCommand.objects.create(name='uj-test-adhoc', inventory=inventory)
+
+    with CaptureQueriesContext(connection) as ctx:
+        response = get(reverse('api:unified_job_list'), org_admin)
+
+    assert response.status_code == 200
+    assert response.data['count'] >= 3
+
+    uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'UNION' in q['sql'] and 'main_unifiedjob' in q['sql']]
+    assert uj_rbac_queries, "Expected at least one query using UNION for unified job RBAC filtering"
+
+
+@pytest.mark.django_db
+def test_unified_job_list_org_auditor_sees_jobs(user, setup_managed_roles, get):
+    """Org auditors see unified jobs in their org via the audit_organization RBAC branch."""
+    org = Organization.objects.create(name='uj-audit-org')
+    auditor = user('uj-auditor')
+    RoleDefinition.objects.get(name='Organization Audit').give_permission(auditor, org)
+
+    inventory = org.inventories.create(name='uj-audit-inv')
+    project = Project.objects.create(name='uj-audit-project', organization=org)
+    jt = JobTemplate.objects.create(name='uj-audit-jt', project=project, inventory=inventory, organization=org)
+    job = jt.create_unified_job()
+
+    response = get(reverse('api:unified_job_list'), auditor)
+    assert response.status_code == 200
+    result_ids = [r['id'] for r in response.data['results']]
+    assert job.pk in result_ids
+
+
+@pytest.mark.django_db
+def test_unified_job_list_inventory_viewer_sees_inventory_updates(user, setup_managed_roles, get):
+    """Users with inventory view permission see inventory updates via the inventory RBAC branch."""
+    org = Organization.objects.create(name='uj-inv-org')
+    inventory = org.inventories.create(name='uj-inv-test')
+    inv_viewer = user('uj-inv-viewer')
+    RoleDefinition.objects.get(name='Inventory Admin').give_permission(inv_viewer, inventory)
+
+    inv_src = InventorySource.objects.create(name='uj-inv-src', inventory=inventory, source='ec2')
+    inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source)
+
+    response = get(reverse('api:unified_job_list'), inv_viewer)
+    assert response.status_code == 200
+    result_ids = [r['id'] for r in response.data['results']]
+    assert inv_update.pk in result_ids
+
+
+@pytest.mark.django_db
+def test_unified_job_list_rando_sees_nothing(rando, setup_managed_roles, get):
+    """Unprivileged user sees no unified jobs."""
+    org = Organization.objects.create(name='uj-rando-org')
+    inventory = org.inventories.create(name='uj-rando-inv')
+    project = Project.objects.create(name='uj-rando-project', organization=org)
+    jt = JobTemplate.objects.create(name='uj-rando-jt', project=project, inventory=inventory, organization=org)
+    jt.create_unified_job()
+    AdHocCommand.objects.create(name='uj-rando-adhoc', inventory=inventory)
+
+    response = get(reverse('api:unified_job_list'), rando)
+    assert response.status_code == 200
+    assert len(response.data['results']) == 0
+
+
+@pytest.mark.django_db
+def test_unified_job_list_pagination_uses_unfiltered_count(rando, setup_managed_roles, get):
+    """The pagination count should reflect total unified job rows, not
+    the RBAC-filtered subset.  The RBAC-filtered COUNT is catastrophically
+    slow on large tables with pk__in UNION subqueries."""
+    org = Organization.objects.create(name='uj-count-org')
+    inventory = org.inventories.create(name='uj-count-inv')
+    project = Project.objects.create(name='uj-count-project', organization=org)
+    jt = JobTemplate.objects.create(name='uj-count-jt', project=project, inventory=inventory, organization=org)
+    jt.create_unified_job()
+
+    total_jobs = UnifiedJob.objects.count()
+    assert total_jobs > 0
+
+    response = get(reverse('api:unified_job_list'), rando)
+    assert response.status_code == 200
+    assert len(response.data['results']) == 0
+    assert response.data['count'] == total_jobs

awx/main/tests/functional/rbac/test_rbac_execution_environment.py

@@ -33,6 +33,7 @@ def org_ee_rd():
 def test_old_ee_role_maps_to_correct_permissions(organization):
     assert set(get_role_codenames(organization.execution_environment_admin_role)) == {
         'view_organization',
+        'member_organization',
         'add_executionenvironment',
         'change_executionenvironment',
         'delete_executionenvironment',

awx/main/tests/functional/tasks/test_tasks_system.py

@@ -8,9 +8,18 @@ from unittest import mock
 
 import pytest
 
-from awx.main.tasks.system import CleanupImagesAndFiles, execution_node_health_check, inspect_established_receptor_connections, clear_setting_cache
+from awx.main.tasks.system import (
+    CleanupImagesAndFiles,
+    execution_node_health_check,
+    inspect_established_receptor_connections,
+    clear_setting_cache,
+    _batched_delete_inventory,
+)
 from awx.main.management.commands.dispatcherd import Command
-from awx.main.models import Instance, Job, ReceptorAddress, InstanceLink
+from django.db import DatabaseError
+
+from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink
+from awx.main.models.inventory import Group, Host
 
 
 @pytest.mark.django_db
@@ -105,6 +114,83 @@ def test_folder_cleanup_multiple_running_jobs(job_folder_factory, me_inst):
     assert [os.path.exists(d) for d in dirs] == [True for i in range(num_jobs)]
 
 
+@pytest.mark.django_db
+class TestBatchedDeleteInventory:
+    def _make_inventory_with_hosts(self, count):
+        from django.utils import timezone
+
+        now = timezone.now()
+        org = Organization.objects.create(name='test-org')
+        inv = Inventory.objects.create(name='test-inv', organization=org)
+        group = Group.objects.create(name='test-group', inventory=inv)
+        hosts = [Host(name=f'host-{i}', inventory=inv, created=now, modified=now) for i in range(count)]
+        Host.objects.bulk_create(hosts)
+        group.hosts.set(Host.objects.filter(inventory=inv))
+        return inv
+
+    def test_deletes_all_hosts_and_inventory(self):
+        inv = self._make_inventory_with_hosts(10)
+        inv_id = inv.id
+        _batched_delete_inventory(inv, batch_size=3)
+        assert not Host.objects.filter(inventory_id=inv_id).exists()
+        assert not Group.objects.filter(inventory_id=inv_id).exists()
+        assert not Inventory.objects.filter(id=inv_id).exists()
+
+    def test_no_hosts(self):
+        inv = self._make_inventory_with_hosts(0)
+        inv_id = inv.id
+        _batched_delete_inventory(inv)
+        assert not Inventory.objects.filter(id=inv_id).exists()
+
+    def test_exactly_one_batch(self):
+        inv = self._make_inventory_with_hosts(5)
+        inv_id = inv.id
+        _batched_delete_inventory(inv, batch_size=5)
+        assert not Host.objects.filter(inventory_id=inv_id).exists()
+        assert not Inventory.objects.filter(id=inv_id).exists()
+
+    def test_idempotent_after_partial_delete(self):
+        """Simulate a crash mid-way: delete some hosts manually, then run
+        _batched_delete_inventory — it should finish the job cleanly."""
+        inv = self._make_inventory_with_hosts(10)
+        inv_id = inv.id
+
+        # Simulate a partial deletion (as if the task crashed after 4 hosts)
+        partial_pks = list(Host.objects.filter(inventory=inv).values_list('pk', flat=True)[:4])
+        Host.objects.filter(pk__in=partial_pks).delete()
+        assert Host.objects.filter(inventory_id=inv_id).count() == 6
+
+        # Re-running should delete the remaining hosts and the inventory
+        inv.refresh_from_db()
+        _batched_delete_inventory(inv, batch_size=3)
+        assert not Host.objects.filter(inventory_id=inv_id).exists()
+        assert not Inventory.objects.filter(id=inv_id).exists()
+
+    def test_delete_inventory_retries_on_database_error(self):
+        """DatabaseError during deletion triggers a retry."""
+        from awx.main.tasks.system import delete_inventory
+
+        inv = self._make_inventory_with_hosts(3)
+        inv_id = inv.id
+
+        call_count = {'n': 0}
+        original = _batched_delete_inventory.__wrapped__ if hasattr(_batched_delete_inventory, '__wrapped__') else _batched_delete_inventory
+
+        def flaky_delete(inventory, batch_size=500):
+            call_count['n'] += 1
+            if call_count['n'] == 1:
+                raise DatabaseError('connection reset')
+            return original(inventory, batch_size=batch_size)
+
+        with mock.patch('awx.main.tasks.system._batched_delete_inventory', side_effect=flaky_delete):
+            with mock.patch('awx.main.tasks.system.emit_channel_notification'):
+                with mock.patch('awx.main.tasks.system.time.sleep'):
+                    delete_inventory(inv_id, None, retries=2)
+
+        assert call_count['n'] == 2
+        assert not Inventory.objects.filter(id=inv_id).exists()
+
+
 @pytest.mark.django_db
 def test_clear_setting_cache_log_level_branch(settings):
     settings.LOG_AGGREGATOR_LEVEL = 'DEBUG'

awx/main/tests/unit/api/serializers/test_unified_serializers.py

@@ -39,7 +39,7 @@ def test_unified_job_detail_exclusive_fields():
     For each type, assert that the only fields allowed to be exclusive to
     detail view are the allowed types
     """
-    allowed_detail_fields = frozenset(('result_traceback', 'job_args', 'job_cwd', 'job_env', 'event_processing_finished', 'artifacts'))
+    allowed_detail_fields = frozenset(('result_traceback', 'job_args', 'job_cwd', 'job_env', 'event_processing_finished'))
     for cls in UnifiedJob.__subclasses__():
         list_serializer = getattr(serializers, '{}ListSerializer'.format(cls.__name__))
         detail_serializer = getattr(serializers, '{}Serializer'.format(cls.__name__))

awx/main/tests/unit/api/test_pagination.py

@@ -0,0 +1,123 @@
+from unittest.mock import patch, MagicMock
+
+from awx.api.pagination import ActivityStreamPaginator, ActivityStreamPagination, UnifiedJobPaginator, UnifiedJobPagination, DisabledPaginator
+
+
+class TestActivityStreamPaginator:
+    def test_count_uses_unfiltered_table_count(self):
+        with patch('awx.api.pagination.ActivityStream') as mock_as:
+            mock_as.objects.count.return_value = 713000
+            paginator = ActivityStreamPaginator(object_list=[], per_page=25)
+            assert paginator.count == 713000
+            mock_as.objects.count.assert_called_once()
+
+    def test_count_is_cached(self):
+        with patch('awx.api.pagination.ActivityStream') as mock_as:
+            mock_as.objects.count.return_value = 500
+            paginator = ActivityStreamPaginator(object_list=[], per_page=25)
+            _ = paginator.count
+            _ = paginator.count
+            mock_as.objects.count.assert_called_once()
+
+
+class TestActivityStreamPagination:
+    def test_default_paginator_class(self):
+        pagination = ActivityStreamPagination()
+        assert pagination.django_paginator_class is ActivityStreamPaginator
+
+    def test_normal_request_preserves_activity_stream_paginator(self):
+        pagination = ActivityStreamPagination()
+        request = MagicMock()
+        request.query_params = {}
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', return_value=[]):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert pagination.count_disabled is False
+        assert pagination.django_paginator_class is ActivityStreamPaginator
+
+    def test_count_disabled_restores_activity_stream_paginator(self):
+        pagination = ActivityStreamPagination()
+        request = MagicMock()
+        request.query_params = {'count_disabled': 'true'}
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', return_value=[]):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert pagination.count_disabled is True
+        assert pagination.django_paginator_class is ActivityStreamPaginator
+
+    def test_count_disabled_temporarily_uses_disabled_paginator(self):
+        pagination = ActivityStreamPagination()
+        request = MagicMock()
+        request.query_params = {'count_disabled': 'true'}
+        captured_class = {}
+
+        def capture_paginator_class(self_inner, queryset, request, **kwargs):
+            captured_class['during'] = pagination.django_paginator_class
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', capture_paginator_class):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert captured_class['during'] is DisabledPaginator
+        assert pagination.django_paginator_class is ActivityStreamPaginator
+
+
+class TestUnifiedJobPaginator:
+    def test_count_uses_unfiltered_table_count(self):
+        with patch('awx.api.pagination.UnifiedJob') as mock_uj:
+            mock_uj.objects.count.return_value = 42000
+            paginator = UnifiedJobPaginator(object_list=[], per_page=25)
+            assert paginator.count == 42000
+            mock_uj.objects.count.assert_called_once()
+
+    def test_count_is_cached(self):
+        with patch('awx.api.pagination.UnifiedJob') as mock_uj:
+            mock_uj.objects.count.return_value = 500
+            paginator = UnifiedJobPaginator(object_list=[], per_page=25)
+            _ = paginator.count
+            _ = paginator.count
+            mock_uj.objects.count.assert_called_once()
+
+
+class TestUnifiedJobPagination:
+    def test_default_paginator_class(self):
+        pagination = UnifiedJobPagination()
+        assert pagination.django_paginator_class is UnifiedJobPaginator
+
+    def test_normal_request_preserves_unified_job_paginator(self):
+        pagination = UnifiedJobPagination()
+        request = MagicMock()
+        request.query_params = {}
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', return_value=[]):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert pagination.count_disabled is False
+        assert pagination.django_paginator_class is UnifiedJobPaginator
+
+    def test_count_disabled_restores_unified_job_paginator(self):
+        pagination = UnifiedJobPagination()
+        request = MagicMock()
+        request.query_params = {'count_disabled': 'true'}
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', return_value=[]):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert pagination.count_disabled is True
+        assert pagination.django_paginator_class is UnifiedJobPaginator
+
+    def test_count_disabled_temporarily_uses_disabled_paginator(self):
+        pagination = UnifiedJobPagination()
+        request = MagicMock()
+        request.query_params = {'count_disabled': 'true'}
+        captured_class = {}
+
+        def capture_paginator_class(self_inner, queryset, request, **kwargs):
+            captured_class['during'] = pagination.django_paginator_class
+
+        with patch('rest_framework.pagination.PageNumberPagination.paginate_queryset', capture_paginator_class):
+            pagination.paginate_queryset(MagicMock(), request)
+
+        assert captured_class['during'] is DisabledPaginator
+        assert pagination.django_paginator_class is UnifiedJobPaginator

awx/main/tests/unit/commands/test_cleanup_jobs.py

@@ -0,0 +1,137 @@
+from unittest import mock
+
+from awx.main.management.commands.cleanup_jobs import _pre_delete_job_host_summaries, JHS_CHUNK_SIZE
+
+
+class TestPreDeleteJobHostSummaries:
+    def test_empty_list_is_noop(self):
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            _pre_delete_job_host_summaries([])
+            mock_conn.cursor.assert_not_called()
+
+    def test_single_chunk(self):
+        job_pks = [1, 2, 3]
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            mock_cursor = mock.MagicMock()
+            mock_conn.cursor.return_value.__enter__ = mock.Mock(return_value=mock_cursor)
+            mock_conn.cursor.return_value.__exit__ = mock.Mock(return_value=False)
+
+            _pre_delete_job_host_summaries(job_pks)
+
+            assert mock_cursor.execute.call_count == 2
+            update_call = mock_cursor.execute.call_args_list[0]
+            assert 'UPDATE main_host SET last_job_host_summary_id = NULL' in update_call[0][0]
+            assert 'ANY(%s)' in update_call[0][0]
+            assert update_call[0][1] == [[1, 2, 3]]
+
+            delete_call = mock_cursor.execute.call_args_list[1]
+            assert 'DELETE FROM main_jobhostsummary' in delete_call[0][0]
+            assert 'ANY(%s)' in delete_call[0][0]
+            assert delete_call[0][1] == [[1, 2, 3]]
+
+    def test_multiple_chunks(self):
+        job_pks = list(range(1, JHS_CHUNK_SIZE + 500))
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            mock_cursor = mock.MagicMock()
+            mock_conn.cursor.return_value.__enter__ = mock.Mock(return_value=mock_cursor)
+            mock_conn.cursor.return_value.__exit__ = mock.Mock(return_value=False)
+
+            _pre_delete_job_host_summaries(job_pks)
+
+            # 2 chunks x 2 SQL statements each = 4 execute calls
+            assert mock_cursor.execute.call_count == 4
+
+            # First chunk should have JHS_CHUNK_SIZE items
+            first_update = mock_cursor.execute.call_args_list[0]
+            assert len(first_update[0][1][0]) == JHS_CHUNK_SIZE
+
+            # Second chunk should have the remainder
+            second_update = mock_cursor.execute.call_args_list[2]
+            assert len(second_update[0][1][0]) == 499
+
+    def test_sql_is_fully_static(self):
+        """SQL strings contain no interpolated values — only ANY(%s) placeholders."""
+        job_pks = [100, 200]
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            mock_cursor = mock.MagicMock()
+            mock_conn.cursor.return_value.__enter__ = mock.Mock(return_value=mock_cursor)
+            mock_conn.cursor.return_value.__exit__ = mock.Mock(return_value=False)
+
+            _pre_delete_job_host_summaries(job_pks)
+
+            for call in mock_cursor.execute.call_args_list:
+                sql = call[0][0]
+                assert 'ANY(%s)' in sql
+                assert '100' not in sql
+                assert '200' not in sql
+
+    def test_logger_called_per_chunk(self):
+        job_pks = [1, 2, 3]
+        logger = mock.MagicMock()
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            mock_cursor = mock.MagicMock()
+            mock_conn.cursor.return_value.__enter__ = mock.Mock(return_value=mock_cursor)
+            mock_conn.cursor.return_value.__exit__ = mock.Mock(return_value=False)
+
+            _pre_delete_job_host_summaries(job_pks, logger=logger)
+
+            logger.debug.assert_called_once()
+
+    def test_update_runs_before_delete(self):
+        """Host FK must be NULLed before JHS rows are deleted."""
+        job_pks = [1]
+        with mock.patch('awx.main.management.commands.cleanup_jobs.connection') as mock_conn:
+            mock_cursor = mock.MagicMock()
+            mock_conn.cursor.return_value.__enter__ = mock.Mock(return_value=mock_cursor)
+            mock_conn.cursor.return_value.__exit__ = mock.Mock(return_value=False)
+
+            _pre_delete_job_host_summaries(job_pks)
+
+            first_sql = mock_cursor.execute.call_args_list[0][0][0]
+            second_sql = mock_cursor.execute.call_args_list[1][0][0]
+            assert 'UPDATE' in first_sql
+            assert 'DELETE' in second_sql
+
+
+class TestDeleteMetaPreDelete:
+    """Verify DeleteMeta.delete_jobs() calls _pre_delete_job_host_summaries correctly."""
+
+    @mock.patch('awx.main.management.commands.cleanup_jobs._pre_delete_job_host_summaries')
+    def test_called_for_job_class(self, mock_pre_delete):
+        from awx.main.management.commands.cleanup_jobs import DeleteMeta
+        from awx.main.models import Job
+
+        dm = DeleteMeta(logger=mock.MagicMock(), job_class=Job, cutoff=mock.MagicMock(), dry_run=False)
+        dm.jobs_pk_list = [10, 20, 30]
+
+        with mock.patch.object(Job.objects, 'filter') as mock_filter:
+            mock_filter.return_value.delete.return_value = (3, {})
+            dm.delete_jobs()
+
+        mock_pre_delete.assert_called_once_with([10, 20, 30], dm.logger)
+
+    @mock.patch('awx.main.management.commands.cleanup_jobs._pre_delete_job_host_summaries')
+    def test_skipped_for_non_job_class(self, mock_pre_delete):
+        from awx.main.management.commands.cleanup_jobs import DeleteMeta
+        from awx.main.models import ProjectUpdate
+
+        dm = DeleteMeta(logger=mock.MagicMock(), job_class=ProjectUpdate, cutoff=mock.MagicMock(), dry_run=False)
+        dm.jobs_pk_list = [10, 20]
+
+        with mock.patch.object(ProjectUpdate.objects, 'filter') as mock_filter:
+            mock_filter.return_value.delete.return_value = (2, {})
+            dm.delete_jobs()
+
+        mock_pre_delete.assert_not_called()
+
+    @mock.patch('awx.main.management.commands.cleanup_jobs._pre_delete_job_host_summaries')
+    def test_skipped_for_dry_run(self, mock_pre_delete):
+        from awx.main.management.commands.cleanup_jobs import DeleteMeta
+        from awx.main.models import Job
+
+        dm = DeleteMeta(logger=mock.MagicMock(), job_class=Job, cutoff=mock.MagicMock(), dry_run=True)
+        dm.jobs_pk_list = [10, 20]
+
+        dm.delete_jobs()
+
+        mock_pre_delete.assert_not_called()

awx/main/tests/unit/test_capacity.py

@@ -177,10 +177,21 @@ class TestSelectBestInstanceForTask(object):
         'task,instances,instance_fit_index,reason',
         [
             (Job(task_impact=100), Is([100]), 0, "Only one, pick it"),
-            (Job(task_impact=100), Is([100, 100]), 0, "Two equally good fits, pick the first"),
+            (Job(task_impact=100), Is([100, 100]), 0, "Two equally good fits and equal jobs_running, pick the first"),
             (Job(task_impact=100), Is([50, 100]), 1, "First instance not as good as second instance"),
             (Job(task_impact=100), Is([50, 0, 20, 100, 100, 100, 30, 20]), 3, "Pick Instance [3] as it is the first that the task fits in."),
             (Job(task_impact=100), Is([50, 0, 20, 99, 11, 1, 5, 99]), None, "The task don't a fit, you must a quit!"),
+            # Tie-breaking: equal would_be_remaining but different jobs_running
+            # capacity=243 with 1 running (impact 43) → remaining 200 → would_be 100
+            # capacity=200 with 0 running → remaining 200 → would_be 100
+            (Job(task_impact=100), Is([(1, 243), (0, 200)]), 1, "Equal would_be_remaining, pick instance with fewer jobs running"),
+            # Three-way tie: capacities chosen so that remaining after consumption yields equal would_be
+            # (2 running, cap 286): remaining=286-86=200, would_be=100
+            # (0 running, cap 200): remaining=200, would_be=100
+            # (1 running, cap 243): remaining=243-43=200, would_be=100
+            (Job(task_impact=100), Is([(2, 286), (0, 200), (1, 243)]), 1, "Three-way tie, pick instance with fewest jobs running"),
+            (Job(task_impact=100), Is([(0, 200), (0, 300)]), 1, "Different capacity, pick higher capacity regardless of jobs"),
+            (Job(task_impact=100), Is([(5, 600), (0, 200)]), 0, "Higher remaining capacity wins even with more jobs running"),
         ],
     )
     def test_fit_task_to_most_remaining_capacity_instance(self, task, instances, instance_fit_index, reason):
@@ -198,6 +209,63 @@ class TestSelectBestInstanceForTask(object):
         else:
             assert instance_picked.hostname == instances[instance_fit_index].hostname, reason
 
+    def test_controller_node_tie_break_with_container_group_jobs(self):
+        """Verify that controller nodes managing container-group jobs track jobs_running
+        and tie-break correctly. Container-group jobs have controller_node set but no
+        execution_node, simulating the real burst workload scenario."""
+        ig = InstanceGroup(id=10, name='controlplane')
+        ctrl_a = Instance(hostname='ctrl-a', capacity=200, node_type='control')
+        ctrl_b = Instance(hostname='ctrl-b', capacity=200, node_type='control')
+        ctrl_c = Instance(hostname='ctrl-c', capacity=200, node_type='control')
+        ig.instances.add(ctrl_a, ctrl_b, ctrl_c)
+
+        # Simulate container-group jobs: ctrl-a has 3, ctrl-b has 1, ctrl-c has 0
+        tasks = [
+            Job(controller_node='ctrl-a', execution_node='', instance_group=ig),
+            Job(controller_node='ctrl-a', execution_node='', instance_group=ig),
+            Job(controller_node='ctrl-a', execution_node='', instance_group=ig),
+            Job(controller_node='ctrl-b', execution_node='', instance_group=ig),
+        ]
+        tm_models = TaskManagerModels.init_with_consumed_capacity(tasks=tasks, instances=[ctrl_a, ctrl_b, ctrl_c], instance_groups=[ig])
+
+        # ctrl-a: consumed=3, jobs_running=3, remaining=197
+        # ctrl-b: consumed=1, jobs_running=1, remaining=199
+        # ctrl-c: consumed=0, jobs_running=0, remaining=200
+        # New task with impact=1 (control): ctrl-c has most remaining (200) → wins by capacity
+        task = Job(task_impact=1, capacity_type='control')
+        picked = tm_models.instance_groups.fit_task_to_most_remaining_capacity_instance(task, 'controlplane')
+        assert picked.hostname == 'ctrl-c', "Should pick the node with most remaining capacity"
+
+    def test_controller_node_tie_break_equal_capacity(self):
+        """When control nodes have exactly equal remaining capacity, prefer the one
+        with fewer jobs_running. This is the core burst-distribution scenario."""
+        ig = InstanceGroup(id=10, name='controlplane')
+        ctrl_a = Instance(hostname='ctrl-a', capacity=200, node_type='control')
+        ctrl_b = Instance(hostname='ctrl-b', capacity=200, node_type='control')
+        ig.instances.add(ctrl_a, ctrl_b)
+
+        # Both nodes have same consumed capacity (1 each) but ctrl-a has 1 job, ctrl-b has 1 job
+        # After this, both have remaining=199, jobs_running=1 → tie → first wins
+        tasks = [
+            Job(controller_node='ctrl-a', execution_node='', instance_group=ig),
+            Job(controller_node='ctrl-b', execution_node='', instance_group=ig),
+        ]
+        tm_models = TaskManagerModels.init_with_consumed_capacity(tasks=tasks, instances=[ctrl_a, ctrl_b], instance_groups=[ig])
+        task = Job(task_impact=1, capacity_type='control')
+        picked = tm_models.instance_groups.fit_task_to_most_remaining_capacity_instance(task, 'controlplane')
+        # Both tied on capacity (199) and jobs_running (1) → first in iteration wins
+        assert picked.hostname == 'ctrl-a', "Equal capacity and jobs: first in iteration wins"
+
+        # Now give ctrl-a one more job, making it 2 vs 1
+        tasks.append(Job(controller_node='ctrl-a', execution_node='', instance_group=ig))
+        tm_models = TaskManagerModels.init_with_consumed_capacity(tasks=tasks, instances=[ctrl_a, ctrl_b], instance_groups=[ig])
+        # ctrl-a: consumed=2, jobs_running=2, remaining=198
+        # ctrl-b: consumed=1, jobs_running=1, remaining=199
+        # ctrl-b wins by capacity (199 > 198)
+        task = Job(task_impact=1, capacity_type='control')
+        picked = tm_models.instance_groups.fit_task_to_most_remaining_capacity_instance(task, 'controlplane')
+        assert picked.hostname == 'ctrl-b', "Node with fewer jobs has more remaining capacity, wins"
+
     @pytest.mark.parametrize(
         'instances,instance_fit_index,reason',
         [

awx/main/tests/unit/test_db.py

@@ -1,7 +1,6 @@
 import collections
 import os
 import sqlite3
-import sys
 import unittest
 
 import pytest
@@ -125,7 +124,7 @@ def test_sql_above_threshold(tmpdir):
         args, kw = _call
         assert args == ('EXPLAIN VERBOSE {}'.format(QUERY['sql']),)
 
-    path = os.path.join(tmpdir, '{}.sqlite'.format(os.path.basename(sys.argv[0])))
+    path = os.path.join(tmpdir, 'unknown.sqlite')
     assert os.path.exists(path)
 
     # verify the results

awx/main/tests/unit/test_statement_timeout.py

@@ -0,0 +1,79 @@
+import types
+from unittest import mock
+
+from awx.settings.functions import merge_statement_timeout
+
+PG_ENGINE = "django.db.backends.postgresql"
+SQLITE_ENGINE = "django.db.backends.sqlite3"
+
+
+def _make_settings(engine=PG_ENGINE, timeout=None, existing_options=""):
+    """Build a dict that quacks like DYNACONF.get() for merge_statement_timeout."""
+    data = {"DATABASES__default__ENGINE": engine}
+    if timeout is not None:
+        data["DATABASE_STATEMENT_TIMEOUT"] = timeout
+    if existing_options:
+        data["DATABASES__default__OPTIONS__options"] = existing_options
+    return data
+
+
+def _fake_uwsgi(harakiri):
+    mod = types.ModuleType('uwsgi')
+    mod.opt = {b'harakiri': str(harakiri).encode()}
+    return mod
+
+
+class TestMergeStatementTimeout:
+    def test_derives_from_uwsgi_harakiri(self):
+        settings = _make_settings()
+        with mock.patch.dict('sys.modules', {'uwsgi': _fake_uwsgi(115)}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=110000"}
+
+    def test_returns_empty_without_uwsgi_or_setting(self):
+        settings = _make_settings()
+        with mock.patch.dict('sys.modules', {'uwsgi': None}):
+            result = merge_statement_timeout(settings)
+        assert result == {}
+
+    def test_falls_back_to_setting(self):
+        settings = _make_settings(timeout=60000)
+        with mock.patch.dict('sys.modules', {'uwsgi': None}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=60000"}
+
+    def test_uwsgi_takes_precedence_over_setting(self):
+        settings = _make_settings(timeout=60000)
+        with mock.patch.dict('sys.modules', {'uwsgi': _fake_uwsgi(115)}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=110000"}
+
+    def test_harakiri_zero_falls_back_to_setting(self):
+        settings = _make_settings(timeout=90000)
+        with mock.patch.dict('sys.modules', {'uwsgi': _fake_uwsgi(0)}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=90000"}
+
+    def test_harakiri_very_low_clamps_to_one_second(self):
+        settings = _make_settings()
+        with mock.patch.dict('sys.modules', {'uwsgi': _fake_uwsgi(1)}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=1000"}
+
+    def test_harakiri_midrange_uses_proportional_margin(self):
+        settings = _make_settings()
+        with mock.patch.dict('sys.modules', {'uwsgi': _fake_uwsgi(30)}):
+            # margin = min(5, max(1, int(30*0.1))) = 3 → timeout = 27s
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c statement_timeout=27000"}
+
+    def test_skips_sqlite(self):
+        settings = _make_settings(engine=SQLITE_ENGINE, timeout=60000)
+        result = merge_statement_timeout(settings)
+        assert result == {}
+
+    def test_appends_to_existing_options(self):
+        settings = _make_settings(timeout=60000, existing_options="-c lock_timeout=5000")
+        with mock.patch.dict('sys.modules', {'uwsgi': None}):
+            result = merge_statement_timeout(settings)
+        assert result == {"DATABASES__default__OPTIONS__options": "-c lock_timeout=5000 -c statement_timeout=60000"}

awx/main/tests/unit/utils/test_validate_rh.py

@@ -1,3 +1,6 @@
+import configparser
+
+import pytest
 from unittest.mock import patch
 from awx.main.utils.licensing import Licenser
 
@@ -6,21 +9,25 @@ def test_validate_rh_basic_auth_rhsm():
     """
     Assert get_rhsm_subs is called when
     - basic_auth=True
+    - REDHAT_CANDLEPIN_HOST is not set
     - host is subscription.rhsm.redhat.com
     """
     licenser = Licenser()
 
-    with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com') as mock_get_host, patch.object(
-        licenser, 'get_rhsm_subs', return_value=[]
-    ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs') as mock_get_satellite, patch.object(
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(
+        licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'
+    ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs', return_value=[]) as mock_get_rhsm, patch.object(
+        licenser, 'get_satellite_subs'
+    ) as mock_get_satellite, patch.object(
         licenser, 'get_crc_subs'
     ) as mock_get_crc, patch.object(
         licenser, 'generate_license_options_from_entitlements'
     ) as mock_generate:
 
+        mock_settings.REDHAT_CANDLEPIN_HOST = None
+
         licenser.validate_rh('testuser', 'testpass', basic_auth=True)
 
-        # Assert the correct methods were called
         mock_get_host.assert_called_once()
         mock_get_rhsm.assert_called_once_with('https://subscription.rhsm.redhat.com', 'testuser', 'testpass')
         mock_get_satellite.assert_not_called()
@@ -32,21 +39,25 @@ def test_validate_rh_basic_auth_satellite():
     """
     Assert get_satellite_subs is called when
     - basic_auth=True
-    - custom satellite host
+    - REDHAT_CANDLEPIN_HOST is not set
+    - rhsm.conf points to a non-RHSM host
     """
     licenser = Licenser()
 
-    with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://satellite.example.com') as mock_get_host, patch.object(
-        licenser, 'get_rhsm_subs'
-    ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object(
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(
+        licenser, 'get_host_from_rhsm_config', return_value='https://satellite.example.com'
+    ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs') as mock_get_rhsm, patch.object(
+        licenser, 'get_satellite_subs', return_value=[]
+    ) as mock_get_satellite, patch.object(
         licenser, 'get_crc_subs'
     ) as mock_get_crc, patch.object(
         licenser, 'generate_license_options_from_entitlements'
     ) as mock_generate:
 
+        mock_settings.REDHAT_CANDLEPIN_HOST = None
+
         licenser.validate_rh('testuser', 'testpass', basic_auth=True)
 
-        # Assert the correct methods were called
         mock_get_host.assert_called_once()
         mock_get_rhsm.assert_not_called()
         mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass')
@@ -73,7 +84,6 @@ def test_validate_rh_service_account_crc():
 
         licenser.validate_rh('client_id', 'client_secret', basic_auth=False)
 
-        # Assert the correct methods were called
         mock_get_host.assert_not_called()
         mock_get_rhsm.assert_not_called()
         mock_get_satellite.assert_not_called()
@@ -81,74 +91,127 @@ def test_validate_rh_service_account_crc():
         mock_generate.assert_called_once_with([], is_candlepin=False)
 
 
+def test_validate_rh_candlepin_host_prioritized_over_rhsm_config():
+    """Test REDHAT_CANDLEPIN_HOST takes priority over rhsm.conf
+    - basic_auth=True
+    - REDHAT_CANDLEPIN_HOST is set
+    - rhsm.conf should NOT be consulted
+    """
+    licenser = Licenser()
+
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config') as mock_get_host, patch.object(
+        licenser, 'get_rhsm_subs'
+    ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object(
+        licenser, 'get_crc_subs'
+    ) as mock_get_crc, patch.object(
+        licenser, 'generate_license_options_from_entitlements'
+    ) as mock_generate:
+
+        mock_settings.REDHAT_CANDLEPIN_HOST = 'https://satellite.example.com'
+        licenser.validate_rh('testuser', 'testpass', basic_auth=True)
+
+        mock_get_host.assert_not_called()
+        mock_get_rhsm.assert_not_called()
+        mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass')
+        mock_get_crc.assert_not_called()
+        mock_generate.assert_called_once_with([], is_candlepin=True)
+
+
+def test_validate_rh_prepends_scheme_when_missing():
+    """REDHAT_CANDLEPIN_HOST without a scheme gets https:// prepended"""
+    licenser = Licenser()
+
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config'), patch.object(
+        licenser, 'get_rhsm_subs'
+    ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object(
+        licenser, 'get_crc_subs'
+    ), patch.object(
+        licenser, 'generate_license_options_from_entitlements'
+    ):
+
+        mock_settings.REDHAT_CANDLEPIN_HOST = 'satellite.example.com'
+        licenser.validate_rh('testuser', 'testpass', basic_auth=True)
+
+        mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass')
+        mock_get_rhsm.assert_not_called()
+
+
 def test_validate_rh_missing_user_raises_error():
     """Test validate_rh raises ValueError when user is missing"""
     licenser = Licenser()
 
-    with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'):
-        try:
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(
+        licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'
+    ):
+        mock_settings.REDHAT_CANDLEPIN_HOST = None
+        with pytest.raises(ValueError, match='subscriptions_client_id or subscriptions_username is required'):
             licenser.validate_rh(None, 'testpass', basic_auth=True)
-            assert False, "Expected ValueError to be raised"
-        except ValueError as e:
-            assert 'subscriptions_client_id or subscriptions_username is required' in str(e)
 
 
 def test_validate_rh_missing_password_raises_error():
     """Test validate_rh raises ValueError when password is missing"""
     licenser = Licenser()
 
-    with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'):
-        try:
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(
+        licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'
+    ):
+        mock_settings.REDHAT_CANDLEPIN_HOST = None
+        with pytest.raises(ValueError, match='subscriptions_client_secret or subscriptions_password is required'):
             licenser.validate_rh('testuser', None, basic_auth=True)
-            assert False, "Expected ValueError to be raised"
-        except ValueError as e:
-            assert 'subscriptions_client_secret or subscriptions_password is required' in str(e)
 
 
-def test_validate_rh_no_host_fallback_to_candlepin():
-    """Test validate_rh falls back to REDHAT_CANDLEPIN_HOST when no host from config
+@pytest.mark.parametrize(
+    'host_input, rhsm_port, expected_in_url, not_expected_in_url',
+    [
+        ('https://satellite.example.com:8443', '443', ':8443', ':8443:443'),
+        ('https://satellite.example.com', '8443', ':8443', None),
+        ('https://satellite.example.com/', '8443', ':8443', '/:'),
+    ],
+    ids=['skip-port-when-present', 'append-port-when-missing', 'strip-trailing-slash'],
+)
+def test_get_satellite_subs_port_handling(host_input, rhsm_port, expected_in_url, not_expected_in_url):
+    licenser = Licenser()
+    licenser.config = configparser.ConfigParser()
+    licenser.config.read_string(f"[server]\nhostname=satellite.example.com\nport={rhsm_port}\n[rhsm]\nrepo_ca_cert=/etc/rhsm/ca/redhat-uep.pem\n")
+
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch('awx.main.utils.licensing.requests') as mock_requests:
+        mock_settings.REDHAT_CANDLEPIN_VERIFY = None
+        mock_orgs = mock_requests.get.return_value
+        mock_orgs.json.return_value = {'results': []}
+
+        licenser.get_satellite_subs(host_input, 'user', 'pw')
+
+        called_url = mock_requests.get.call_args[0][0]
+        assert expected_in_url in called_url
+        if not_expected_in_url:
+            assert not_expected_in_url not in called_url
+
+
+def test_get_satellite_subs_uses_candlepin_verify_setting():
+    """REDHAT_CANDLEPIN_VERIFY should take priority over rhsm.conf ca_cert"""
+    licenser = Licenser()
+    licenser.config = configparser.ConfigParser()
+    licenser.config.read_string("[server]\nhostname=satellite.example.com\n[rhsm]\nrepo_ca_cert=/etc/rhsm/ca/redhat-uep.pem\n")
+
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch('awx.main.utils.licensing.requests') as mock_requests:
+        mock_settings.REDHAT_CANDLEPIN_VERIFY = False
+        mock_orgs = mock_requests.get.return_value
+        mock_orgs.json.return_value = {'results': []}
+
+        licenser.get_satellite_subs('https://satellite.example.com', 'user', 'pw')
+
+        assert mock_requests.get.call_args[1]['verify'] is False
+
+
+def test_validate_rh_no_host_raises_error():
+    """Test validate_rh raises ValueError when no host is available
     - basic_auth=True
-    - no host from config
-    - REDHAT_CANDLEPIN_HOST is set
+    - REDHAT_CANDLEPIN_HOST is not set
+    - rhsm.conf returns None
     """
     licenser = Licenser()
 
-    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(
-        licenser, 'get_host_from_rhsm_config', return_value=None
-    ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs', return_value=[]) as mock_get_rhsm, patch.object(
-        licenser, 'get_satellite_subs', return_value=[]
-    ) as mock_get_satellite, patch.object(
-        licenser, 'get_crc_subs'
-    ) as mock_get_crc, patch.object(
-        licenser, 'generate_license_options_from_entitlements'
-    ) as mock_generate:
-
-        mock_settings.REDHAT_CANDLEPIN_HOST = 'https://candlepin.example.com'
-        licenser.validate_rh('testuser', 'testpass', basic_auth=True)
-
-        # Assert the correct methods were called
-        mock_get_host.assert_called_once()
-        mock_get_rhsm.assert_not_called()
-        mock_get_satellite.assert_called_once_with('https://candlepin.example.com', 'testuser', 'testpass')
-        mock_get_crc.assert_not_called()
-        mock_generate.assert_called_once_with([], is_candlepin=True)
-
-
-def test_validate_rh_empty_credentials_basic_auth():
-    """Test validate_rh with empty string credentials raises ValueError"""
-    licenser = Licenser()
-
-    with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'):
-        # Test empty user
-        try:
-            licenser.validate_rh(None, 'testpass', basic_auth=True)
-            assert False, "Expected ValueError to be raised"
-        except ValueError as e:
-            assert 'subscriptions_client_id or subscriptions_username is required' in str(e)
-
-        # Test empty password
-        try:
-            licenser.validate_rh('testuser', None, basic_auth=True)
-            assert False, "Expected ValueError to be raised"
-        except ValueError as e:
-            assert 'subscriptions_client_secret or subscriptions_password is required' in str(e)
+    with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config', return_value=None):
+        mock_settings.REDHAT_CANDLEPIN_HOST = None
+        with pytest.raises(ValueError, match='Could not get host url for subscriptions'):
+            licenser.validate_rh('testuser', 'testpass', basic_auth=True)

awx/main/utils/licensing.py

@@ -19,6 +19,7 @@ import json
 import logging
 import re
 import requests
+from urllib.parse import urlparse
 import time
 import zipfile
 
@@ -228,18 +229,16 @@ class Licenser(object):
         return host
 
     def validate_rh(self, user, pw, basic_auth):
-        # if basic auth is True, host is read from rhsm.conf (subscription.rhsm.redhat.com)
-        # if basic auth is False, host is settings.SUBSCRIPTIONS_RHSM_URL (console.redhat.com)
-        # if rhsm.conf is not found, host is settings.REDHAT_CANDLEPIN_HOST (satellite server)
         if basic_auth:
-            host = self.get_host_from_rhsm_config()
-            if not host:
-                host = getattr(settings, 'REDHAT_CANDLEPIN_HOST', None)
+            if not (host := getattr(settings, 'REDHAT_CANDLEPIN_HOST', None)):
+                host = self.get_host_from_rhsm_config()
         else:
             host = settings.SUBSCRIPTIONS_RHSM_URL
 
         if not host:
             raise ValueError('Could not get host url for subscriptions')
+        if not host.startswith(('https://', 'http://')):
+            host = 'https://' + host
 
         if not user:
             raise ValueError('subscriptions_client_id or subscriptions_username is required')
@@ -308,13 +307,20 @@ class Licenser(object):
 
     def get_satellite_subs(self, host, user, pw):
         port = None
+        if (verify := getattr(settings, 'REDHAT_CANDLEPIN_VERIFY', None)) is None:
+            try:
+                verify = str(self.config.get("rhsm", "repo_ca_cert"))
+            except Exception as e:
+                logger.exception(f'Unable to read rhsm config to get ca_cert location. {e}')
+                verify = True
         try:
-            verify = str(self.config.get("rhsm", "repo_ca_cert"))
             port = str(self.config.get("server", "port"))
-        except Exception as e:
-            logger.exception('Unable to read rhsm config to get ca_cert location. {}'.format(str(e)))
-            verify = True
-        if port:
+        except Exception:
+            port = None
+        host = host.rstrip('/')
+        # Append port from rhsm.conf only if the host URL doesn't already include one
+        # (REDHAT_CANDLEPIN_HOST may already contain a port)
+        if port and not urlparse(host).port:
             host = ':'.join([host, port])
         json = []
         try:

awx/settings/__init__.py

@@ -12,6 +12,7 @@ from ansible_base.lib.dynamic_config import (
 from .functions import (
     assert_production_settings,
     merge_application_name,
+    merge_statement_timeout,
     add_backwards_compatibility,
     load_extra_development_files,
 )
@@ -78,6 +79,11 @@ DYNACONF.update(
     loader_identifier="awx.settings:merge_application_name",
     merge=True,
 )
+DYNACONF.update(
+    merge_statement_timeout(DYNACONF),
+    loader_identifier="awx.settings:merge_statement_timeout",
+    merge=True,
+)
 
 # Update django.conf.settings with DYNACONF values
 export(__name__, DYNACONF)

awx/settings/defaults.py

@@ -43,6 +43,11 @@ LISTENER_DATABASES = {
     }
 }
 
+# Optional manual override for statement_timeout (ms) on web worker DB
+# connections.  When running under uwsgi, the timeout is auto-derived from
+# the harakiri value.  Set this for non-uwsgi deployments or to override.
+DATABASE_STATEMENT_TIMEOUT = None
+
 # Whether or not the deployment is a K8S-based deployment
 # In K8S-based deployments, instances have zero capacity - all playbook
 # automation is intended to flow through defined Container Groups that
@@ -1103,6 +1108,7 @@ ANSIBLE_BASE_CACHE_PARENT_PERMISSIONS = True
 
 # Currently features are enabled to keep compatibility with old system, except custom roles
 ANSIBLE_BASE_ALLOW_TEAM_ORG_ADMIN = False
+ANSIBLE_BASE_ALLOW_TEAM_ORG_MEMBER = True
 # ANSIBLE_BASE_ALLOW_CUSTOM_ROLES = True
 ANSIBLE_BASE_ALLOW_TEAM_PARENTS = False
 ANSIBLE_BASE_ALLOW_CUSTOM_TEAM_ROLES = False

awx/settings/functions.py

@@ -12,6 +12,41 @@ def merge_application_name(settings):
     return data
 
 
+def merge_statement_timeout(settings):
+    """Return a dynaconf merge dict to set statement_timeout for web worker DB connections.
+
+    Under uwsgi, derives timeout from harakiri with a safety margin so
+    PostgreSQL cancels the query before uwsgi kills the worker.  The margin
+    is 10% of harakiri, clamped to [1s, 5s].  Falls back to the
+    DATABASE_STATEMENT_TIMEOUT setting for non-uwsgi deployments.
+    """
+    if "sqlite3" in settings.get("DATABASES__default__ENGINE", ""):
+        return {}
+
+    timeout_ms = None
+    try:
+        import uwsgi
+
+        harakiri = int(uwsgi.opt.get(b'harakiri', 0))
+        if harakiri > 0:
+            margin = min(5, max(1, int(harakiri * 0.1)))
+            timeout_ms = max(1000, (harakiri - margin) * 1000)
+    except (ImportError, ValueError):
+        pass
+
+    if timeout_ms is None:
+        timeout_ms = settings.get("DATABASE_STATEMENT_TIMEOUT")
+
+    if timeout_ms is None:
+        return {}
+
+    existing = settings.get("DATABASES__default__OPTIONS__options", "")
+    new_opt = f"-c statement_timeout={timeout_ms}"
+    value = f"{existing} {new_opt}".strip() if existing else new_opt
+
+    return {"DATABASES__default__OPTIONS__options": value}
+
+
 def add_backwards_compatibility():
     """Add backwards compatibility for AWX_MODE.
 

awx_collection/plugins/module_utils/controller_api.py

@@ -438,7 +438,7 @@ class ControllerAPIModule(ControllerModule):
             raise RuntimeError('Expected list from API at {0}, got: {1}'.format(endpoint, response))
         next_page = response['json']['next']
 
-        if response['json']['count'] > 10000:
+        if response['json'].get('count', 0) > 10000:
             self.fail_json(msg='The number of items being queried for is higher than 10,000.')
 
         while next_page is not None:
@@ -493,8 +493,11 @@ class ControllerAPIModule(ControllerModule):
                     fail_msg += ', detail: {0}'.format(response['json']['detail'])
                 self.fail_json(msg=fail_msg)
 
-            if 'count' not in response['json'] or 'results' not in response['json']:
-                self.fail_json(msg="The endpoint did not provide count and results")
+            if 'results' not in response['json']:
+                self.fail_json(msg="The endpoint did not provide a results list")
+
+            if 'count' not in response['json']:
+                response['json']['count'] = len(response['json']['results'])
 
         if response['json']['count'] == 0:
             if allow_none:

awx_collection/plugins/modules/credential_input_source.py

@@ -93,7 +93,15 @@ def main():
     metadata = module.params.get('metadata')
     state = module.params.get('state')
 
-    target_credential_id = module.resolve_name_to_id('credentials', target_credential)
+    # The target credential lookup should not fail if the target credential is absent and the
+    # state on the credential input source is also absent. If the credential input source has a
+    # state of present, then this should fail as the target credential cannot be nonexistent.
+    target_credential_lookup = module.get_one('credentials', name_or_id=target_credential, allow_none=(state == 'absent'))
+
+    if target_credential_lookup is None:
+        module.exit_json(**{'changed': False})
+    else:
+        target_credential_id = target_credential_lookup['id']
 
     # Attempt to look up the object based on the target credential and input field
     lookup_data = {

awx_collection/plugins/modules/subscriptions.py

@@ -62,13 +62,18 @@ subscriptions:
 EXAMPLES = '''
 - name: Get subscriptions
   subscriptions:
-    client_id: "c6bd7594-d776-46e5-8156-6d17af147479"
-    client_secret: "MO9QUvoOZ5fc5JQKXoTch1AsTLI7nFsZ"
+    client_id: "00000000-0000-0000-0000-000000000000"
+    client_secret: "your-client-secret-here"
+
+- name: Get subscriptions with username and password
+  subscriptions:
+    username: "my_username"
+    password: "my_password"
 
 - name: Get subscriptions with a filter
   subscriptions:
-    client_id: "c6bd7594-d776-46e5-8156-6d17af147479"
-    client_secret: "MO9QUvoOZ5fc5JQKXoTch1AsTLI7nFsZ"
+    client_id: "00000000-0000-0000-0000-000000000000"
+    client_secret: "your-client-secret-here"
     filters:
       product_name: "Red Hat Ansible Automation Platform"
       support_level: "Self-Support"

awx_collection/test/awx/test_credential_input_source.py

@@ -358,3 +358,95 @@ def test_centrify_vault_credential_source(run_module, admin_user, organization,
     assert cis.target_credential.name == tgt_cred.name
     assert cis.input_field_name == 'password'
     assert result['id'] == cis.pk
+
+
+@pytest.mark.django_db
+def test_credential_input_source_delete(run_module, admin_user, organization, source_cred_aim, silence_deprecation):
+    ct = CredentialType.defaults['ssh']()
+    ct.save()
+    tgt_cred = Credential.objects.create(name='Test Machine Credential', organization=organization, credential_type=ct, inputs={'username': 'nick'})
+
+    result = run_module(
+        'credential_input_source',
+        dict(
+            source_credential=source_cred_aim.name,
+            target_credential=tgt_cred.name,
+            input_field_name='password',
+            metadata={"object_query": "Safe=SUPERSAFE;Object=MyAccount"},
+            state='present',
+        ),
+        admin_user,
+    )
+
+    assert not result.get('failed', False), result.get('msg', result)
+    assert result.get('changed'), result
+    assert CredentialInputSource.objects.count() == 1
+
+    delete_result = run_module(
+        'credential_input_source',
+        dict(
+            target_credential=tgt_cred.name,
+            input_field_name='password',
+            state='absent',
+        ),
+        admin_user,
+    )
+
+    assert not delete_result.get('failed', False), delete_result.get('msg', delete_result)
+    assert delete_result.get('changed'), delete_result
+    assert CredentialInputSource.objects.count() == 0
+
+
+@pytest.mark.django_db
+def test_credential_input_source_delete_nonexistent(run_module, admin_user, organization, source_cred_aim, silence_deprecation):
+    ct = CredentialType.defaults['ssh']()
+    ct.save()
+    tgt_cred = Credential.objects.create(name='Test Machine Credential', organization=organization, credential_type=ct, inputs={'username': 'bob'})
+
+    result = run_module(
+        'credential_input_source',
+        dict(
+            target_credential=tgt_cred.name,
+            input_field_name='password',
+            state='absent',
+        ),
+        admin_user,
+    )
+
+    assert not result.get('failed', False), result.get('msg', result)
+    assert not result.get('changed'), result
+    assert CredentialInputSource.objects.count() == 0
+
+
+@pytest.mark.django_db
+def test_credential_input_source_delete_missing_target_credential(run_module, admin_user, organization, silence_deprecation):
+    result = run_module(
+        'credential_input_source',
+        dict(
+            target_credential='nonexistent-credential',
+            input_field_name='password',
+            state='absent',
+        ),
+        admin_user,
+    )
+
+    assert not result.get('failed', False), result.get('msg', result)
+    assert not result.get('changed'), result
+    assert CredentialInputSource.objects.count() == 0
+
+
+@pytest.mark.django_db
+def test_credential_input_source_create_missing_target_credential(run_module, admin_user, organization, source_cred_aim, silence_deprecation):
+    result = run_module(
+        'credential_input_source',
+        dict(
+            source_credential=source_cred_aim.name,
+            target_credential='nonexistent-credential',
+            input_field_name='password',
+            state='present',
+        ),
+        admin_user,
+    )
+
+    assert result.get('failed', True), "Should fail when target credential doesn't exist with state: present"
+    assert CredentialInputSource.objects.count() == 0

awxkit/conftest.py

@@ -0,0 +1,16 @@
+# This conftest registers pytest-django CLI options and INI keys as harmless
+# no-ops so that the awxkit tox environment (which does not install Django or
+# pytest-django) does not crash when the root pytest.ini config leaks through.
+#
+# Root pytest.ini contains settings like --reuse-db, --nomigrations, and
+# DJANGO_SETTINGS_MODULE that are only meaningful for Django tests.  Placing
+# this conftest at the awxkit/ level ensures it is loaded before argument
+# parsing regardless of which config file pytest ultimately selects.
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("awxkit-compat", "awxkit tox compatibility shims")
+    group.addoption("--reuse-db", action="store_true", default=False, help="(no-op in awxkit) pytest-django compat")
+    group.addoption("--nomigrations", action="store_true", default=False, help="(no-op in awxkit) pytest-django compat")
+    group.addoption("--create-db", action="store_true", default=False, help="(no-op in awxkit) pytest-django compat")
+    parser.addini("DJANGO_SETTINGS_MODULE", help="(no-op in awxkit) pytest-django compat", default="")

awxkit/pytest.ini

@@ -0,0 +1,9 @@
+[pytest]
+addopts = -v --tb=native
+testpaths = test
+python_files = test_*.py
+
+filterwarnings =
+  error
+
+junit_family=xunit2

awxkit/tox.ini

@@ -19,7 +19,7 @@ deps =
     pytest-mock
 
 commands =
-    coverage run --parallel --source awxkit -m pytest --doctest-glob='*.md' --junit-xml=report.xml {posargs}
+    coverage run --parallel --source awxkit -m pytest -c pytest.ini test --doctest-glob='*.md' --junit-xml=report.xml {posargs}
     coverage combine
     coverage xml
 
@@ -44,6 +44,8 @@ max-line-length = 120
 
 [pytest]
 addopts = -v --tb=native
+testpaths = test
+python_files = test_*.py
 
 filterwarnings =
   error

docs/release_process.md

@@ -62,6 +62,12 @@ If the latest release of `AWX` is 19.5.0:
 
 With very few exceptions the new `AWX Operator` release will always be a Y-stream release.
 
+## Cutting a Stable Branch
+
+When cutting a new stable branch for a release, there are additional dependency verification and branch protection steps that must be completed. See the full checklist on Confluence:
+
+[Cutting a Stable Branch](https://redhat.atlassian.net/wiki/spaces/AAP/pages/417433803/Cutting+a+Stable+Branch)
+
 ## Stage the release
 
 To stage the release, maintainers of this repository can run the [Stage Release](https://github.com/ansible/awx/actions/workflows/stage.yml) workflow.
@@ -228,4 +234,4 @@ Here are the steps needed to revert an AWX and an AWX-Operator release. Dependin
 
 7. Navigate to the [PyPi](https://pypi.org/project/awxkit/#history) and delete the bad AWX tag and release that got published.
 
-8. [Restart the Release Process](#releasing-awx-and-awx-operator)
+8. [Restart the Release Process](#releasing-awx-and-awx-operator)

licenses/backports-zstd.txt

@@ -0,0 +1,277 @@
+A. HISTORY OF THE SOFTWARE
+==========================
+
+Python was created in the early 1990s by Guido van Rossum at Stichting
+Mathematisch Centrum (CWI, see https://www.cwi.nl) in the Netherlands
+as a successor of a language called ABC.  Guido remains Python's
+principal author, although it includes many contributions from others.
+
+In 1995, Guido continued his work on Python at the Corporation for
+National Research Initiatives (CNRI, see https://www.cnri.reston.va.us)
+in Reston, Virginia where he released several versions of the
+software.
+
+In May 2000, Guido and the Python core development team moved to
+BeOpen.com to form the BeOpen PythonLabs team.  In October of the same
+year, the PythonLabs team moved to Digital Creations, which became
+Zope Corporation.  In 2001, the Python Software Foundation (PSF, see
+https://www.python.org/psf/) was formed, a non-profit organization
+created specifically to own Python-related Intellectual Property.
+Zope Corporation was a sponsoring member of the PSF.
+
+All Python releases are Open Source (see https://opensource.org for
+the Open Source Definition).  Historically, most, but not all, Python
+releases have also been GPL-compatible; the table below summarizes
+the various releases.
+
+    Release         Derived     Year        Owner       GPL-
+                    from                                compatible? (1)
+
+    0.9.0 thru 1.2              1991-1995   CWI         yes
+    1.3 thru 1.5.2  1.2         1995-1999   CNRI        yes
+    1.6             1.5.2       2000        CNRI        no
+    2.0             1.6         2000        BeOpen.com  no
+    1.6.1           1.6         2001        CNRI        yes (2)
+    2.1             2.0+1.6.1   2001        PSF         no
+    2.0.1           2.0+1.6.1   2001        PSF         yes
+    2.1.1           2.1+2.0.1   2001        PSF         yes
+    2.1.2           2.1.1       2002        PSF         yes
+    2.1.3           2.1.2       2002        PSF         yes
+    2.2 and above   2.1.1       2001-now    PSF         yes
+
+Footnotes:
+
+(1) GPL-compatible doesn't mean that we're distributing Python under
+    the GPL.  All Python licenses, unlike the GPL, let you distribute
+    a modified version without making your changes open source.  The
+    GPL-compatible licenses make it possible to combine Python with
+    other software that is released under the GPL; the others don't.
+
+(2) According to Richard Stallman, 1.6.1 is not GPL-compatible,
+    because its license has a choice of law clause.  According to
+    CNRI, however, Stallman's lawyer has told CNRI's lawyer that 1.6.1
+    is "not incompatible" with the GPL.
+
+Thanks to the many outside volunteers who have worked under Guido's
+direction to make these releases possible.
+
+
+B. TERMS AND CONDITIONS FOR ACCESSING OR OTHERWISE USING PYTHON
+===============================================================
+
+Python software and documentation are licensed under the
+Python Software Foundation License Version 2.
+
+Starting with Python 3.8.6, examples, recipes, and other code in
+the documentation are dual licensed under the PSF License Version 2
+and the Zero-Clause BSD license.
+
+Some software incorporated into Python is under different licenses.
+The licenses are listed with code falling under that license.
+
+
+PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2
+--------------------------------------------
+
+1. This LICENSE AGREEMENT is between the Python Software Foundation
+("PSF"), and the Individual or Organization ("Licensee") accessing and
+otherwise using this software ("Python") in source or binary form and
+its associated documentation.
+
+2. Subject to the terms and conditions of this License Agreement, PSF hereby
+grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce,
+analyze, test, perform and/or display publicly, prepare derivative works,
+distribute, and otherwise use Python alone or in any derivative version,
+provided, however, that PSF's License Agreement and PSF's notice of copyright,
+i.e., "Copyright (c) 2001 Python Software Foundation; All Rights Reserved"
+are retained in Python alone or in any derivative version prepared by Licensee.
+
+3. In the event Licensee prepares a derivative work that is based on
+or incorporates Python or any part thereof, and wants to make
+the derivative work available to others as provided herein, then
+Licensee hereby agrees to include in any such work a brief summary of
+the changes made to Python.
+
+4. PSF is making Python available to Licensee on an "AS IS"
+basis.  PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR
+IMPLIED.  BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND
+DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS
+FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT
+INFRINGE ANY THIRD PARTY RIGHTS.
+
+5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON
+FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS
+A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON,
+OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF.
+
+6. This License Agreement will automatically terminate upon a material
+breach of its terms and conditions.
+
+7. Nothing in this License Agreement shall be deemed to create any
+relationship of agency, partnership, or joint venture between PSF and
+Licensee.  This License Agreement does not grant permission to use PSF
+trademarks or trade name in a trademark sense to endorse or promote
+products or services of Licensee, or any third party.
+
+8. By copying, installing or otherwise using Python, Licensee
+agrees to be bound by the terms and conditions of this License
+Agreement.
+
+
+BEOPEN.COM LICENSE AGREEMENT FOR PYTHON 2.0
+-------------------------------------------
+
+BEOPEN PYTHON OPEN SOURCE LICENSE AGREEMENT VERSION 1
+
+1. This LICENSE AGREEMENT is between BeOpen.com ("BeOpen"), having an
+office at 160 Saratoga Avenue, Santa Clara, CA 95051, and the
+Individual or Organization ("Licensee") accessing and otherwise using
+this software in source or binary form and its associated
+documentation ("the Software").
+
+2. Subject to the terms and conditions of this BeOpen Python License
+Agreement, BeOpen hereby grants Licensee a non-exclusive,
+royalty-free, world-wide license to reproduce, analyze, test, perform
+and/or display publicly, prepare derivative works, distribute, and
+otherwise use the Software alone or in any derivative version,
+provided, however, that the BeOpen Python License is retained in the
+Software, alone or in any derivative version prepared by Licensee.
+
+3. BeOpen is making the Software available to Licensee on an "AS IS"
+basis.  BEOPEN MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR
+IMPLIED.  BY WAY OF EXAMPLE, BUT NOT LIMITATION, BEOPEN MAKES NO AND
+DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS
+FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF THE SOFTWARE WILL NOT
+INFRINGE ANY THIRD PARTY RIGHTS.
+
+4. BEOPEN SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF THE
+SOFTWARE FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS
+AS A RESULT OF USING, MODIFYING OR DISTRIBUTING THE SOFTWARE, OR ANY
+DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF.
+
+5. This License Agreement will automatically terminate upon a material
+breach of its terms and conditions.
+
+6. This License Agreement shall be governed by and interpreted in all
+respects by the law of the State of California, excluding conflict of
+law provisions.  Nothing in this License Agreement shall be deemed to
+create any relationship of agency, partnership, or joint venture
+between BeOpen and Licensee.  This License Agreement does not grant
+permission to use BeOpen trademarks or trade names in a trademark
+sense to endorse or promote products or services of Licensee, or any
+third party.  As an exception, the "BeOpen Python" logos available at
+http://www.pythonlabs.com/logos.html may be used according to the
+permissions granted on that web page.
+
+7. By copying, installing or otherwise using the software, Licensee
+agrees to be bound by the terms and conditions of this License
+Agreement.
+
+
+CNRI LICENSE AGREEMENT FOR PYTHON 1.6.1
+---------------------------------------
+
+1. This LICENSE AGREEMENT is between the Corporation for National
+Research Initiatives, having an office at 1895 Preston White Drive,
+Reston, VA 20191 ("CNRI"), and the Individual or Organization
+("Licensee") accessing and otherwise using Python 1.6.1 software in
+source or binary form and its associated documentation.
+
+2. Subject to the terms and conditions of this License Agreement, CNRI
+hereby grants Licensee a nonexclusive, royalty-free, world-wide
+license to reproduce, analyze, test, perform and/or display publicly,
+prepare derivative works, distribute, and otherwise use Python 1.6.1
+alone or in any derivative version, provided, however, that CNRI's
+License Agreement and CNRI's notice of copyright, i.e., "Copyright (c)
+1995-2001 Corporation for National Research Initiatives; All Rights
+Reserved" are retained in Python 1.6.1 alone or in any derivative
+version prepared by Licensee.  Alternately, in lieu of CNRI's License
+Agreement, Licensee may substitute the following text (omitting the
+quotes): "Python 1.6.1 is made available subject to the terms and
+conditions in CNRI's License Agreement.  This Agreement together with
+Python 1.6.1 may be located on the internet using the following
+unique, persistent identifier (known as a handle): 1895.22/1013.  This
+Agreement may also be obtained from a proxy server on the internet
+using the following URL: http://hdl.handle.net/1895.22/1013".
+
+3. In the event Licensee prepares a derivative work that is based on
+or incorporates Python 1.6.1 or any part thereof, and wants to make
+the derivative work available to others as provided herein, then
+Licensee hereby agrees to include in any such work a brief summary of
+the changes made to Python 1.6.1.
+
+4. CNRI is making Python 1.6.1 available to Licensee on an "AS IS"
+basis.  CNRI MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR
+IMPLIED.  BY WAY OF EXAMPLE, BUT NOT LIMITATION, CNRI MAKES NO AND
+DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS
+FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON 1.6.1 WILL NOT
+INFRINGE ANY THIRD PARTY RIGHTS.
+
+5. CNRI SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON
+1.6.1 FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS
+A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON 1.6.1,
+OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF.
+
+6. This License Agreement will automatically terminate upon a material
+breach of its terms and conditions.
+
+7. This License Agreement shall be governed by the federal
+intellectual property law of the United States, including without
+limitation the federal copyright law, and, to the extent such
+U.S. federal law does not apply, by the law of the Commonwealth of
+Virginia, excluding Virginia's conflict of law provisions.
+Notwithstanding the foregoing, with regard to derivative works based
+on Python 1.6.1 that incorporate non-separable material that was
+previously distributed under the GNU General Public License (GPL), the
+law of the Commonwealth of Virginia shall govern this License
+Agreement only as to issues arising under or with respect to
+Paragraphs 4, 5, and 7 of this License Agreement.  Nothing in this
+License Agreement shall be deemed to create any relationship of
+agency, partnership, or joint venture between CNRI and Licensee.  This
+License Agreement does not grant permission to use CNRI trademarks or
+trade name in a trademark sense to endorse or promote products or
+services of Licensee, or any third party.
+
+8. By clicking on the "ACCEPT" button where indicated, or by copying,
+installing or otherwise using Python 1.6.1, Licensee agrees to be
+bound by the terms and conditions of this License Agreement.
+
+        ACCEPT
+
+
+CWI LICENSE AGREEMENT FOR PYTHON 0.9.0 THROUGH 1.2
+--------------------------------------------------
+
+Copyright (c) 1991 - 1995, Stichting Mathematisch Centrum Amsterdam,
+The Netherlands.  All rights reserved.
+
+Permission to use, copy, modify, and distribute this software and its
+documentation for any purpose and without fee is hereby granted,
+provided that the above copyright notice appear in all copies and that
+both that copyright notice and this permission notice appear in
+supporting documentation, and that the name of Stichting Mathematisch
+Centrum or CWI not be used in advertising or publicity pertaining to
+distribution of the software without specific, written prior
+permission.
+
+STICHTING MATHEMATISCH CENTRUM DISCLAIMS ALL WARRANTIES WITH REGARD TO
+THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
+FITNESS, IN NO EVENT SHALL STICHTING MATHEMATISCH CENTRUM BE LIABLE
+FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
+OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+ZERO-CLAUSE BSD LICENSE FOR CODE IN THE PYTHON DOCUMENTATION
+----------------------------------------------------------------------
+
+Permission to use, copy, modify, and/or distribute this software for any
+purpose with or without fee is hereby granted.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+PERFORMANCE OF THIS SOFTWARE.

licenses/zstandard.txt

@@ -1,27 +0,0 @@
-Copyright (c) 2016, Gregory Szorc
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without modification,
-are permitted provided that the following conditions are met:
-
-1. Redistributions of source code must retain the above copyright notice, this
-list of conditions and the following disclaimer.
-
-2. Redistributions in binary form must reproduce the above copyright notice,
-this list of conditions and the following disclaimer in the documentation
-and/or other materials provided with the distribution.
-
-3. Neither the name of the copyright holder nor the names of its contributors
-may be used to endorse or promote products derived from this software without
-specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
-ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
-ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
-(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
-LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
-ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

requirements/requirements.in

@@ -1,4 +1,4 @@
-aiohttp>=3.12.14 # CVE-2024-30251
+aiohttp>=3.13.5 # CVE-2024-30251, kubernetes>=36.0.2 requires >=3.13.5
 ansi2html  # Used to format the stdout from jobs into html for display
 jq  # used for indirect host counting feature
 asn1
@@ -36,7 +36,7 @@ maturin  # pydantic-core build dep
 msgpack
 msrestazure
 OPA-python-client==2.0.2 # upgrading requires urllib3 2.5.0+ which is blocked by other deps
-kubernetes>=36.0.0  # fixes NO_PROXY silently being reset to None
+kubernetes>=36.0.2  # fixes NO_PROXY silently being reset to None
 openshift
 opentelemetry-api~=1.37     # new y streams can be drastically different, in a good way
 opentelemetry-sdk~=1.37

requirements/requirements.txt

@@ -6,7 +6,7 @@ aiofiles==24.1.0
     # via opa-python-client
 aiohappyeyeballs==2.6.1
     # via aiohttp
-aiohttp[speedups]==3.13.0
+aiohttp[speedups]==3.14.1
     # via
     #   -r /awx_devel/requirements/requirements.in
     #   aiohttp-retry
@@ -67,6 +67,8 @@ azure-keyvault-keys==4.11.0
     # via azure-keyvault
 azure-keyvault-secrets==4.10.0
     # via azure-keyvault
+backports-zstd==1.6.0
+    # via aiohttp
 boto3==1.40.46
     # via -r /awx_devel/requirements/requirements.in
 botocore==1.40.46
@@ -74,7 +76,7 @@ botocore==1.40.46
     #   -r /awx_devel/requirements/requirements.in
     #   boto3
     #   s3transfer
-brotli==1.1.0
+brotli==1.2.0
     # via aiohttp
 cachetools==6.2.0
     # via -r /awx_devel/requirements/requirements.in
@@ -252,7 +254,7 @@ jsonschema==4.25.1
     #   drf-spectacular
 jsonschema-specifications==2025.9.1
     # via jsonschema
-kubernetes==36.0.0
+kubernetes==36.0.2
     # via
     #   -r /awx_devel/requirements/requirements.in
     #   openshift
@@ -500,6 +502,7 @@ txaio==25.9.2
     # via autobahn
 typing-extensions==4.15.0
     # via
+    #   aiohttp
     #   aiosignal
     #   azure-core
     #   azure-identity
@@ -544,8 +547,6 @@ zipp==3.23.0
     # via importlib-metadata
 zope-interface==8.0.1
     # via twisted
-zstandard==0.25.0
-    # via aiohttp
 
 # The following packages are considered to be unsafe in a requirements file:
 pip==25.3

sonar-project.properties

@@ -113,10 +113,13 @@ sonar.cpd.exclusions=\
 # =============================================================================
 
 # Ignore specific rules for certain file patterns
-sonar.issue.ignore.multicriteria=e1
+sonar.issue.ignore.multicriteria=e1,e2
 # Ignore "should be a variable" in migrations
 sonar.issue.ignore.multicriteria.e1.ruleKey=python:S1192
 sonar.issue.ignore.multicriteria.e1.resourceKey=**/migrations/**/*
+# Ignore "use literal instead of constructor" in collection modules — dict() is idiomatic Ansible
+sonar.issue.ignore.multicriteria.e2.ruleKey=python:S7498
+sonar.issue.ignore.multicriteria.e2.resourceKey=awx_collection/plugins/modules/**
 
 # =============================================================================
 # GITHUB INTEGRATION

tox.ini

@@ -17,7 +17,7 @@ commands =
 
 [flake8]
 select = F401,F402,F821,F823,F841,F811,E265,E266,F541,W605,E722,F822,F523,W291,F405
-exclude = env,awx_collection_build
+exclude = env,awx_collection_build,.tox
 
 [testenv:pip-compile-docs]
 description = Compile docs build lockfiles