stash
This commit is contained in:
parent
6ee5ae38b1
commit
9685b4020e
11 changed files with 331 additions and 268 deletions
|
|
@ -2,7 +2,7 @@ from nacl.exceptions import BadSignatureError
|
|||
from nacl.signing import VerifyKey
|
||||
from rest_framework import authentication
|
||||
|
||||
from authentication.models import KnownIdentity, ToolshedUser
|
||||
from authentication.models import Group, KnownIdentity, ToolshedUser
|
||||
|
||||
|
||||
def split_userhandle_or_throw(userhandle):
|
||||
|
|
@ -22,6 +22,21 @@ def split_grouphandle_or_throw(grouphandle):
|
|||
return split_userhandle_or_throw(grouphandle[1:])
|
||||
|
||||
|
||||
def resolve_owner_handle(handle):
|
||||
"""Resolves a "user@domain" or "+name@domain" handle to (owner_user, owner_group), exactly one set. Raises ValueError if the handle doesn't parse at all (no '@'); returns (None, None) if it parses but names nothing that exists."""
|
||||
is_group = handle.startswith('+')
|
||||
name, domain = split_userhandle_or_throw(handle[1:] if is_group else handle)
|
||||
if is_group:
|
||||
try:
|
||||
return None, Group.objects.get(name=name, domain=domain)
|
||||
except Group.DoesNotExist:
|
||||
return None, None
|
||||
try:
|
||||
return ToolshedUser.objects.get(username=name, domain=domain), None
|
||||
except ToolshedUser.DoesNotExist:
|
||||
return None, None
|
||||
|
||||
|
||||
def verify_request(request, raw_request_body):
|
||||
authentication_header = request.META.get('HTTP_AUTHORIZATION')
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
from django.db.models import Q
|
||||
from django.urls import path
|
||||
from rest_framework import status
|
||||
from rest_framework.decorators import api_view, permission_classes, authentication_classes
|
||||
|
|
|
|||
|
|
@ -1,11 +1,13 @@
|
|||
from django.urls import path
|
||||
from rest_framework.decorators import api_view, authentication_classes, permission_classes
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import APIView
|
||||
from rest_framework.viewsets import ViewSetMixin
|
||||
|
||||
from authentication.models import KnownIdentity
|
||||
from authentication.models import KnownIdentity, ToolshedUser, Group
|
||||
from authentication.signature_auth import SignatureAuthentication
|
||||
from toolshed.models import InventoryItem, StorageLocation
|
||||
from toolshed.serializers import FriendSerializer, GroupIdMapSerializer
|
||||
|
||||
|
||||
|
|
@ -23,6 +25,45 @@ class IdMap(APIView, ViewSetMixin):
|
|||
})
|
||||
|
||||
|
||||
@api_view(['GET'])
|
||||
@authentication_classes([SignatureAuthentication])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def resolve_short_id(request, kind, owner_id, local_id):
|
||||
"""Resolves a short id's (kind, owner_id, local_id) against this backend's own numbering. See docs/implementation.md#domain-qualified-short-id-resolution."""
|
||||
if kind in ('item', 'storage_location'):
|
||||
try:
|
||||
owner = KnownIdentity.objects.get(pk=owner_id).user.get()
|
||||
except (KnownIdentity.DoesNotExist, ToolshedUser.DoesNotExist):
|
||||
return Response(status=404)
|
||||
if owner not in request.user.friends_or_self():
|
||||
return Response(status=403)
|
||||
model = InventoryItem if kind == 'item' else StorageLocation
|
||||
try:
|
||||
obj = model.objects.get(owner=owner, id=local_id)
|
||||
except model.DoesNotExist:
|
||||
return Response(status=404)
|
||||
is_owner = request.user.user.filter(pk=owner.pk).exists()
|
||||
if getattr(obj, 'availability_policy', 'share') == 'private' and not is_owner:
|
||||
return Response(status=403)
|
||||
return Response({'handle': f'{owner.username}@{owner.domain}', 'id': obj.id})
|
||||
if kind in ('group_item', 'group_storage_location'):
|
||||
try:
|
||||
group = Group.objects.get(pk=owner_id)
|
||||
except Group.DoesNotExist:
|
||||
return Response(status=404)
|
||||
if not group.is_member(request.user):
|
||||
return Response(status=403)
|
||||
model = InventoryItem if kind == 'group_item' else StorageLocation
|
||||
try:
|
||||
obj = model.objects.get(owner_group=group, id=local_id)
|
||||
except model.DoesNotExist:
|
||||
return Response(status=404)
|
||||
return Response({'handle': str(group), 'id': obj.id})
|
||||
return Response(status=400)
|
||||
|
||||
|
||||
urlpatterns = [
|
||||
path('idmap/', IdMap.as_view(), name='idmap'),
|
||||
path('resolve_short_id/<str:kind>/<int:owner_id>/<int:local_id>/', resolve_short_id,
|
||||
name='resolve_short_id'),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -2,33 +2,16 @@ from django.db import transaction
|
|||
from django.urls import path
|
||||
from rest_framework import routers, viewsets, status
|
||||
from rest_framework.decorators import authentication_classes, api_view, permission_classes, action
|
||||
from rest_framework.exceptions import NotFound, PermissionDenied
|
||||
from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
|
||||
from authentication.models import ToolshedUser, KnownIdentity, Group
|
||||
from authentication.signature_auth import SignatureAuthentication, split_userhandle_or_throw
|
||||
from authentication.models import ToolshedUser, KnownIdentity
|
||||
from authentication.signature_auth import SignatureAuthentication, resolve_owner_handle
|
||||
from files.models import File
|
||||
from toolshed.models import InventoryItem, StorageLocation, WorkflowInstance
|
||||
from toolshed.serializers import InventoryItemSerializer, StorageLocationSerializer, WorkflowInstanceSerializer
|
||||
|
||||
router = routers.SimpleRouter()
|
||||
|
||||
|
||||
def resolve_group_by_handle(handle):
|
||||
"""handle is "name@domain" (no leading '#') - the same format/parser group.py's GroupDetail
|
||||
uses, so a group reference parses identically everywhere it appears (URL path, ?group= query
|
||||
param, or an owner_group payload field), rather than some spots taking a handle and others a
|
||||
bare pk."""
|
||||
try:
|
||||
name, domain = split_userhandle_or_throw(handle)
|
||||
except ValueError:
|
||||
return None
|
||||
try:
|
||||
return Group.objects.get(name=name, domain=domain)
|
||||
except Group.DoesNotExist:
|
||||
return None
|
||||
|
||||
|
||||
def inventory_items(identity):
|
||||
try:
|
||||
|
|
@ -50,47 +33,48 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
|
|||
serializer_class = InventoryItemSerializer
|
||||
authentication_classes = [SignatureAuthentication]
|
||||
permission_classes = [IsAuthenticated]
|
||||
# Detail routes address an item by its owner-scoped id, not the internal row id. See
|
||||
# docs/implementation.md#inventory-detail-routes-use-owner-scoped-ids.
|
||||
# Detail routes address an item by its owner-scoped id, not the internal row id. See docs/implementation.md#owner-handle-scoped-routes.
|
||||
lookup_field = 'id'
|
||||
lookup_url_kwarg = 'pk'
|
||||
|
||||
def get_queryset(self):
|
||||
# A pure group-member KnownIdentity may have no local ToolshedUser account; only
|
||||
# personal items require .user.exists(). See
|
||||
# docs/implementation.md#group-member-identities-without-local-accounts.
|
||||
# Every route is scoped by the owner handle URL param (own/friend/group); no handle-less fallback exists. See docs/implementation.md#owner-handle-scoped-routes.
|
||||
if type(self.request.user) != KnownIdentity:
|
||||
return InventoryItem.objects.none()
|
||||
identity = self.request.user
|
||||
group_items = InventoryItem.objects.filter(owner_group__in=identity.member_of_groups.all())
|
||||
if self.action != 'list':
|
||||
# retrieve/update/destroy: any item the caller may act on, own or group. See
|
||||
# docs/implementation.md#inventory-queryset-scope-by-action.
|
||||
if identity.user.exists():
|
||||
return InventoryItem.objects.filter(owner=identity.user.get()) | group_items
|
||||
return group_items
|
||||
group_handle = self.request.query_params.get('group')
|
||||
if group_handle:
|
||||
group = resolve_group_by_handle(group_handle)
|
||||
if not group or not group.is_member(identity):
|
||||
try:
|
||||
owner_user, owner_group = resolve_owner_handle(self.kwargs.get('handle', ''))
|
||||
except ValueError:
|
||||
raise ValidationError('invalid owner handle')
|
||||
if owner_group:
|
||||
if not owner_group.is_member(identity):
|
||||
return InventoryItem.objects.none()
|
||||
return InventoryItem.objects.filter(owner_group=group)
|
||||
if identity.user.exists():
|
||||
return InventoryItem.objects.filter(owner=identity.user.get())
|
||||
return InventoryItem.objects.filter(owner_group=owner_group)
|
||||
if owner_user:
|
||||
if owner_user not in identity.friends_or_self():
|
||||
return InventoryItem.objects.none()
|
||||
queryset = InventoryItem.objects.filter(owner=owner_user)
|
||||
if not identity.user.filter(pk=owner_user.pk).exists():
|
||||
queryset = queryset.exclude(availability_policy='private')
|
||||
return queryset
|
||||
return InventoryItem.objects.none()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
group_handle = self.request.data.get('owner_group')
|
||||
try:
|
||||
owner_user, owner_group = resolve_owner_handle(self.kwargs.get('handle', ''))
|
||||
except ValueError:
|
||||
raise ValidationError('invalid owner handle')
|
||||
with transaction.atomic():
|
||||
if group_handle:
|
||||
group = resolve_group_by_handle(group_handle)
|
||||
if not group:
|
||||
raise NotFound('No such group')
|
||||
if not group.is_member(self.request.user):
|
||||
if owner_group:
|
||||
if not owner_group.is_member(self.request.user):
|
||||
raise PermissionDenied('Not a member of this group')
|
||||
serializer.save(owner=None, owner_group=group).clean()
|
||||
serializer.save(owner=None, owner_group=owner_group).clean()
|
||||
elif owner_user:
|
||||
if not self.request.user.user.filter(pk=owner_user.pk).exists():
|
||||
raise PermissionDenied('Cannot create items for another user')
|
||||
serializer.save(owner=owner_user).clean()
|
||||
else:
|
||||
serializer.save(owner=self.request.user.user.get()).clean()
|
||||
raise NotFound('No such owner')
|
||||
|
||||
@staticmethod
|
||||
def _is_authorized(request, instance):
|
||||
|
|
@ -99,13 +83,16 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
|
|||
return instance.owner_group.is_member(request.user)
|
||||
|
||||
def perform_update(self, serializer):
|
||||
# get_queryset() may return a friend's read-only item; reject explicitly rather than silently no-op. See docs/implementation.md#perform-update-rejects-non-owned-items-explicitly.
|
||||
if not self._is_authorized(self.request, serializer.instance):
|
||||
raise PermissionDenied('Not authorized to modify this item')
|
||||
with transaction.atomic():
|
||||
if self._is_authorized(self.request, serializer.instance):
|
||||
serializer.save().clean()
|
||||
serializer.save().clean()
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
if self._is_authorized(self.request, instance):
|
||||
instance.delete()
|
||||
if not self._is_authorized(self.request, instance):
|
||||
raise PermissionDenied('Not authorized to delete this item')
|
||||
instance.delete()
|
||||
|
||||
|
||||
def matches_query(item, query):
|
||||
|
|
@ -130,75 +117,49 @@ def search_inventory_items(request):
|
|||
return Response({'error': 'No query provided.'}, status=400)
|
||||
|
||||
|
||||
@api_view(['GET'])
|
||||
@authentication_classes([SignatureAuthentication])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def get_shared_item(request, handle, id):
|
||||
"""Fetch a single item by its owner's handle and local id, for /i/<handle>/<id> or
|
||||
/inventory/shared/<handle>/<id>. See docs/implementation.md#get-shared-item-looks-up-by-owner."""
|
||||
try:
|
||||
username, domain = split_userhandle_or_throw(handle)
|
||||
except ValueError:
|
||||
return Response(status=400)
|
||||
try:
|
||||
owner = ToolshedUser.objects.get(username=username, domain=domain)
|
||||
except ToolshedUser.DoesNotExist:
|
||||
return Response(status=404)
|
||||
if owner not in request.user.friends_or_self():
|
||||
return Response(status=403)
|
||||
try:
|
||||
item = owner.inventory_items.get(id=id)
|
||||
except InventoryItem.DoesNotExist:
|
||||
return Response(status=404)
|
||||
is_owner = request.user.user.filter(pk=owner.pk).exists()
|
||||
if item.availability_policy == 'private' and not is_owner:
|
||||
return Response(status=403)
|
||||
return Response(InventoryItemSerializer(item).data)
|
||||
|
||||
|
||||
class StorageLocationViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = StorageLocationSerializer
|
||||
authentication_classes = [SignatureAuthentication]
|
||||
permission_classes = [IsAuthenticated]
|
||||
# Detail routes address a location by its owner-scoped id, not the internal row id. See
|
||||
# docs/implementation.md#inventory-detail-routes-use-owner-scoped-ids.
|
||||
# Detail routes address a location by its owner-scoped id, not the internal row id. See docs/implementation.md#owner-handle-scoped-routes.
|
||||
lookup_field = 'id'
|
||||
lookup_url_kwarg = 'pk'
|
||||
|
||||
def get_queryset(self):
|
||||
# Mirrors InventoryItemViewSet.get_queryset() - see its own comments for why the
|
||||
# list/detail scopes differ and why group membership alone (no linked ToolshedUser
|
||||
# required) is enough for the group branch.
|
||||
# See docs/implementation.md#owner-handle-scoped-routes; unlike items, StorageLocation has no availability_policy, so a friend sees all of a user's locations.
|
||||
if type(self.request.user) != KnownIdentity:
|
||||
return StorageLocation.objects.none()
|
||||
identity = self.request.user
|
||||
group_locations = StorageLocation.objects.filter(owner_group__in=identity.member_of_groups.all())
|
||||
if self.action != 'list':
|
||||
if identity.user.exists():
|
||||
return StorageLocation.objects.filter(owner=identity.user.get()) | group_locations
|
||||
return group_locations
|
||||
group_handle = self.request.query_params.get('group')
|
||||
if group_handle:
|
||||
group = resolve_group_by_handle(group_handle)
|
||||
if not group or not group.is_member(identity):
|
||||
try:
|
||||
owner_user, owner_group = resolve_owner_handle(self.kwargs.get('handle', ''))
|
||||
except ValueError:
|
||||
raise ValidationError('invalid owner handle')
|
||||
if owner_group:
|
||||
if not owner_group.is_member(identity):
|
||||
return StorageLocation.objects.none()
|
||||
return StorageLocation.objects.filter(owner_group=group)
|
||||
if identity.user.exists():
|
||||
return StorageLocation.objects.filter(owner=identity.user.get())
|
||||
return StorageLocation.objects.filter(owner_group=owner_group)
|
||||
if owner_user:
|
||||
if owner_user not in identity.friends_or_self():
|
||||
return StorageLocation.objects.none()
|
||||
return StorageLocation.objects.filter(owner=owner_user)
|
||||
return StorageLocation.objects.none()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
group_handle = self.request.data.get('owner_group')
|
||||
try:
|
||||
owner_user, owner_group = resolve_owner_handle(self.kwargs.get('handle', ''))
|
||||
except ValueError:
|
||||
raise ValidationError('invalid owner handle')
|
||||
with transaction.atomic():
|
||||
if group_handle:
|
||||
group = resolve_group_by_handle(group_handle)
|
||||
if not group:
|
||||
raise NotFound('No such group')
|
||||
if not group.is_member(self.request.user):
|
||||
if owner_group:
|
||||
if not owner_group.is_member(self.request.user):
|
||||
raise PermissionDenied('Not a member of this group')
|
||||
serializer.save(owner=None, owner_group=group).clean()
|
||||
serializer.save(owner=None, owner_group=owner_group).clean()
|
||||
elif owner_user:
|
||||
if not self.request.user.user.filter(pk=owner_user.pk).exists():
|
||||
raise PermissionDenied('Cannot create locations for another user')
|
||||
serializer.save(owner=owner_user).clean()
|
||||
else:
|
||||
serializer.save(owner=self.request.user.user.get()).clean()
|
||||
raise NotFound('No such owner')
|
||||
|
||||
@staticmethod
|
||||
def _is_authorized(request, instance):
|
||||
|
|
@ -207,13 +168,16 @@ class StorageLocationViewSet(viewsets.ModelViewSet):
|
|||
return instance.owner_group.is_member(request.user)
|
||||
|
||||
def perform_update(self, serializer):
|
||||
# See docs/implementation.md#perform-update-rejects-non-owned-items-explicitly.
|
||||
if not self._is_authorized(self.request, serializer.instance):
|
||||
raise PermissionDenied('Not authorized to modify this location')
|
||||
with transaction.atomic():
|
||||
if self._is_authorized(self.request, serializer.instance):
|
||||
serializer.save().clean()
|
||||
serializer.save().clean()
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
if self._is_authorized(self.request, instance):
|
||||
instance.delete()
|
||||
if not self._is_authorized(self.request, instance):
|
||||
raise PermissionDenied('Not authorized to delete this location')
|
||||
instance.delete()
|
||||
|
||||
|
||||
class WorkflowInstanceViewSet(viewsets.ModelViewSet):
|
||||
|
|
@ -246,11 +210,11 @@ class WorkflowInstanceViewSet(viewsets.ModelViewSet):
|
|||
file.delete()
|
||||
|
||||
|
||||
router.register(r'inventory_items', InventoryItemViewSet, basename='inventory_items')
|
||||
router.register(r'storage_locations', StorageLocationViewSet, basename='storage_locations')
|
||||
router = routers.SimpleRouter()
|
||||
router.register(r'inventory_items/(?P<handle>[^/]+)', InventoryItemViewSet, basename='inventory_items')
|
||||
router.register(r'storage_locations/(?P<handle>[^/]+)', StorageLocationViewSet, basename='storage_locations')
|
||||
router.register(r'workflows', WorkflowInstanceViewSet, basename='workflows')
|
||||
|
||||
urlpatterns = router.urls + [
|
||||
path('search/', search_inventory_items, name='search_inventory_items'),
|
||||
path('inventory_items/<str:handle>/<int:id>/', get_shared_item, name='shared_inventory_item'),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -19,12 +19,12 @@ class CombinedApiTestCase(UserTestMixin, CategoryTestMixin, TagTestMixin, Proper
|
|||
def test_version_anonymous(self):
|
||||
response = anonymous_client.get('/api/version/')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.json(), {'version': settings.TOOLSHED_VERSION})
|
||||
self.assertEqual(response.json(), {'version': settings.TOOLSHED_VERSION, 'commit': settings.GIT_COMMIT})
|
||||
|
||||
def test_version_authenticated(self):
|
||||
response = client.get('/api/version/', self.f['local_user1'])
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.json(), {'version': settings.TOOLSHED_VERSION})
|
||||
self.assertEqual(response.json(), {'version': settings.TOOLSHED_VERSION, 'commit': settings.GIT_COMMIT})
|
||||
|
||||
def test_domains_anonymous(self):
|
||||
response = anonymous_client.get('/api/domains/')
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
from django.test import Client
|
||||
|
||||
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
|
||||
from toolshed.models import InventoryItem, StorageLocation
|
||||
from toolshed.tests import InventoryTestMixin, LocationTestMixin
|
||||
|
||||
client = SignatureAuthClient()
|
||||
|
||||
|
|
@ -46,3 +48,95 @@ class IdMapTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
def test_idmap_unauthenticated(self):
|
||||
reply = Client().get('/api/idmap/')
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
|
||||
|
||||
class ResolveShortIdApiTestCase(UserTestMixin, InventoryTestMixin, GroupTestMixin, LocationTestMixin,
|
||||
ToolshedTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.prepare_users()
|
||||
self.prepare_categories()
|
||||
self.prepare_tags()
|
||||
self.prepare_properties()
|
||||
self.prepare_inventory()
|
||||
self.prepare_groups()
|
||||
self.prepare_locations()
|
||||
self.owner_id = self.f['local_user1'].public_identity.pk
|
||||
|
||||
def test_resolve_item_as_owner(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': 'testuser1@example.com', 'id': self.f['item1'].id})
|
||||
|
||||
def test_resolve_item_as_friend(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
|
||||
def test_resolve_item_not_friend(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
|
||||
def test_resolve_item_private_not_owner(self):
|
||||
private_item = InventoryItem.create_for_owner(
|
||||
owner=self.f['local_user1'], owned_quantity=1, name='secret', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{private_item.id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{private_item.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
|
||||
def test_resolve_item_unknown_owner(self):
|
||||
reply = client.get('/api/resolve_short_id/item/999999/1/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_item_unknown_local_id(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/999999/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_unsupported_kind(self):
|
||||
# workflow/category/file/group aren't wired up yet - see
|
||||
# docs/handles-and-shortids.md#domain-qualified-short-id's note on the remaining gap.
|
||||
reply = client.get(f'/api/resolve_short_id/workflow/{self.owner_id}/1/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 400)
|
||||
|
||||
def test_resolve_storage_location_as_owner(self):
|
||||
reply = client.get(f'/api/resolve_short_id/storage_location/{self.owner_id}/{self.f["loc1"].id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': 'testuser1@example.com', 'id': self.f['loc1'].id})
|
||||
|
||||
def test_resolve_storage_location_unknown_local_id(self):
|
||||
reply = client.get(f'/api/resolve_short_id/storage_location/{self.owner_id}/999999/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_group_item_as_member(self):
|
||||
item = InventoryItem.create_for_owner(
|
||||
owner_group=self.f['group1'], owned_quantity=1, name='drill', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/group_item/{self.f["group1"].pk}/{item.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': str(self.f['group1']), 'id': item.id})
|
||||
|
||||
def test_resolve_group_item_non_member_denied(self):
|
||||
item = InventoryItem.create_for_owner(
|
||||
owner_group=self.f['group1'], owned_quantity=1, name='drill', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/group_item/{self.f["group1"].pk}/{item.id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
|
||||
def test_resolve_group_storage_location_as_member(self):
|
||||
location = StorageLocation.create_for_owner(owner_group=self.f['group1'], name='shelf')
|
||||
reply = client.get(
|
||||
f'/api/resolve_short_id/group_storage_location/{self.f["group1"].pk}/{location.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': str(self.f['group1']), 'id': location.id})
|
||||
|
||||
def test_resolve_unknown_group(self):
|
||||
reply = client.get('/api/resolve_short_id/group_item/999999/1/', self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
|
|
|||
|
|
@ -260,98 +260,6 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
|
|||
self.assertEqual(reply.status_code, 400)
|
||||
|
||||
|
||||
class ResolveShortIdApiTestCase(UserTestMixin, InventoryTestMixin, GroupTestMixin, LocationTestMixin,
|
||||
ToolshedTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.prepare_users()
|
||||
self.prepare_categories()
|
||||
self.prepare_tags()
|
||||
self.prepare_properties()
|
||||
self.prepare_inventory()
|
||||
self.prepare_groups()
|
||||
self.prepare_locations()
|
||||
self.owner_id = self.f['local_user1'].public_identity.pk
|
||||
|
||||
def test_resolve_item_as_owner(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': 'testuser1@example.com', 'id': self.f['item1'].id})
|
||||
|
||||
def test_resolve_item_as_friend(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
|
||||
def test_resolve_item_not_friend(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{self.f["item1"].id}/',
|
||||
self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
|
||||
def test_resolve_item_private_not_owner(self):
|
||||
private_item = InventoryItem.create_for_owner(
|
||||
owner=self.f['local_user1'], owned_quantity=1, name='secret', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{private_item.id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/{private_item.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
|
||||
def test_resolve_item_unknown_owner(self):
|
||||
reply = client.get('/api/resolve_short_id/item/999999/1/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_item_unknown_local_id(self):
|
||||
reply = client.get(f'/api/resolve_short_id/item/{self.owner_id}/999999/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_unsupported_kind(self):
|
||||
# workflow/category/file/group aren't wired up yet - see
|
||||
# docs/handles-and-shortids.md#domain-qualified-short-id's note on the remaining gap.
|
||||
reply = client.get(f'/api/resolve_short_id/workflow/{self.owner_id}/1/', self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 400)
|
||||
|
||||
def test_resolve_storage_location_as_owner(self):
|
||||
reply = client.get(f'/api/resolve_short_id/storage_location/{self.owner_id}/{self.f["loc1"].id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': 'testuser1@example.com', 'id': self.f['loc1'].id})
|
||||
|
||||
def test_resolve_storage_location_unknown_local_id(self):
|
||||
reply = client.get(f'/api/resolve_short_id/storage_location/{self.owner_id}/999999/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_resolve_group_item_as_member(self):
|
||||
item = InventoryItem.create_for_owner(
|
||||
owner_group=self.f['group1'], owned_quantity=1, name='drill', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/group_item/{self.f["group1"].pk}/{item.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': str(self.f['group1']), 'id': item.id})
|
||||
|
||||
def test_resolve_group_item_non_member_denied(self):
|
||||
item = InventoryItem.create_for_owner(
|
||||
owner_group=self.f['group1'], owned_quantity=1, name='drill', availability_policy='private')
|
||||
reply = client.get(f'/api/resolve_short_id/group_item/{self.f["group1"].pk}/{item.id}/',
|
||||
self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
|
||||
def test_resolve_group_storage_location_as_member(self):
|
||||
location = StorageLocation.create_for_owner(owner_group=self.f['group1'], name='shelf')
|
||||
reply = client.get(
|
||||
f'/api/resolve_short_id/group_storage_location/{self.f["group1"].pk}/{location.id}/',
|
||||
self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json(), {'handle': str(self.f['group1']), 'id': location.id})
|
||||
|
||||
def test_resolve_unknown_group(self):
|
||||
reply = client.get('/api/resolve_short_id/group_item/999999/1/', self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
|
||||
class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, InventoryTestMixin, ToolshedTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue