Skip to content

Commit bcd92f2

Browse files
committed
Use ansible_id references in access lists
1 parent 894af11 commit bcd92f2

File tree

3 files changed

+224
-1
lines changed

3 files changed

+224
-1
lines changed

ansible_base/rbac/api/serializers.py

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import logging
2+
13
from django.apps import apps
24
from django.core.exceptions import ObjectDoesNotExist
35
from django.db import transaction
@@ -20,6 +22,8 @@
2022
from ..remote import RemoteObject
2123
from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
2224

25+
logger = logging.getLogger(__name__)
26+
2327

2428
class RoleDefinitionSerializer(CommonModelSerializer):
2529
permissions = serializers.SlugRelatedField(
@@ -257,9 +261,20 @@ def _get_related(self, obj) -> dict[str, str]:
257261
return {}
258262
related_fields = {}
259263
actor_cls = self.Meta.model
264+
265+
# Use ansible_id if available, otherwise fall back to pk
266+
actor_identifier = obj.pk
267+
try:
268+
if hasattr(obj, 'resource') and obj.resource:
269+
actor_identifier = str(obj.resource.ansible_id)
270+
except ObjectDoesNotExist:
271+
# Resource doesn't exist, stick with pk
272+
logger.error(f"Resource does not exist for {self.Meta.model} {obj.pk}")
273+
pass
274+
260275
related_fields['details'] = get_relative_url(
261276
f'role-{actor_cls._meta.model_name}-access-assignments',
262-
kwargs={'model_name': self.context.get("content_type").api_slug, 'pk': self.context.get("related_object").pk, 'actor_pk': obj.pk},
277+
kwargs={'model_name': self.context.get("content_type").api_slug, 'pk': self.context.get("related_object").pk, 'actor_pk': actor_identifier},
263278
)
264279
return related_fields
265280

ansible_base/rbac/api/views.py

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
1+
import uuid
12
from collections import OrderedDict
23
from typing import Type
34

5+
from django.apps import apps
6+
from django.core.exceptions import ObjectDoesNotExist
47
from django.db import transaction
58
from django.db.models import Model
69
from django.utils.translation import gettext_lazy as _
@@ -379,6 +382,28 @@ class UserAccessAssignmentViewSet(
379382
def get_url_actor(self):
380383
actor_pk = self.kwargs.get("actor_pk")
381384
actor_cls = self.get_actor_model()
385+
386+
# First, try to parse as UUID for ansible_id lookup
387+
try:
388+
parsed_uuid = uuid.UUID(actor_pk)
389+
# It's a valid UUID, try resource lookup first
390+
try:
391+
resource_cls = apps.get_model('dab_resource_registry', 'Resource')
392+
resource = resource_cls.objects.get(ansible_id=parsed_uuid)
393+
actor = resource.content_object
394+
# Verify the content object is the correct type
395+
if isinstance(actor, actor_cls):
396+
return actor
397+
else:
398+
raise NotFound(f'Resource with ansible_id {parsed_uuid} is not a {actor_cls._meta.model_name}')
399+
except (LookupError, ObjectDoesNotExist):
400+
# Resource registry not available or resource not found with this UUID
401+
raise NotFound(f'The {actor_cls._meta.model_name} with ansible_id={actor_pk} can not be found')
402+
except ValueError:
403+
# Not a valid UUID, continue with primary key lookup
404+
pass
405+
406+
# Fallback to primary key lookup (only for non-UUID values)
382407
try:
383408
return actor_cls.objects.get(pk=actor_pk)
384409
except actor_cls.DoesNotExist:
Lines changed: 183 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,183 @@
1+
import uuid
2+
3+
import pytest
4+
5+
from ansible_base.lib.utils.response import get_relative_url
6+
from test_app.models import Team, User
7+
8+
9+
@pytest.mark.django_db
10+
class TestUserAccessAssignmentUUIDLookup:
11+
"""Test UUID-based lookup for UserAccessAssignmentViewSet"""
12+
13+
def test_get_url_actor_with_primary_key(self, admin_api_client, inventory, inv_rd):
14+
"""Test existing functionality - lookup by primary key should still work"""
15+
user = User.objects.create(username='test-user')
16+
inv_rd.give_permission(user, inventory)
17+
18+
url = get_relative_url('role-user-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': str(user.pk)})
19+
20+
response = admin_api_client.get(url)
21+
assert response.status_code == 200
22+
assert response.data['count'] >= 1
23+
24+
def test_get_url_actor_with_ansible_id(self, admin_api_client, inventory, inv_rd):
25+
"""Test new functionality - lookup by ansible_id (UUID)"""
26+
user = User.objects.create(username='test-user')
27+
inv_rd.give_permission(user, inventory)
28+
29+
# Get the user's ansible_id from their resource
30+
user_ansible_id = user.resource.ansible_id
31+
32+
url = get_relative_url('role-user-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': str(user_ansible_id)})
33+
34+
response = admin_api_client.get(url)
35+
assert response.status_code == 200
36+
assert response.data['count'] >= 1
37+
38+
def test_get_url_actor_with_invalid_uuid(self, admin_api_client, inventory):
39+
"""Test error handling for invalid UUID"""
40+
invalid_uuid = str(uuid.uuid4())
41+
42+
url = get_relative_url('role-user-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': invalid_uuid})
43+
44+
response = admin_api_client.get(url)
45+
assert response.status_code == 404
46+
assert 'can not be found' in response.data['detail']
47+
48+
def test_get_url_actor_with_nonexistent_pk(self, admin_api_client, inventory):
49+
"""Test error handling for non-existent primary key"""
50+
nonexistent_pk = '99999'
51+
52+
url = get_relative_url('role-user-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': nonexistent_pk})
53+
54+
response = admin_api_client.get(url)
55+
assert response.status_code == 404
56+
assert 'can not be found' in response.data['detail']
57+
58+
59+
@pytest.mark.django_db
60+
class TestTeamAccessAssignmentUUIDLookup:
61+
"""Test UUID-based lookup for TeamAccessAssignmentViewSet"""
62+
63+
def test_get_url_actor_with_primary_key(self, admin_api_client, inventory, inv_rd, organization):
64+
"""Test existing functionality - lookup by primary key should still work"""
65+
team = Team.objects.create(name='test-team', organization=organization)
66+
inv_rd.give_permission(team, inventory)
67+
68+
url = get_relative_url('role-team-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': str(team.pk)})
69+
70+
response = admin_api_client.get(url)
71+
assert response.status_code == 200
72+
assert response.data['count'] >= 1
73+
74+
def test_get_url_actor_with_ansible_id(self, admin_api_client, inventory, inv_rd, organization):
75+
"""Test new functionality - lookup by ansible_id (UUID)"""
76+
team = Team.objects.create(name='test-team', organization=organization)
77+
inv_rd.give_permission(team, inventory)
78+
79+
# Get the team's ansible_id from their resource
80+
team_ansible_id = team.resource.ansible_id
81+
82+
url = get_relative_url('role-team-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': str(team_ansible_id)})
83+
84+
response = admin_api_client.get(url)
85+
assert response.status_code == 200
86+
assert response.data['count'] >= 1
87+
88+
def test_get_url_actor_with_invalid_uuid(self, admin_api_client, inventory):
89+
"""Test error handling for invalid UUID"""
90+
invalid_uuid = str(uuid.uuid4())
91+
92+
url = get_relative_url('role-team-access-assignments', kwargs={'model_name': 'aap.inventory', 'pk': inventory.pk, 'actor_pk': invalid_uuid})
93+
94+
response = admin_api_client.get(url)
95+
assert response.status_code == 404
96+
assert 'can not be found' in response.data['detail']
97+
98+
99+
@pytest.mark.django_db
100+
class TestAccessListRelatedLinksWithAnsibleID:
101+
"""Test that related links in access list serializers use ansible_id when available"""
102+
103+
def test_user_access_list_related_links_use_ansible_id(self, admin_api_client, inventory, inv_rd):
104+
"""Test that user access list provides related links with ansible_id"""
105+
user = User.objects.create(username='test-user')
106+
inv_rd.give_permission(user, inventory)
107+
108+
url = get_relative_url('role-user-access', kwargs={'pk': inventory.pk, 'model_name': 'aap.inventory'})
109+
response = admin_api_client.get(url)
110+
assert response.status_code == 200
111+
112+
# Find our test user in the results
113+
user_data = None
114+
for user_detail in response.data['results']:
115+
if user_detail['username'] == 'test-user':
116+
user_data = user_detail
117+
break
118+
119+
assert user_data is not None
120+
assert 'related' in user_data
121+
assert 'details' in user_data['related']
122+
123+
# The details URL should use ansible_id instead of primary key
124+
expected_ansible_id = str(user.resource.ansible_id)
125+
assert expected_ansible_id in user_data['related']['details']
126+
# Check that the URL ends with the ansible_id, not the primary key
127+
assert user_data['related']['details'].endswith(f'{expected_ansible_id}/')
128+
assert not user_data['related']['details'].endswith(f'{user.pk}/')
129+
130+
def test_team_access_list_related_links_use_ansible_id(self, admin_api_client, inventory, inv_rd, organization):
131+
"""Test that team access list provides related links with ansible_id"""
132+
team = Team.objects.create(name='test-team', organization=organization)
133+
inv_rd.give_permission(team, inventory)
134+
135+
url = get_relative_url('role-team-access', kwargs={'pk': inventory.pk, 'model_name': 'aap.inventory'})
136+
response = admin_api_client.get(url)
137+
assert response.status_code == 200
138+
139+
# Find our test team in the results
140+
team_data = None
141+
for team_detail in response.data['results']:
142+
if team_detail['name'] == 'test-team':
143+
team_data = team_detail
144+
break
145+
146+
assert team_data is not None
147+
assert 'related' in team_data
148+
assert 'details' in team_data['related']
149+
150+
# The details URL should use ansible_id instead of primary key
151+
expected_ansible_id = str(team.resource.ansible_id)
152+
assert expected_ansible_id in team_data['related']['details']
153+
# Check that the URL ends with the ansible_id, not the primary key
154+
assert team_data['related']['details'].endswith(f'{expected_ansible_id}/')
155+
assert not team_data['related']['details'].endswith(f'{team.pk}/')
156+
157+
def test_user_access_list_fallback_to_pk_when_no_resource(self, admin_api_client, inventory, inv_rd):
158+
"""Test fallback to primary key when user has no resource (edge case)"""
159+
user = User.objects.create(username='test-user-no-resource')
160+
161+
# Manually delete the resource to simulate edge case
162+
if hasattr(user, 'resource') and user.resource:
163+
user.resource.delete()
164+
165+
inv_rd.give_permission(user, inventory)
166+
167+
url = get_relative_url('role-user-access', kwargs={'pk': inventory.pk, 'model_name': 'aap.inventory'})
168+
response = admin_api_client.get(url)
169+
assert response.status_code == 200
170+
171+
# Find our test user in the results
172+
user_data = None
173+
for user_detail in response.data['results']:
174+
if user_detail['username'] == 'test-user-no-resource':
175+
user_data = user_detail
176+
break
177+
178+
assert user_data is not None
179+
assert 'related' in user_data
180+
assert 'details' in user_data['related']
181+
182+
# Should fall back to primary key when no resource available
183+
assert str(user.pk) in user_data['related']['details']

0 commit comments

Comments
 (0)