From 32564e94a4e9595ab05afe4ebd5be7a8e774f212 Mon Sep 17 00:00:00 2001 From: John Westcott IV Date: Mon, 22 Sep 2025 11:31:14 -0400 Subject: [PATCH 1/7] Decoupling apps from ansible_base.rbac --- ansible_base/authentication/utils/claims.py | 21 ++++++++++++++----- .../routers/association_resource_router.py | 11 +++++----- 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/ansible_base/authentication/utils/claims.py b/ansible_base/authentication/utils/claims.py index 8f2e21413..1690b582b 100644 --- a/ansible_base/authentication/utils/claims.py +++ b/ansible_base/authentication/utils/claims.py @@ -21,8 +21,6 @@ from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel from ansible_base.lib.utils.auth import get_organization_model, get_team_model from ansible_base.lib.utils.string import is_empty -from ansible_base.rbac.models import DABContentType -from ansible_base.rbac.remote import get_local_resource_prefix from .trigger_definition import TRIGGER_DEFINITION @@ -32,6 +30,9 @@ User = get_user_model() +is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS + + class TriggerResult(Enum): ALLOW = auto() DENY = auto() @@ -722,7 +723,7 @@ def reconcile_user_claims(cls, user: AbstractUser, authenticator_user: Authentic claims = getattr(user, 'claims', authenticator_user.claims) - if 'ansible_base.rbac' in settings.INSTALLED_APPS: + if is_rbac_installed: cls(claims, user, authenticator_user).manage_permissions() else: logger.info(_("Skipping user claims with RBAC roles, because RBAC app is not installed")) @@ -876,7 +877,11 @@ class RoleUserAssignmentsCache: def __init__(self): self.cache = {} # NOTE(cutwater): We may probably execute this query once and cache the query results. - self.content_types = {content_type.model: content_type for content_type in DABContentType.objects.get_for_models(Organization, Team).values()} + self.content_types = {} + if is_rbac_installed: + from ansible_base.rbac.models import DABContentType + + self.content_types = {content_type.model: content_type for content_type in DABContentType.objects.get_for_models(Organization, Team).values()} self.role_definitions = {} def items(self): @@ -956,6 +961,12 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None: - All cached assignments are marked with STATUS_EXISTING status - Role definitions are also cached separately in self.role_definitions """ + local_resource_prefixes = ["shared"] + if is_rbac_installed: + from ansible_base.rbac.remote import get_local_resource_prefix + + local_resource_prefixes.append(get_local_resource_prefix()) + for role_assignment in role_assignments: # Cache role definition if (role_definition := self._rd_by_id(role_assignment)) is None: @@ -965,7 +976,7 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None: # Skip role assignments that should not be cached if not ( role_assignment.content_type is None # Global/system roles (e.g., System Auditor) - or role_assignment.content_type.service in [get_local_resource_prefix(), "shared"] + or role_assignment.content_type.service in local_resource_prefixes ): # Local object roles continue diff --git a/ansible_base/lib/routers/association_resource_router.py b/ansible_base/lib/routers/association_resource_router.py index 19855be16..af4d7c5d5 100644 --- a/ansible_base/lib/routers/association_resource_router.py +++ b/ansible_base/lib/routers/association_resource_router.py @@ -17,8 +17,6 @@ from rest_framework.response import Response from rest_framework.viewsets import ViewSetMixin -from ansible_base.rbac.permission_registry import permission_registry - logger = logging.getLogger('ansible_base.lib.routers.association_resource_router') @@ -119,10 +117,13 @@ def check_parent_object_permissions(self, request, parent_obj: Model) -> None: will not check "change" permissions to the parent object on POST this method checks parent change permission, view permission should be handled by filter_queryset """ - if (request.method not in SAFE_METHODS) and 'ansible_base.rbac' in settings.INSTALLED_APPS and permission_registry.is_registered(parent_obj): - from ansible_base.rbac.policies import check_content_obj_permission + if (request.method not in SAFE_METHODS) and 'ansible_base.rbac' in settings.INSTALLED_APPS: + from ansible_base.rbac.permission_registry import permission_registry + + if permission_registry.is_registered(parent_obj): + from ansible_base.rbac.policies import check_content_obj_permission - check_content_obj_permission(request.user, parent_obj) + check_content_obj_permission(request.user, parent_obj) def get_parent_object(self) -> Model: """Modeled mostly after DRF get_object, but for the parent model From 436ccce0ff93c45c4023605bf546df77bfe008c4 Mon Sep 17 00:00:00 2001 From: John Westcott IV Date: Mon, 22 Sep 2025 12:27:37 -0400 Subject: [PATCH 2/7] Make resource_registry work without rbac --- ansible_base/resource_registry/registry.py | 19 +- ansible_base/resource_registry/rest_client.py | 15 ++ .../resource_registry/shared_types.py | 28 ++- ansible_base/resource_registry/tasks/sync.py | 29 ++- test_app/resource_api.py | 20 +- .../test_rbac_conditional.py | 206 ++++++++++++++++++ .../resource_registry/test_resource_sync.py | 2 + 7 files changed, 296 insertions(+), 23 deletions(-) create mode 100644 test_app/tests/resource_registry/test_rbac_conditional.py diff --git a/ansible_base/resource_registry/registry.py b/ansible_base/resource_registry/registry.py index f50dee126..fc1cc3b1f 100644 --- a/ansible_base/resource_registry/registry.py +++ b/ansible_base/resource_registry/registry.py @@ -1,6 +1,7 @@ from collections import namedtuple from typing import List, Optional +from django.conf import settings from django.contrib.auth import authenticate from django.utils.translation import gettext_lazy as _ @@ -23,12 +24,16 @@ class ServiceAPIConfig: This will be the interface for configuring the resource registry for each service. """ - _default_resource_processors = { - "shared.team": ResourceTypeProcessor, - "shared.organization": ResourceTypeProcessor, - "shared.user": ResourceTypeProcessor, - "shared.roledefinition": RoleDefinitionProcessor, - } + @classmethod + def _get_default_resource_processors(cls): + processors = { + "shared.team": ResourceTypeProcessor, + "shared.organization": ResourceTypeProcessor, + "shared.user": ResourceTypeProcessor, + } + if 'ansible_base.rbac' in settings.INSTALLED_APPS: + processors["shared.roledefinition"] = RoleDefinitionProcessor + return processors custom_resource_processors = {} @@ -43,7 +48,7 @@ def authenticate_local_user(username: str, password: str): @classmethod def get_processor(cls, resource_type): - combined_processors = {**cls._default_resource_processors, **cls.custom_resource_processors} + combined_processors = {**cls._get_default_resource_processors(), **cls.custom_resource_processors} return combined_processors[resource_type] diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index b363eb057..e45130f79 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -6,9 +6,17 @@ import requests import urllib3 from django.apps import apps +from django.conf import settings from ansible_base.resource_registry.resource_server import get_resource_server_config, get_service_token + +def _check_rbac_installed(): + """Check if ansible_base.rbac is installed and raise RuntimeError if not.""" + if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + raise RuntimeError("This operation requires ansible_base.rbac to be installed") + + ResourceRequestBody = namedtuple( "ResourceRequestBody", ["ansible_id", "service_id", "is_partially_migrated", "resource_type", "resource_data"], @@ -166,13 +174,16 @@ def get_resource_type_manifest(self, name, filters: Optional[dict] = None): # RBAC related methods def list_role_types(self, filters: Optional[dict] = None): + _check_rbac_installed() return self._make_request("get", "role-types/", params=filters) def list_role_permissions(self, filters: Optional[dict] = None): + _check_rbac_installed() return self._make_request("get", "role-permissions/", params=filters) def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List user role assignments.""" + _check_rbac_installed() params = (filters or {}).copy() if user_ansible_id is not None: params['user_ansible_id'] = user_ansible_id @@ -180,12 +191,14 @@ def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: def list_team_assignments(self, team_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List team role assignments.""" + _check_rbac_installed() params = (filters or {}).copy() if team_ansible_id is not None: params['team_ansible_id'] = team_ansible_id return self._make_request("get", "role-team-assignments/", params=params) def sync_assignment(self, assignment): + _check_rbac_installed() from ansible_base.rbac.service_api.serializers import ServiceRoleTeamAssignmentSerializer, ServiceRoleUserAssignmentSerializer if assignment._meta.model_name == 'roleuserassignment': @@ -196,6 +209,7 @@ def sync_assignment(self, assignment): return self._sync_assignment(serializer.data) def sync_unassignment(self, role_definition, actor, content_object): + _check_rbac_installed() data = {'role_definition': role_definition.name} data[f'{actor._meta.model_name}_ansible_id'] = str(actor.resource.ansible_id) @@ -214,6 +228,7 @@ def sync_unassignment(self, role_definition, actor, content_object): def sync_object_deletion(self, content_object): """Sync object deletion to Gateway for cleanup of all related role assignments""" + _check_rbac_installed() from ansible_base.rbac.models import DABContentType # Get the content type information diff --git a/ansible_base/resource_registry/shared_types.py b/ansible_base/resource_registry/shared_types.py index 819377b1d..66cad375c 100644 --- a/ansible_base/resource_registry/shared_types.py +++ b/ansible_base/resource_registry/shared_types.py @@ -1,7 +1,7 @@ +from django.conf import settings from rest_framework import serializers from rest_framework.exceptions import ValidationError -from ansible_base.rbac.models import DABContentType, DABPermission from ansible_base.resource_registry.utils.resource_type_serializers import AnsibleResourceForeignKeyField, SharedResourceTypeSerializer from ansible_base.resource_registry.utils.sso_provider import get_sso_provider_server @@ -84,6 +84,10 @@ class LenientPermissionSlugListField(serializers.ListField): child = serializers.CharField() def to_internal_value(self, data): + if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + raise RuntimeError("LenientPermissionSlugListField requires ansible_base.rbac to be installed") + from ansible_base.rbac.models import DABPermission + data = super().to_internal_value(data) return list(DABPermission.objects.filter(api_slug__in=data)) @@ -98,14 +102,24 @@ class RoleDefinitionType(SharedResourceTypeSerializer): name = serializers.CharField() description = serializers.CharField(default="", allow_blank=True) managed = serializers.BooleanField() - content_type = serializers.SlugRelatedField( - slug_field='api_slug', - queryset=DABContentType.objects.all(), - allow_null=True, - default=None, - ) permissions = LenientPermissionSlugListField() + def __init__(self, *args, **kwargs): + if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + raise RuntimeError("RoleDefinitionType requires ansible_base.rbac to be installed") + + super().__init__(*args, **kwargs) + + # Set up content_type field only when rbac is available + from ansible_base.rbac.models import DABContentType + + self.fields['content_type'] = serializers.SlugRelatedField( + slug_field='api_slug', + queryset=DABContentType.objects.all(), + allow_null=True, + default=None, + ) + def is_valid(self, raise_exception=False): try: return super().is_valid(raise_exception=raise_exception) diff --git a/ansible_base/resource_registry/tasks/sync.py b/ansible_base/resource_registry/tasks/sync.py index 774d1584f..7a099a392 100644 --- a/ansible_base/resource_registry/tasks/sync.py +++ b/ansible_base/resource_registry/tasks/sync.py @@ -18,7 +18,6 @@ from django.db.utils import Error, IntegrityError from requests import HTTPError -from ansible_base.rbac.models.role import AssignmentBase, RoleDefinition, RoleTeamAssignment, RoleUserAssignment from ansible_base.resource_registry.models import Resource, ResourceType from ansible_base.resource_registry.models.service_identifier import service_id from ansible_base.resource_registry.registry import get_registry @@ -26,6 +25,8 @@ logger = logging.getLogger('ansible_base.resources_api.tasks.sync') +_is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS + class ManifestNotFound(HTTPError): """Raise when server returns 404 for a manifest""" @@ -139,7 +140,9 @@ def fetch_manifest( return [ManifestItem(**row) for row in csv_reader] -def get_ansible_id_or_pk(assignment: AssignmentBase) -> str: +def get_ansible_id_or_pk(assignment) -> str: + if not _is_rbac_installed: + raise RuntimeError("get_ansible_id_or_pk requires ansible_base.rbac to be installed") # For object-scoped assignments, try to get the object's ansible_id if assignment.content_type.model in ('organization', 'team'): object_resource = Resource.objects.filter(object_id=assignment.object_id, content_type__model=assignment.content_type.model).first() @@ -153,7 +156,9 @@ def get_ansible_id_or_pk(assignment: AssignmentBase) -> str: return str(ansible_id_or_pk) -def get_content_object(role_definition: RoleDefinition, assignment_tuple: AssignmentTuple) -> Any: +def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> Any: + if not _is_rbac_installed: + raise RuntimeError("get_content_object requires ansible_base.rbac to be installed") content_object = None if role_definition.content_type.model in ('organization', 'team'): object_resource = Resource.objects.get(ansible_id=assignment_tuple.ansible_id_or_pk) @@ -167,6 +172,8 @@ def get_content_object(role_definition: RoleDefinition, assignment_tuple: Assign def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple]: """Fetch remote assignments from the resource server and convert to tuples.""" + if not _is_rbac_installed: + raise RuntimeError("get_remote_assignments requires ansible_base.rbac to be installed") assignments = set() # Fetch user assignments with pagination @@ -238,6 +245,10 @@ def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple def get_local_assignments() -> set[AssignmentTuple]: """Get local assignments and convert to tuples.""" + if not _is_rbac_installed: + raise RuntimeError("get_local_assignments requires ansible_base.rbac to be installed") + from ansible_base.rbac.models.role import RoleTeamAssignment, RoleUserAssignment + assignments = set() # Get user assignments @@ -294,6 +305,10 @@ def get_local_assignments() -> set[AssignmentTuple]: def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Delete a local assignment based on the tuple.""" + if not _is_rbac_installed: + raise RuntimeError("delete_local_assignment requires ansible_base.rbac to be installed") + from ansible_base.rbac.models.role import RoleDefinition + try: role_definition = RoleDefinition.objects.get(name=assignment_tuple.role_definition_name) @@ -320,6 +335,10 @@ def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: def create_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Create a local assignment based on the tuple.""" + if not _is_rbac_installed: + raise RuntimeError("create_local_assignment requires ansible_base.rbac to be installed") + from ansible_base.rbac.models.role import RoleDefinition + try: role_definition = RoleDefinition.objects.get(name=assignment_tuple.role_definition_name) @@ -694,6 +713,10 @@ def _sync_assignments(self): if not self.sync_assignments: return + if not _is_rbac_installed: + self.write(">>> Skipping role assignments sync (rbac not installed)") + return + self.write(">>> Syncing role assignments") try: diff --git a/test_app/resource_api.py b/test_app/resource_api.py index be514b4ab..aed0a3d4b 100644 --- a/test_app/resource_api.py +++ b/test_app/resource_api.py @@ -1,9 +1,9 @@ +from django.conf import settings from django.contrib.auth import get_user_model from ansible_base.authentication.models import Authenticator -from ansible_base.rbac.models import RoleDefinition from ansible_base.resource_registry.registry import ResourceConfig, ServiceAPIConfig, SharedResource -from ansible_base.resource_registry.shared_types import OrganizationType, RoleDefinitionType, TeamType, UserType +from ansible_base.resource_registry.shared_types import OrganizationType, TeamType, UserType from ansible_base.resource_registry.utils.resource_type_processor import ResourceTypeProcessor from test_app.models import Organization, Original1, Proxy2, ResourceMigrationTestModel, Team @@ -38,13 +38,21 @@ class APIConfig(ServiceAPIConfig): Organization, shared_resource=SharedResource(serializer=OrganizationType, is_provider=False), ), - ResourceConfig( - RoleDefinition, - shared_resource=SharedResource(serializer=RoleDefinitionType, is_provider=False), - ), # Authenticators won't be a shared resource in production, but it's a convenient model to use for testing. ResourceConfig(Authenticator), ResourceConfig(ResourceMigrationTestModel), ResourceConfig(Original1), ResourceConfig(Proxy2), ] + +# Conditionally add RoleDefinition if RBAC is installed +if 'ansible_base.rbac' in settings.INSTALLED_APPS: + from ansible_base.rbac.models import RoleDefinition + from ansible_base.resource_registry.shared_types import RoleDefinitionType + + RESOURCE_LIST.append( + ResourceConfig( + RoleDefinition, + shared_resource=SharedResource(serializer=RoleDefinitionType, is_provider=False), + ) + ) diff --git a/test_app/tests/resource_registry/test_rbac_conditional.py b/test_app/tests/resource_registry/test_rbac_conditional.py new file mode 100644 index 000000000..b5d74eefb --- /dev/null +++ b/test_app/tests/resource_registry/test_rbac_conditional.py @@ -0,0 +1,206 @@ +""" +Test suite to verify resource_registry works both with and without rbac installed. +This test module specifically tests our conditional rbac functionality. +""" + +import pytest +from django.conf import settings +from django.test import override_settings + +from ansible_base.resource_registry.registry import ServiceAPIConfig +from ansible_base.resource_registry.rest_client import ResourceAPIClient +from ansible_base.resource_registry.shared_types import LenientPermissionSlugListField, RoleDefinitionType + + +class TestResourceRegistryWithoutRBAC: + """Test resource registry functionality when rbac is NOT installed.""" + + @pytest.fixture(autouse=True) + def setup_without_rbac(self): + """Override settings to remove rbac from INSTALLED_APPS for these tests.""" + # Create a copy of INSTALLED_APPS without rbac + apps_without_rbac = [app for app in settings.INSTALLED_APPS if 'rbac' not in app] + + with override_settings(INSTALLED_APPS=apps_without_rbac): + yield + + def test_resource_api_client_rbac_methods_raise_errors(self): + """Test that RBAC-dependent methods in ResourceAPIClient raise appropriate errors.""" + client = ResourceAPIClient("http://test", "/test/") + + # Test all RBAC-dependent methods raise RuntimeError + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.list_role_types() + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.list_role_permissions() + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.list_user_assignments() + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.list_team_assignments() + + def test_resource_api_client_sync_methods_raise_errors(self): + """Test that sync methods raise appropriate errors when rbac is not available.""" + client = ResourceAPIClient("http://test", "/test/") + + # Mock assignment object + class MockAssignment: + class _meta: + model_name = 'roleuserassignment' + + def __init__(self): + self.role_definition = None # Add the required attribute + + assignment = MockAssignment() + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.sync_assignment(assignment) + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.sync_unassignment(None, None, None) + + with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): + client.sync_object_deletion(None) + + def test_lenient_permission_slug_list_field_raises_error(self): + """Test that LenientPermissionSlugListField raises error when rbac not available.""" + field = LenientPermissionSlugListField() + + with pytest.raises(RuntimeError, match="requires ansible_base.rbac to be installed"): + field.to_internal_value(['test-permission']) + + def test_role_definition_type_raises_error(self): + """Test that RoleDefinitionType raises error when rbac not available.""" + with pytest.raises(RuntimeError, match="requires ansible_base.rbac to be installed"): + RoleDefinitionType() + + def test_service_api_config_excludes_role_definition_processor(self): + """Test that ServiceAPIConfig excludes RoleDefinitionProcessor when rbac not available.""" + processors = ServiceAPIConfig._get_default_resource_processors() + + # Should not include shared.roledefinition when rbac is not installed + assert "shared.roledefinition" not in processors + + # Should still include other processors + assert "shared.user" in processors + assert "shared.team" in processors + assert "shared.organization" in processors + + @pytest.mark.django_db + def test_resource_registry_basic_functionality_works(self): + """Test that basic resource registry functionality still works without rbac.""" + from ansible_base.resource_registry.models import service_id + from ansible_base.resource_registry.registry import get_registry + + # These should work without rbac + current_service_id = service_id() + assert current_service_id is not None + + # Registry should still work + registry = get_registry() + assert registry is not False + + +class TestResourceRegistryWithRBAC: + """Test resource registry functionality when rbac IS installed (normal case).""" + + def test_resource_api_client_rbac_methods_work(self): + """Test that RBAC methods work when rbac is available.""" + # This test runs with rbac installed (default test environment) + client = ResourceAPIClient("http://test", "/test/") + + # These should not raise RuntimeError (though they may raise other errors due to network/auth) + # We're just testing that the conditional check passes + try: + client.list_role_types() + except RuntimeError as e: + if "requires ansible_base.rbac to be installed" in str(e): + pytest.fail("Should not raise rbac requirement error when rbac is installed") + except Exception: + # Other exceptions (network, auth, etc.) are expected and OK + pass + + def test_role_definition_type_works(self): + """Test that RoleDefinitionType works when rbac is available.""" + # Should not raise RuntimeError about rbac requirement + try: + serializer = RoleDefinitionType() + # Should have content_type field when rbac is available + assert 'content_type' in serializer.fields + except RuntimeError as e: + if "requires ansible_base.rbac to be installed" in str(e): + pytest.fail("Should not raise rbac requirement error when rbac is installed") + + def test_lenient_permission_slug_list_field_works(self): + """Test that LenientPermissionSlugListField works when rbac is available.""" + field = LenientPermissionSlugListField() + + # Should not raise RuntimeError about rbac requirement + # (though it may raise other validation errors) + try: + field.to_internal_value([]) + except RuntimeError as e: + if "requires ansible_base.rbac to be installed" in str(e): + pytest.fail("Should not raise rbac requirement error when rbac is installed") + except Exception: + # Other exceptions (validation, etc.) are expected and OK + pass + + def test_service_api_config_includes_role_definition_processor(self): + """Test that ServiceAPIConfig includes RoleDefinitionProcessor when rbac is available.""" + processors = ServiceAPIConfig._get_default_resource_processors() + + # Should include shared.roledefinition when rbac is installed + assert "shared.roledefinition" in processors + + # Should also include other processors + assert "shared.user" in processors + assert "shared.team" in processors + assert "shared.organization" in processors + + +class TestResourceRegistryConditionalImports: + """Test the conditional import behavior directly.""" + + def test_imports_work_with_rbac(self): + """Test that conditional imports work when rbac is available.""" + # These imports should work without errors + from ansible_base.resource_registry import registry, rest_client, shared_types + from ansible_base.resource_registry.tasks import sync + + # All modules should be importable + assert rest_client is not None + assert shared_types is not None + assert sync is not None + assert registry is not None + + @override_settings( + INSTALLED_APPS=[ + 'django.contrib.contenttypes', + 'django.contrib.auth', + 'rest_framework', + 'ansible_base.resource_registry', + 'test_app', + ] + ) + def test_imports_work_without_rbac(self): + """Test that conditional imports work when rbac is not available.""" + # Force reload modules to pick up new settings + import importlib + + from ansible_base.resource_registry import registry, rest_client, shared_types + from ansible_base.resource_registry.tasks import sync + + # Force reload to pick up the modified INSTALLED_APPS + importlib.reload(rest_client) + importlib.reload(shared_types) + importlib.reload(sync) + importlib.reload(registry) + + # All modules should still be importable + assert rest_client is not None + assert shared_types is not None + assert sync is not None + assert registry is not None diff --git a/test_app/tests/resource_registry/test_resource_sync.py b/test_app/tests/resource_registry/test_resource_sync.py index 8a9524a73..1547953fe 100644 --- a/test_app/tests/resource_registry/test_resource_sync.py +++ b/test_app/tests/resource_registry/test_resource_sync.py @@ -71,12 +71,14 @@ def write(self, text): return Stdout() +@pytest.mark.django_db def test_manifest_not_found(static_api_client, stdout): executor = SyncExecutor(api_client=static_api_client, resource_type_names=["shared.team"], stdout=stdout) executor.run() assert 'manifest for shared.team NOT FOUND.' in stdout.lines +@pytest.mark.django_db def test_raises_manifest_stream_is_unavailable(static_api_client, stdout): static_api_client.router["resource-types/shared.organization/manifest/"] = {"status_code": 500, "content": "Server Error"} with pytest.raises(ResourceSyncHTTPError): From 8affe9c21bd3fbffe396d163187f5664ef5d2caa Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 29 Sep 2025 13:41:46 -0400 Subject: [PATCH 3/7] Implement my review comments --- ansible_base/authentication/utils/claims.py | 10 ++++------ ansible_base/lib/utils/apps.py | 4 ++++ ansible_base/resource_registry/rest_client.py | 7 ++----- ansible_base/resource_registry/shared_types.py | 6 ++---- ansible_base/resource_registry/tasks/sync.py | 17 +++++++---------- 5 files changed, 19 insertions(+), 25 deletions(-) create mode 100644 ansible_base/lib/utils/apps.py diff --git a/ansible_base/authentication/utils/claims.py b/ansible_base/authentication/utils/claims.py index 1690b582b..92aea3d3b 100644 --- a/ansible_base/authentication/utils/claims.py +++ b/ansible_base/authentication/utils/claims.py @@ -19,6 +19,7 @@ from ansible_base.authentication.models import Authenticator, AuthenticatorMap, AuthenticatorUser from ansible_base.authentication.utils.authenticator_map import check_role_type, expand_syntax from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.lib.utils.auth import get_organization_model, get_team_model from ansible_base.lib.utils.string import is_empty @@ -30,9 +31,6 @@ User = get_user_model() -is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS - - class TriggerResult(Enum): ALLOW = auto() DENY = auto() @@ -723,7 +721,7 @@ def reconcile_user_claims(cls, user: AbstractUser, authenticator_user: Authentic claims = getattr(user, 'claims', authenticator_user.claims) - if is_rbac_installed: + if is_rbac_installed(): cls(claims, user, authenticator_user).manage_permissions() else: logger.info(_("Skipping user claims with RBAC roles, because RBAC app is not installed")) @@ -878,7 +876,7 @@ def __init__(self): self.cache = {} # NOTE(cutwater): We may probably execute this query once and cache the query results. self.content_types = {} - if is_rbac_installed: + if is_rbac_installed(): from ansible_base.rbac.models import DABContentType self.content_types = {content_type.model: content_type for content_type in DABContentType.objects.get_for_models(Organization, Team).values()} @@ -962,7 +960,7 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None: - Role definitions are also cached separately in self.role_definitions """ local_resource_prefixes = ["shared"] - if is_rbac_installed: + if is_rbac_installed(): from ansible_base.rbac.remote import get_local_resource_prefix local_resource_prefixes.append(get_local_resource_prefix()) diff --git a/ansible_base/lib/utils/apps.py b/ansible_base/lib/utils/apps.py new file mode 100644 index 000000000..1ebf44401 --- /dev/null +++ b/ansible_base/lib/utils/apps.py @@ -0,0 +1,4 @@ +def is_rbac_installed() -> bool: + from django.conf import settings + + return bool('ansible_base.rbac' in settings.INSTALLED_APPS) diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index e45130f79..d8024d1bd 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -8,12 +8,13 @@ from django.apps import apps from django.conf import settings +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.resource_server import get_resource_server_config, get_service_token def _check_rbac_installed(): """Check if ansible_base.rbac is installed and raise RuntimeError if not.""" - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + if is_rbac_installed(): raise RuntimeError("This operation requires ansible_base.rbac to be installed") @@ -174,16 +175,13 @@ def get_resource_type_manifest(self, name, filters: Optional[dict] = None): # RBAC related methods def list_role_types(self, filters: Optional[dict] = None): - _check_rbac_installed() return self._make_request("get", "role-types/", params=filters) def list_role_permissions(self, filters: Optional[dict] = None): - _check_rbac_installed() return self._make_request("get", "role-permissions/", params=filters) def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List user role assignments.""" - _check_rbac_installed() params = (filters or {}).copy() if user_ansible_id is not None: params['user_ansible_id'] = user_ansible_id @@ -191,7 +189,6 @@ def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: def list_team_assignments(self, team_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List team role assignments.""" - _check_rbac_installed() params = (filters or {}).copy() if team_ansible_id is not None: params['team_ansible_id'] = team_ansible_id diff --git a/ansible_base/resource_registry/shared_types.py b/ansible_base/resource_registry/shared_types.py index 66cad375c..bcef8956f 100644 --- a/ansible_base/resource_registry/shared_types.py +++ b/ansible_base/resource_registry/shared_types.py @@ -1,7 +1,7 @@ -from django.conf import settings from rest_framework import serializers from rest_framework.exceptions import ValidationError +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.utils.resource_type_serializers import AnsibleResourceForeignKeyField, SharedResourceTypeSerializer from ansible_base.resource_registry.utils.sso_provider import get_sso_provider_server @@ -84,8 +84,6 @@ class LenientPermissionSlugListField(serializers.ListField): child = serializers.CharField() def to_internal_value(self, data): - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: - raise RuntimeError("LenientPermissionSlugListField requires ansible_base.rbac to be installed") from ansible_base.rbac.models import DABPermission data = super().to_internal_value(data) @@ -105,7 +103,7 @@ class RoleDefinitionType(SharedResourceTypeSerializer): permissions = LenientPermissionSlugListField() def __init__(self, *args, **kwargs): - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + if not is_rbac_installed(): raise RuntimeError("RoleDefinitionType requires ansible_base.rbac to be installed") super().__init__(*args, **kwargs) diff --git a/ansible_base/resource_registry/tasks/sync.py b/ansible_base/resource_registry/tasks/sync.py index 7a099a392..4e2d60741 100644 --- a/ansible_base/resource_registry/tasks/sync.py +++ b/ansible_base/resource_registry/tasks/sync.py @@ -18,6 +18,7 @@ from django.db.utils import Error, IntegrityError from requests import HTTPError +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.models import Resource, ResourceType from ansible_base.resource_registry.models.service_identifier import service_id from ansible_base.resource_registry.registry import get_registry @@ -25,8 +26,6 @@ logger = logging.getLogger('ansible_base.resources_api.tasks.sync') -_is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS - class ManifestNotFound(HTTPError): """Raise when server returns 404 for a manifest""" @@ -141,7 +140,7 @@ def fetch_manifest( def get_ansible_id_or_pk(assignment) -> str: - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_ansible_id_or_pk requires ansible_base.rbac to be installed") # For object-scoped assignments, try to get the object's ansible_id if assignment.content_type.model in ('organization', 'team'): @@ -157,7 +156,7 @@ def get_ansible_id_or_pk(assignment) -> str: def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> Any: - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_content_object requires ansible_base.rbac to be installed") content_object = None if role_definition.content_type.model in ('organization', 'team'): @@ -172,8 +171,6 @@ def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> An def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple]: """Fetch remote assignments from the resource server and convert to tuples.""" - if not _is_rbac_installed: - raise RuntimeError("get_remote_assignments requires ansible_base.rbac to be installed") assignments = set() # Fetch user assignments with pagination @@ -245,7 +242,7 @@ def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple def get_local_assignments() -> set[AssignmentTuple]: """Get local assignments and convert to tuples.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_local_assignments requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleTeamAssignment, RoleUserAssignment @@ -305,7 +302,7 @@ def get_local_assignments() -> set[AssignmentTuple]: def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Delete a local assignment based on the tuple.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("delete_local_assignment requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleDefinition @@ -335,7 +332,7 @@ def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: def create_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Create a local assignment based on the tuple.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("create_local_assignment requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleDefinition @@ -713,7 +710,7 @@ def _sync_assignments(self): if not self.sync_assignments: return - if not _is_rbac_installed: + if not is_rbac_installed(): self.write(">>> Skipping role assignments sync (rbac not installed)") return From 152f5819ee7f8ccfc16311c0385d3532dfaaab55 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 29 Sep 2025 15:30:36 -0400 Subject: [PATCH 4/7] Add back missing negative --- ansible_base/resource_registry/rest_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index d8024d1bd..0dbaa74f7 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -14,7 +14,7 @@ def _check_rbac_installed(): """Check if ansible_base.rbac is installed and raise RuntimeError if not.""" - if is_rbac_installed(): + if not is_rbac_installed(): raise RuntimeError("This operation requires ansible_base.rbac to be installed") From 2fff96abc81ae1a316af964534b169de4c489c53 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 20 Oct 2025 14:21:37 -0400 Subject: [PATCH 5/7] Fix test bug --- ansible_base/lib/testing/util.py | 10 ++++++---- ansible_base/resource_registry/rest_client.py | 1 - 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/ansible_base/lib/testing/util.py b/ansible_base/lib/testing/util.py index 71d6b478d..265d23ca7 100644 --- a/ansible_base/lib/testing/util.py +++ b/ansible_base/lib/testing/util.py @@ -47,10 +47,12 @@ def delete_authenticator(authenticator): class StaticResourceAPIClient(ResourceAPIClient): """A testing API client that reads response router attribute or static files.""" - router = {} - # Route is used to force a certain status,response for a route - # It has to be a mutable default but the fixture instantiates one for - # each test. + def __init__(self, *args, **kwargs): + # Route is used to force a certain status,response for a route + # It has to be a mutable default but the fixture instantiates one for + # each test. + self.router = {} + super().__init__(*args, **kwargs) def _make_request(self, method, path, data=None, params=None, stream=False): response = Response() diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index 0dbaa74f7..f83f7c4e7 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -6,7 +6,6 @@ import requests import urllib3 from django.apps import apps -from django.conf import settings from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.resource_server import get_resource_server_config, get_service_token From 309ce1aab2b22edfa21ffc35c9243be80bb04470 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 20 Oct 2025 17:27:53 -0400 Subject: [PATCH 6/7] Remove undesired tests --- .../test_rbac_conditional.py | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/test_app/tests/resource_registry/test_rbac_conditional.py b/test_app/tests/resource_registry/test_rbac_conditional.py index b5d74eefb..66298fa97 100644 --- a/test_app/tests/resource_registry/test_rbac_conditional.py +++ b/test_app/tests/resource_registry/test_rbac_conditional.py @@ -24,23 +24,6 @@ def setup_without_rbac(self): with override_settings(INSTALLED_APPS=apps_without_rbac): yield - def test_resource_api_client_rbac_methods_raise_errors(self): - """Test that RBAC-dependent methods in ResourceAPIClient raise appropriate errors.""" - client = ResourceAPIClient("http://test", "/test/") - - # Test all RBAC-dependent methods raise RuntimeError - with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): - client.list_role_types() - - with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): - client.list_role_permissions() - - with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): - client.list_user_assignments() - - with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): - client.list_team_assignments() - def test_resource_api_client_sync_methods_raise_errors(self): """Test that sync methods raise appropriate errors when rbac is not available.""" client = ResourceAPIClient("http://test", "/test/") @@ -64,13 +47,6 @@ def __init__(self): with pytest.raises(RuntimeError, match="This operation requires ansible_base.rbac to be installed"): client.sync_object_deletion(None) - def test_lenient_permission_slug_list_field_raises_error(self): - """Test that LenientPermissionSlugListField raises error when rbac not available.""" - field = LenientPermissionSlugListField() - - with pytest.raises(RuntimeError, match="requires ansible_base.rbac to be installed"): - field.to_internal_value(['test-permission']) - def test_role_definition_type_raises_error(self): """Test that RoleDefinitionType raises error when rbac not available.""" with pytest.raises(RuntimeError, match="requires ansible_base.rbac to be installed"): From b6cd309ea61ee56ac7665ca9d90850f594df5927 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Tue, 21 Oct 2025 16:42:05 -0400 Subject: [PATCH 7/7] Add more coverage --- .../test_sync_rbac_errors.py | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 test_app/tests/resource_registry/test_sync_rbac_errors.py diff --git a/test_app/tests/resource_registry/test_sync_rbac_errors.py b/test_app/tests/resource_registry/test_sync_rbac_errors.py new file mode 100644 index 000000000..52f72eb18 --- /dev/null +++ b/test_app/tests/resource_registry/test_sync_rbac_errors.py @@ -0,0 +1,109 @@ +""" +Test coverage for RuntimeError raises in sync.py when RBAC is not installed. +Generated by Claude Sonnet 4.5 +""" + +import pytest +from django.test import override_settings + + +class TestSyncWithoutRBAC: + """Test that sync functions raise appropriate errors when rbac is NOT installed.""" + + @pytest.fixture(autouse=True) + def setup_without_rbac(self): + """Override settings to remove rbac from INSTALLED_APPS for these tests.""" + # Create a copy of INSTALLED_APPS without rbac + from django.conf import settings + + apps_without_rbac = [app for app in settings.INSTALLED_APPS if 'rbac' not in app] + + with override_settings(INSTALLED_APPS=apps_without_rbac): + yield + + def test_get_local_assignments_raises_error(self): + """Test that get_local_assignments raises error when rbac not available.""" + # Force reload to pick up the modified INSTALLED_APPS + import importlib + + from ansible_base.resource_registry.tasks import sync + + importlib.reload(sync) + + with pytest.raises(RuntimeError, match="get_local_assignments requires ansible_base.rbac to be installed"): + sync.get_local_assignments() + + def test_delete_local_assignment_raises_error(self): + """Test that delete_local_assignment raises error when rbac not available.""" + import importlib + + from ansible_base.resource_registry.tasks import sync + + importlib.reload(sync) + + assignment = sync.AssignmentTuple( + actor_ansible_id="test-id", + ansible_id_or_pk="obj-id", + role_definition_name="test-role", + assignment_type="user", + ) + + with pytest.raises(RuntimeError, match="delete_local_assignment requires ansible_base.rbac to be installed"): + sync.delete_local_assignment(assignment) + + def test_create_local_assignment_raises_error(self): + """Test that create_local_assignment raises error when rbac not available.""" + import importlib + + from ansible_base.resource_registry.tasks import sync + + importlib.reload(sync) + + assignment = sync.AssignmentTuple( + actor_ansible_id="test-id", + ansible_id_or_pk="obj-id", + role_definition_name="test-role", + assignment_type="user", + ) + + with pytest.raises(RuntimeError, match="create_local_assignment requires ansible_base.rbac to be installed"): + sync.create_local_assignment(assignment) + + def test_get_ansible_id_or_pk_raises_error(self): + """Test that get_ansible_id_or_pk raises error when rbac not available.""" + import importlib + + from ansible_base.resource_registry.tasks import sync + + importlib.reload(sync) + + # Create a mock assignment object + class MockAssignment: + content_type = None + object_id = None + + with pytest.raises(RuntimeError, match="get_ansible_id_or_pk requires ansible_base.rbac to be installed"): + sync.get_ansible_id_or_pk(MockAssignment()) + + def test_get_content_object_raises_error(self): + """Test that get_content_object raises error when rbac not available.""" + import importlib + + from ansible_base.resource_registry.tasks import sync + + importlib.reload(sync) + + # Create a mock role_definition + class MockRoleDefinition: + class content_type: + model = 'organization' + + assignment = sync.AssignmentTuple( + actor_ansible_id="test-id", + ansible_id_or_pk="obj-id", + role_definition_name="test-role", + assignment_type="user", + ) + + with pytest.raises(RuntimeError, match="get_content_object requires ansible_base.rbac to be installed"): + sync.get_content_object(MockRoleDefinition(), assignment)