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 diff --git a/ansible_base/resource_registry/models/service_identifier.py b/ansible_base/resource_registry/models/service_identifier.py index 739499f9e..35d768063 100644 --- a/ansible_base/resource_registry/models/service_identifier.py +++ b/ansible_base/resource_registry/models/service_identifier.py @@ -23,5 +23,11 @@ def save(self, *args, **kwargs): def service_id(): global _service_id if not _service_id: - _service_id = str(ServiceID.objects.first().pk) + service_obj = ServiceID.objects.first() + if service_obj: + _service_id = str(service_obj.pk) + else: + # Create a ServiceID if none exists + service_obj = ServiceID.objects.create() + _service_id = str(service_obj.pk) return _service_id 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):