Move URLs for RBAC service endpoints

Author: AlanCoding

ansible_base/rbac/models/content_type.py

@@ -10,7 +10,7 @@
 from ..remote import RemoteObject, get_local_resource_prefix, get_resource_prefix
 
 
-class DABContentTypeManager(django_models.Manager["DABContentType"]):
+class DABContentTypeManager(django_models.Manager[django_models.Model]):
     """Manager storing DABContentType objects in a local cache like original ContentType.
 
     The major structural difference is that the cache keys have to add the service reference.
@@ -20,23 +20,23 @@ class DABContentTypeManager(django_models.Manager["DABContentType"]):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self._cache: Dict[str, Dict[Union[Tuple[str, str, str], int], "DABContentType"]] = {}
+        self._cache: Dict[str, Dict[Union[Tuple[str, str, str], int], django_models.Model]] = {}
 
     def clear_cache(self) -> None:
         self._cache.clear()
 
-    def create(self, *args: Any, **kwargs: Any) -> "DABContentType":
+    def create(self, *args: Any, **kwargs: Any) -> django_models.Model:
         obj = super().create(*args, **kwargs)
         self._add_to_cache(self.db, obj)
         return obj
 
-    def _add_to_cache(self, using: str, ct: "DABContentType") -> None:
+    def _add_to_cache(self, using: str, ct: django_models.Model) -> None:
         """Store ``ct`` in the manager cache for the given database alias."""
         key = (ct.service, ct.app_label, ct.model)
         self._cache.setdefault(using, {})[key] = ct
         self._cache.setdefault(using, {})[ct.id] = ct
 
-    def _get_from_cache(self, opts: Options, service: str) -> "DABContentType":
+    def _get_from_cache(self, opts: Options, service: str) -> django_models.Model:
         """Return a cached ``DABContentType`` for ``opts`` and ``service``."""
         key = (service, opts.app_label, opts.model_name)
         return self._cache[self.db][key]
@@ -50,7 +50,7 @@ def get_for_model(
         model: Union[Type[django_models.Model], django_models.Model, RemoteObject, Type[RemoteObject]],
         for_concrete_model: bool = True,
         service: Optional[str] = None,
-    ) -> "DABContentType":
+    ) -> django_models.Model:
         # Is a remote object, we only know of these objects by virtue of their content type
         if isinstance(model, RemoteObject):
             ct = model.content_type
@@ -84,15 +84,15 @@ def get_for_models(
         *model_list: Union[Type[django_models.Model], django_models.Model],
         for_concrete_models: bool = True,
         service: Optional[str] = None,
-    ) -> Dict[Type[django_models.Model], "DABContentType"]:
+    ) -> Dict[Type[django_models.Model], django_models.Model]:
         """Return ``DABContentType`` objects for each model in ``model_list``.
 
         This gets deep into the customization of unique rules for DAB RBAC.
         We require that model_name must be unique for a given service,
         and this will rely on that assumption, which compares to app_label
         in the original ContentType model.
         """
-        results: Dict[Type[django_models.Model], "DABContentType"] = {}
+        results: Dict[Type[django_models.Model], django_models.Model] = {}
         # A keyed by (service, app_name) unlike Django where it was just app_name
         needed_models: Dict[Tuple[str, str], set[str]] = defaultdict(set)
         # A dict of (service, app_name, model_name), differs from Django ContentType
@@ -135,8 +135,12 @@ def get_for_models(
                 )
         return results
 
-    def get_by_natural_key(self, *args: str) -> "DABContentType":
-        """Return the content type identified by its natural key."""
+    def get_by_natural_key(self, *args: str) -> django_models.Model:
+        """Return the content type identified by its natural key.
+
+        Note that we can not type hint the return value fully because it is used in migrations.
+        Migrations will return a prior model state.
+        """
         if len(args) == 2:
             service = get_local_resource_prefix()
             app_label, model = args
@@ -160,7 +164,7 @@ def get_by_natural_key(self, *args: str) -> "DABContentType":
             self._add_to_cache(self.db, ct)
             return ct
 
-    def get_for_id(self, id: int) -> "DABContentType":
+    def get_for_id(self, id: int) -> django_models.Model:
         """Return the content type with primary key ``id`` from the cache."""
         try:
             return self._cache[self.db][id]

ansible_base/rbac/service_api/urls.py

@@ -0,0 +1,8 @@
+from django.urls import include, path
+
+from .router import service_router
+
+# These will be included by the resource registry
+rbac_service_urls = [
+    path('', include(service_router.urls)),
+]

ansible_base/rbac/urls.py

@@ -4,21 +4,13 @@
 from ansible_base.rbac.api.views import RoleMetadataView, TeamAccessViewSet, UserAccessViewSet
 from ansible_base.rbac.apps import AnsibleRBACConfig
 
-from .service_api.router import service_router
-
 app_name = AnsibleRBACConfig.label
 
-
-service_urls = [
-    path('', include(service_router.urls)),
-]
-
 user_access_view = UserAccessViewSet.as_view({'get': 'list'})
 team_access_view = TeamAccessViewSet.as_view({'get': 'list'})
 
 api_version_urls = [
     path('', include(router.urls)),
-    path('service-index/', include(service_urls)),
     path(r'role_metadata/', RoleMetadataView.as_view(), name="role-metadata"),
     path('role_user_access/<str:model_name>/<int:pk>/', user_access_view, name="role-user-access"),
     path('role_team_access/<str:model_name>/<int:pk>/', team_access_view, name="role-team-access"),

ansible_base/resource_registry/urls.py

@@ -3,6 +3,7 @@
 from django.urls import include, path
 from rest_framework import routers
 
+from ansible_base.rbac.service_api.urls import rbac_service_urls
 from ansible_base.resource_registry import views
 
 logger = logging.getLogger('ansible_base.resource-urls')
@@ -21,4 +22,5 @@
 
 urlpatterns = [
     path('service-index/', include(service)),
+    path('service-index/', include(rbac_service_urls)),
 ]

ansible_base/resource_registry/views.py

@@ -183,6 +183,11 @@ def get(self, request, format=None):
         data['metadata'] = get_relative_url('service-metadata')
         data['resources'] = get_relative_url('resource-list')
         data['resource-types'] = get_relative_url('resourcetype-list')
+        if 'ansible_base.rbac' in settings.INSTALLED_APPS:
+            data['role-types'] = get_relative_url('dabcontenttype-list')
+            data['role-permissions'] = get_relative_url('dabpermission-list')
+            data['role-user-assignments'] = get_relative_url('serviceuserassignment-list')
+            data['role-team-assignments'] = get_relative_url('serviceteamassignment-list')
         return Response(data)
 
 

test_app/tests/rbac/api/test_user_permissions.py

@@ -202,19 +202,19 @@ def test_team_admins_can_add_children(self, user, user_api_client, organization,
         member_rd.give_permission(user, child_team)
         admin_rd.give_permission(user, child_team)
         response = user_api_client.post(url, data=data)
-        assert response.status_code == 400
+        assert response.status_code == 400, response.data
         assert 'object does not exist' in response.data['team'][0]
         admin_rd.remove_permission(user, child_team)  # hacky, need to test (1) in isolation of (2)
 
         # (2) user does not have admin permissions to the target (child) team, cannot make assignment
         member_rd.give_permission(user, parent_team)
         response = user_api_client.post(url, data=data)
-        assert response.status_code == 403
+        assert response.status_code == 403, response.data
 
         # (3) with admin permission to child team and view permission to parent, can make assignment
         admin_rd.give_permission(user, child_team)
         response = user_api_client.post(url, data=data)
-        assert response.status_code == 201
+        assert response.status_code == 201, response.data
         assert rando.has_obj_perm(inventory, 'change')