538_API_share_with_ids #539
@@ -30,4 +30,22 @@ class ExternalIdentifier(models.Model):
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.external_id} -> {self.internal_id}"
|
||||
return f"{self.external_id} -> {self.internal_id}"
|
||||
|
||||
@staticmethod
|
||||
def resolve_external_identifier(external_identifier: str):
|
||||
""" Returns a ExternalIdentifier object, if the given parameter could be resolved as an external identifier.
|
||||
|
||||
Args:
|
||||
external_identifier (str): An external identifier.
|
||||
|
||||
Returns:
|
||||
ExternalIdentifier | None
|
||||
"""
|
||||
if external_identifier:
|
||||
try:
|
||||
obj = ExternalIdentifier.objects.get(external_id=external_identifier)
|
||||
return obj
|
||||
except ExternalIdentifier.DoesNotExist:
|
||||
pass
|
||||
return None
|
||||
|
||||
@@ -31,9 +31,10 @@ class APIV1SharingTestCase(BaseAPIV1TestCase):
|
||||
def setUpTestData(cls):
|
||||
super().setUpTestData()
|
||||
|
||||
def _run_share_request(self, url, user_list: list):
|
||||
def _run_share_request(self, url, user_list: list, team_list: list):
|
||||
data = {
|
||||
"users": user_list
|
||||
"users": user_list,
|
||||
"teams": team_list
|
||||
}
|
||||
data = json.dumps(data)
|
||||
response = self.client.put(
|
||||
@@ -58,16 +59,22 @@ class APIV1SharingTestCase(BaseAPIV1TestCase):
|
||||
self.superuser.username,
|
||||
self.user.username,
|
||||
]
|
||||
team_list = [
|
||||
str(self.team.id),
|
||||
]
|
||||
|
||||
response = self._run_share_request(url, user_list)
|
||||
response = self._run_share_request(url, user_list, team_list)
|
||||
|
||||
# Must fail, since performing user has no access on requested object
|
||||
self.assertEqual(response.status_code, 500)
|
||||
self.assertTrue(len(json.loads(response.content.decode("utf-8")).get("errors", [])) > 0)
|
||||
|
||||
# Add performing user to shared access users and rerun the request
|
||||
# Add performing user to shared access users, switch from team id to team name and rerun the request
|
||||
obj.users.add(self.superuser)
|
||||
response = self._run_share_request(url, user_list)
|
||||
team_list = [
|
||||
self.team.name
|
||||
]
|
||||
response = self._run_share_request(url, user_list, team_list)
|
||||
|
||||
shared_users = obj.shared_users
|
||||
self.assertEqual(response.status_code, 200)
|
||||
@@ -84,14 +91,14 @@ class APIV1SharingTestCase(BaseAPIV1TestCase):
|
||||
share_url = reverse("api:v1:intervention-share", args=(self.intervention.id,))
|
||||
# Expect the first request to work properly
|
||||
self.intervention.users.add(self.superuser)
|
||||
response = self._run_share_request(share_url, [self.superuser.username])
|
||||
response = self._run_share_request(share_url, [self.superuser.username], [str(self.team.id)])
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Change the token
|
||||
self.header_data["HTTP_ksptoken"] = f"{self.superuser.api_token.token}__X"
|
||||
|
||||
# Expect the request to fail now
|
||||
response = self._run_share_request(share_url, [self.superuser.username])
|
||||
response = self._run_share_request(share_url, [self.superuser.username], [])
|
||||
self.assertEqual(response.status_code, 403)
|
||||
|
||||
def test_api_intervention_sharing(self):
|
||||
@@ -144,11 +151,11 @@ class APIV1SharingTestCase(BaseAPIV1TestCase):
|
||||
self.assertEqual(self.intervention.users.count(), 1)
|
||||
|
||||
# Try to add another user via API -> must work!
|
||||
response = self._run_share_request(share_url, [self.superuser.username, self.user.username])
|
||||
response = self._run_share_request(share_url, [self.superuser.username, self.user.username], [])
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(self.intervention.users.count(), 2)
|
||||
|
||||
# Now try to remove the user again -> expect no changes at all to the shared user list
|
||||
response = self._run_share_request(share_url, [self.superuser.username])
|
||||
response = self._run_share_request(share_url, [self.superuser.username], [])
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(self.intervention.users.count(), 2)
|
||||
|
||||
@@ -179,13 +179,7 @@ class AbstractModelAPISerializerV1(AbstractModelAPISerializer):
|
||||
Returns:
|
||||
ExternalIdentifier | None
|
||||
"""
|
||||
if external_identifier:
|
||||
try:
|
||||
obj = ExternalIdentifier.objects.get(external_id=external_identifier)
|
||||
return obj
|
||||
except ObjectDoesNotExist:
|
||||
pass
|
||||
return None
|
||||
return ExternalIdentifier.resolve_external_identifier(external_identifier)
|
||||
|
||||
def _check_external_identifier_on_entry_creation(self, external_identifier):
|
||||
""" Special check for POST processing:
|
||||
|
||||
@@ -6,13 +6,14 @@ Created on: 21.01.22
|
||||
|
||||
"""
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from django.db.models import QuerySet
|
||||
from django.db.models import QuerySet, Q
|
||||
from django.http import JsonResponse, HttpRequest
|
||||
from django.views import View
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from api.models import APIUserToken
|
||||
from api.models import APIUserToken, ExternalIdentifier
|
||||
from api.settings import KSP_TOKEN_HEADER_IDENTIFIER, KSP_USER_HEADER_IDENTIFIER
|
||||
from compensation.models import EcoAccount
|
||||
from ema.models import Ema
|
||||
@@ -205,6 +206,10 @@ class AbstractModelShareAPIView(AbstractAPIView):
|
||||
|
||||
"""
|
||||
try:
|
||||
external_identifier = ExternalIdentifier.resolve_external_identifier(id)
|
||||
if external_identifier:
|
||||
id = external_identifier.internal_id
|
||||
|
||||
users = self._get_shared_users_of_object(id)
|
||||
teams = self._get_shared_teams_of_object(id)
|
||||
except Exception as e:
|
||||
@@ -237,6 +242,9 @@ class AbstractModelShareAPIView(AbstractAPIView):
|
||||
"""
|
||||
|
||||
try:
|
||||
external_identifier = ExternalIdentifier.resolve_external_identifier(id)
|
||||
if external_identifier:
|
||||
id = external_identifier.internal_id
|
||||
success = self._process_put_body(request.body, id)
|
||||
except Exception as e:
|
||||
return self._return_error_response(e)
|
||||
@@ -361,26 +369,29 @@ class AbstractModelShareAPIView(AbstractAPIView):
|
||||
# Eliminate duplicates
|
||||
new_teams = list(dict.fromkeys(team_list))
|
||||
|
||||
# Make sure each of these names exist as a team
|
||||
new_teams_objs = []
|
||||
for team_name in new_teams:
|
||||
# Resolve team names or ids into objects
|
||||
new_team_ids = []
|
||||
for team in new_teams:
|
||||
try:
|
||||
team_obj = Team.objects.get(name=team_name)
|
||||
except Team.DoesNotExist:
|
||||
raise AssertionError(f"Team {team_name} does not exist!")
|
||||
new_teams_objs.append(team_obj)
|
||||
uuid.UUID(team)
|
||||
try:
|
||||
new_team_ids.append(Team.objects.get(id=team).id)
|
||||
except Team.DoesNotExist:
|
||||
raise AssertionError(f"Team with id {team} does not exist!")
|
||||
except ValueError:
|
||||
# entry is a name and not a uuid -> try to resolve as name!
|
||||
try:
|
||||
new_team_ids.append(Team.objects.get(name=team).id)
|
||||
except Team.DoesNotExist:
|
||||
raise AssertionError(f"Team with name {team} does not exist!")
|
||||
|
||||
new_team_objs = Team.objects.filter(id__in=new_team_ids)
|
||||
if self.user.is_default_group_only():
|
||||
# Default only users are not allowed to remove other users from having access. They can only add new ones!
|
||||
# So we need to keep the ones that already have access from being removed!
|
||||
new_teams_to_be_added = Team.objects.filter(
|
||||
name__in=new_teams
|
||||
).exclude(
|
||||
id__in=obj.shared_teams
|
||||
)
|
||||
new_teams_objs = obj.shared_teams.union(new_teams_to_be_added)
|
||||
new_team_objs = obj.shared_teams.union(new_team_objs)
|
||||
|
||||
obj.share_with_team_list(new_teams_objs)
|
||||
obj.share_with_team_list(new_team_objs)
|
||||
|
||||
class InterventionAPIShareView(AbstractModelShareAPIView):
|
||||
model = Intervention
|
||||
|
||||
Reference in New Issue
Block a user