stash
This commit is contained in:
parent
23185e1721
commit
9f3dc7c993
8 changed files with 335 additions and 103 deletions
|
|
@ -40,18 +40,11 @@ class GroupDetail(APIView, ViewSetMixin):
|
|||
authentication_classes = [SignatureAuthentication]
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def get(self, request, pk, format=None): # /api/groups/<pk>/
|
||||
group = get_object_or_404(Group, pk=pk)
|
||||
if not group.is_member(request.user):
|
||||
return Response(status=status.HTTP_404_NOT_FOUND)
|
||||
return Response(GroupSerializer(group).data)
|
||||
|
||||
|
||||
class GroupDetailByHandle(APIView, ViewSetMixin):
|
||||
authentication_classes = [SignatureAuthentication]
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def get(self, request, name, domain, format=None): # /api/groups/handle/<name>/<domain>/
|
||||
def get(self, request, handle, format=None): # /api/groups/<name@domain>/
|
||||
try:
|
||||
name, domain = split_userhandle_or_throw(handle)
|
||||
except ValueError:
|
||||
return Response(status=status.HTTP_400_BAD_REQUEST, data={'status': 'invalid group handle'})
|
||||
group = get_object_or_404(Group, name=name, domain=domain)
|
||||
if not group.is_member(request.user):
|
||||
return Response(status=status.HTTP_404_NOT_FOUND)
|
||||
|
|
@ -61,8 +54,12 @@ class GroupDetailByHandle(APIView, ViewSetMixin):
|
|||
@api_view(['DELETE'])
|
||||
@authentication_classes([SignatureAuthentication])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def removeGroupMember(request, pk, identity_id, format=None): # /api/groups/<pk>/members/<identity_id>/
|
||||
group = get_object_or_404(Group, pk=pk)
|
||||
def removeGroupMember(request, handle, identity_id, format=None): # /api/groups/<name@domain>/members/<identity_id>/
|
||||
try:
|
||||
name, domain = split_userhandle_or_throw(handle)
|
||||
except ValueError:
|
||||
return Response(status=status.HTTP_400_BAD_REQUEST, data={'status': 'invalid group handle'})
|
||||
group = get_object_or_404(Group, name=name, domain=domain)
|
||||
if not group.is_member(request.user):
|
||||
return Response(status=status.HTTP_404_NOT_FOUND)
|
||||
member = get_object_or_404(group.members, pk=identity_id)
|
||||
|
|
@ -76,8 +73,12 @@ def removeGroupMember(request, pk, identity_id, format=None): # /api/groups/<pk
|
|||
@api_view(['POST'])
|
||||
@authentication_classes([SignatureAuthentication])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def createGroupInvite(request, pk, format=None): # /api/groups/<pk>/invites/
|
||||
group = get_object_or_404(Group, pk=pk)
|
||||
def createGroupInvite(request, handle, format=None): # /api/groups/<name@domain>/invites/
|
||||
try:
|
||||
name, domain = split_userhandle_or_throw(handle)
|
||||
except ValueError:
|
||||
return Response(status=status.HTTP_400_BAD_REQUEST, data={'status': 'invalid group handle'})
|
||||
group = get_object_or_404(Group, name=name, domain=domain)
|
||||
if not group.is_member(request.user):
|
||||
return Response(status=status.HTTP_404_NOT_FOUND)
|
||||
invitee = request.data.get('invitee')
|
||||
|
|
@ -197,10 +198,9 @@ def acceptGroupInvite(request, format=None): # /api/group_invites/accept/ - lan
|
|||
|
||||
urlpatterns = [
|
||||
path('groups/', Groups.as_view(), name='groups'),
|
||||
path('groups/<int:pk>/', GroupDetail.as_view(), name='group_detail'),
|
||||
path('groups/handle/<str:name>/<str:domain>/', GroupDetailByHandle.as_view(), name='group_detail_by_handle'),
|
||||
path('groups/<int:pk>/members/<int:identity_id>/', removeGroupMember, name='remove_group_member'),
|
||||
path('groups/<int:pk>/invites/', createGroupInvite, name='create_group_invite'),
|
||||
path('groups/<str:handle>/', GroupDetail.as_view(), name='group_detail'),
|
||||
path('groups/<str:handle>/members/<int:identity_id>/', removeGroupMember, name='remove_group_member'),
|
||||
path('groups/<str:handle>/invites/', createGroupInvite, name='create_group_invite'),
|
||||
path('groupinvites/', GroupInvitesIncoming.as_view(), name='group_invites_incoming'),
|
||||
path('groupinvites/<int:pk>/', declineGroupInvite, name='decline_group_invite'),
|
||||
path('groupinvites/<int:pk>/accept/', recordGroupMembership, name='record_group_membership'),
|
||||
|
|
|
|||
|
|
@ -15,6 +15,21 @@ from toolshed.serializers import InventoryItemSerializer, StorageLocationSeriali
|
|||
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:
|
||||
user = identity.user.get()
|
||||
|
|
@ -54,13 +69,10 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
|
|||
if identity.user.exists():
|
||||
return InventoryItem.objects.filter(owner=identity.user.get()) | group_items
|
||||
return group_items
|
||||
group_id = self.request.query_params.get('group')
|
||||
if group_id:
|
||||
try:
|
||||
group = Group.objects.get(pk=group_id)
|
||||
except Group.DoesNotExist:
|
||||
return InventoryItem.objects.none()
|
||||
if not group.is_member(identity):
|
||||
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):
|
||||
return InventoryItem.objects.none()
|
||||
return InventoryItem.objects.filter(owner_group=group)
|
||||
if identity.user.exists():
|
||||
|
|
@ -68,12 +80,11 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
|
|||
return InventoryItem.objects.none()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
group_id = self.request.data.get('owner_group')
|
||||
group_handle = self.request.data.get('owner_group')
|
||||
with transaction.atomic():
|
||||
if group_id:
|
||||
try:
|
||||
group = Group.objects.get(pk=group_id)
|
||||
except Group.DoesNotExist:
|
||||
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):
|
||||
raise PermissionDenied('Not a member of this group')
|
||||
|
|
@ -155,21 +166,53 @@ class StorageLocationViewSet(viewsets.ModelViewSet):
|
|||
lookup_url_kwarg = 'pk'
|
||||
|
||||
def get_queryset(self):
|
||||
if type(self.request.user) == KnownIdentity and self.request.user.user.exists():
|
||||
return StorageLocation.objects.filter(owner=self.request.user.user.get())
|
||||
# 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.
|
||||
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):
|
||||
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.none()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
group_handle = self.request.data.get('owner_group')
|
||||
with transaction.atomic():
|
||||
serializer.save(owner=self.request.user.user.get())
|
||||
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):
|
||||
raise PermissionDenied('Not a member of this group')
|
||||
serializer.save(owner=None, owner_group=group).clean()
|
||||
else:
|
||||
serializer.save(owner=self.request.user.user.get()).clean()
|
||||
|
||||
@staticmethod
|
||||
def _is_authorized(request, instance):
|
||||
if instance.owner_id:
|
||||
return request.user.user.filter(pk=instance.owner_id).exists()
|
||||
return instance.owner_group.is_member(request.user)
|
||||
|
||||
def perform_update(self, serializer):
|
||||
with transaction.atomic():
|
||||
if serializer.instance.owner == self.request.user.user.get():
|
||||
serializer.save()
|
||||
if self._is_authorized(self.request, serializer.instance):
|
||||
serializer.save().clean()
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
if instance.owner == self.request.user.user.get():
|
||||
if self._is_authorized(self.request, instance):
|
||||
instance.delete()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,63 @@
|
|||
# Generated by Django 4.2.2 on 2026-08-26 18:55
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('authentication', '0006_groupmembership'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
('toolshed', '0022_storagelocation_id_not_null_and_unique'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RemoveConstraint(
|
||||
model_name='storagelocation',
|
||||
name='storagelocation_unique_owner_scoped_id',
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='ownerstoragelocationsequence',
|
||||
name='owner_group',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='+', to='authentication.group'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='storagelocation',
|
||||
name='owner_group',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='storage_locations', to='authentication.group'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='owneritemsequence',
|
||||
name='id',
|
||||
field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='ownerstoragelocationsequence',
|
||||
name='id',
|
||||
field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='ownerstoragelocationsequence',
|
||||
name='owner',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='+', to=settings.AUTH_USER_MODEL),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='storagelocation',
|
||||
name='owner',
|
||||
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='storage_locations', to=settings.AUTH_USER_MODEL),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name='ownerstoragelocationsequence',
|
||||
constraint=models.UniqueConstraint(condition=models.Q(('owner__isnull', False)), fields=('owner',), name='ownerstoragelocationsequence_unique_owner'),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name='ownerstoragelocationsequence',
|
||||
constraint=models.UniqueConstraint(condition=models.Q(('owner_group__isnull', False)), fields=('owner_group',), name='ownerstoragelocationsequence_unique_owner_group'),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name='storagelocation',
|
||||
constraint=models.UniqueConstraint(fields=('owner', 'owner_group', 'id'), name='storagelocation_unique_owner_scoped_id'),
|
||||
),
|
||||
]
|
||||
|
|
@ -170,15 +170,25 @@ class ItemTag(models.Model):
|
|||
|
||||
|
||||
class OwnerStorageLocationSequence(models.Model):
|
||||
"""Tracks the last StorageLocation id handed out per owner for sequential, gapless allocation
|
||||
(see StorageLocation.create_for_owner)."""
|
||||
owner = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='+', unique=True)
|
||||
"""Tracks the last StorageLocation id handed out per owner/owner_group scope for sequential,
|
||||
gapless allocation (see StorageLocation.create_for_owner); exactly one of owner/owner_group is
|
||||
set, mirroring OwnerItemSequence/InventoryItem's own split."""
|
||||
owner = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, null=True, blank=True, related_name='+')
|
||||
owner_group = models.ForeignKey(Group, on_delete=models.CASCADE, null=True, blank=True, related_name='+')
|
||||
last_id = models.PositiveIntegerField(default=0)
|
||||
|
||||
class Meta:
|
||||
constraints = [
|
||||
models.UniqueConstraint(fields=['owner'], condition=models.Q(owner__isnull=False),
|
||||
name='ownerstoragelocationsequence_unique_owner'),
|
||||
models.UniqueConstraint(fields=['owner_group'], condition=models.Q(owner_group__isnull=False),
|
||||
name='ownerstoragelocationsequence_unique_owner_group'),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def allocate(cls, *, owner):
|
||||
def allocate(cls, *, owner=None, owner_group=None):
|
||||
with transaction.atomic():
|
||||
seq, _ = cls.objects.select_for_update().get_or_create(owner=owner)
|
||||
seq, _ = cls.objects.select_for_update().get_or_create(owner=owner, owner_group=owner_group)
|
||||
seq.last_id += 1
|
||||
seq.save(update_fields=['last_id'])
|
||||
return seq.last_id
|
||||
|
|
@ -186,7 +196,7 @@ class OwnerStorageLocationSequence(models.Model):
|
|||
|
||||
class StorageLocation(models.Model):
|
||||
internal_id = models.AutoField(primary_key=True)
|
||||
# Externally visible id, sequential/gapless within the owner's own locations (see
|
||||
# Externally visible id, sequential/gapless within the owner/owner_group's own locations (see
|
||||
# OwnerStorageLocationSequence), never internal_id; always allocate via create_for_owner, not
|
||||
# .objects.create().
|
||||
id = models.PositiveIntegerField(editable=False)
|
||||
|
|
@ -195,24 +205,32 @@ class StorageLocation(models.Model):
|
|||
category = models.ForeignKey(Category, on_delete=models.CASCADE, null=True, blank=True,
|
||||
related_name='storage_locations')
|
||||
parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='children')
|
||||
owner = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='storage_locations')
|
||||
owner = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, null=True, blank=True,
|
||||
related_name='storage_locations')
|
||||
owner_group = models.ForeignKey(Group, on_delete=models.CASCADE, null=True, blank=True,
|
||||
related_name='storage_locations')
|
||||
|
||||
class Meta:
|
||||
constraints = [
|
||||
models.UniqueConstraint(fields=['owner', 'id'], name='storagelocation_unique_owner_scoped_id'),
|
||||
models.UniqueConstraint(fields=['owner', 'owner_group', 'id'],
|
||||
name='storagelocation_unique_owner_scoped_id'),
|
||||
]
|
||||
|
||||
def __str__(self):
|
||||
parent = str(self.parent) + "/" if self.parent else ""
|
||||
return parent + self.name
|
||||
|
||||
def clean(self):
|
||||
if (self.owner is None) == (self.owner_group is None):
|
||||
raise ValidationError("Exactly one of owner or owner_group must be set")
|
||||
|
||||
@classmethod
|
||||
def create_for_owner(cls, *, owner, **kwargs):
|
||||
def create_for_owner(cls, *, owner=None, owner_group=None, **kwargs):
|
||||
"""The only supported way to create a StorageLocation: atomically allocates the next id
|
||||
for this owner's scope."""
|
||||
for this owner/owner_group's scope."""
|
||||
with transaction.atomic():
|
||||
next_id = OwnerStorageLocationSequence.allocate(owner=owner)
|
||||
return cls.objects.create(owner=owner, id=next_id, **kwargs)
|
||||
next_id = OwnerStorageLocationSequence.allocate(owner=owner, owner_group=owner_group)
|
||||
return cls.objects.create(owner=owner, owner_group=owner_group, id=next_id, **kwargs)
|
||||
|
||||
|
||||
class WorkflowInstance(models.Model):
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from django.db.models import Q
|
||||
from rest_framework import serializers
|
||||
from authentication.models import KnownIdentity, ToolshedUser, FriendRequestIncoming, Group, GroupInviteIncoming, \
|
||||
GroupMembership
|
||||
|
|
@ -177,9 +178,13 @@ class OwnerScopedPrimaryKeyRelatedField(serializers.PrimaryKeyRelatedField):
|
|||
def get_queryset(self):
|
||||
queryset = super().get_queryset()
|
||||
request = self.context.get('request')
|
||||
if request is not None and type(request.user) == KnownIdentity and request.user.user.exists():
|
||||
return queryset.filter(owner=request.user.user.get())
|
||||
if request is None or type(request.user) != KnownIdentity:
|
||||
return queryset.none()
|
||||
identity = request.user
|
||||
owner_filter = Q(owner_group__in=identity.member_of_groups.all())
|
||||
if identity.user.exists():
|
||||
owner_filter |= Q(owner=identity.user.get())
|
||||
return queryset.filter(owner_filter)
|
||||
|
||||
def to_internal_value(self, data):
|
||||
queryset = self.get_queryset()
|
||||
|
|
@ -198,6 +203,7 @@ class OwnerScopedPrimaryKeyRelatedField(serializers.PrimaryKeyRelatedField):
|
|||
|
||||
class StorageLocationSerializer(serializers.ModelSerializer):
|
||||
owner = OwnerSerializer(read_only=True)
|
||||
owner_group = GroupOwnerSerializer(read_only=True)
|
||||
category = serializers.CharField(required=False, allow_null=True, allow_blank=True)
|
||||
parent = OwnerScopedPrimaryKeyRelatedField(queryset=StorageLocation.objects.all(), required=False,
|
||||
allow_null=True)
|
||||
|
|
@ -205,7 +211,7 @@ class StorageLocationSerializer(serializers.ModelSerializer):
|
|||
|
||||
class Meta:
|
||||
model = StorageLocation
|
||||
fields = ['id', 'name', 'description', 'path', 'category', 'owner', 'parent']
|
||||
fields = ['id', 'name', 'description', 'path', 'category', 'owner', 'owner_group', 'parent']
|
||||
read_only_fields = ['id', 'path']
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -53,56 +53,47 @@ class GroupApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
|
||||
def test_group_detail_member(self):
|
||||
self.prepare_groups()
|
||||
reply = client.get('/api/groups/{}/'.format(self.f['group1'].id), self.f['local_user1'])
|
||||
group = self.f['group1']
|
||||
reply = client.get('/api/groups/{}@{}/'.format(group.name, group.domain), self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json()['handle'], str(self.f['group1']))
|
||||
self.assertEqual(reply.json()['handle'], str(group))
|
||||
self.assertEqual(len(reply.json()['members']), 1)
|
||||
|
||||
def test_group_detail_non_member(self):
|
||||
self.prepare_groups()
|
||||
reply = client.get('/api/groups/{}/'.format(self.f['group1'].id), self.f['local_user2'])
|
||||
group = self.f['group1']
|
||||
reply = client.get('/api/groups/{}@{}/'.format(group.name, group.domain), self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_group_detail_by_handle_member(self):
|
||||
self.prepare_groups()
|
||||
group = self.f['group1']
|
||||
reply = client.get('/api/groups/handle/{}/{}/'.format(group.name, group.domain), self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(reply.json()['id'], group.id)
|
||||
self.assertEqual(reply.json()['handle'], str(group))
|
||||
|
||||
def test_group_detail_by_handle_non_member(self):
|
||||
self.prepare_groups()
|
||||
group = self.f['group1']
|
||||
reply = client.get('/api/groups/handle/{}/{}/'.format(group.name, group.domain), self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_group_detail_by_handle_no_such_group(self):
|
||||
reply = client.get('/api/groups/handle/nonexistent/example.com/', self.f['local_user1'])
|
||||
def test_group_detail_no_such_group(self):
|
||||
reply = client.get('/api/groups/nonexistent@example.com/', self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_remove_member(self):
|
||||
self.prepare_groups()
|
||||
self.f['group1'].members.add(self.f['local_user2'].public_identity)
|
||||
reply = client.delete('/api/groups/{}/members/{}/'.format(
|
||||
self.f['group1'].id, self.f['local_user2'].public_identity.id), self.f['local_user1'])
|
||||
group = self.f['group1']
|
||||
group.members.add(self.f['local_user2'].public_identity)
|
||||
reply = client.delete('/api/groups/{}@{}/members/{}/'.format(
|
||||
group.name, group.domain, self.f['local_user2'].public_identity.id), self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 204)
|
||||
self.assertEqual(self.f['group1'].members.count(), 1)
|
||||
self.assertEqual(group.members.count(), 1)
|
||||
|
||||
def test_remove_last_member_blocked(self):
|
||||
self.prepare_groups()
|
||||
reply = client.delete('/api/groups/{}/members/{}/'.format(
|
||||
self.f['group1'].id, self.f['local_user1'].public_identity.id), self.f['local_user1'])
|
||||
group = self.f['group1']
|
||||
reply = client.delete('/api/groups/{}@{}/members/{}/'.format(
|
||||
group.name, group.domain, self.f['local_user1'].public_identity.id), self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 400)
|
||||
self.assertEqual(self.f['group1'].members.count(), 1)
|
||||
self.assertEqual(group.members.count(), 1)
|
||||
|
||||
def test_remove_member_non_member_denied(self):
|
||||
self.prepare_groups()
|
||||
self.f['group1'].members.add(self.f['local_user2'].public_identity)
|
||||
reply = client.delete('/api/groups/{}/members/{}/'.format(
|
||||
self.f['group1'].id, self.f['local_user1'].public_identity.id), self.f['ext_user1'])
|
||||
group = self.f['group1']
|
||||
group.members.add(self.f['local_user2'].public_identity)
|
||||
reply = client.delete('/api/groups/{}@{}/members/{}/'.format(
|
||||
group.name, group.domain, self.f['local_user1'].public_identity.id), self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
self.assertEqual(self.f['group1'].members.count(), 2)
|
||||
self.assertEqual(group.members.count(), 2)
|
||||
|
||||
|
||||
class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
||||
|
|
@ -116,7 +107,8 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
inviter = self.f['local_user1']
|
||||
invitee = self.f['local_user2']
|
||||
|
||||
reply1 = client.post('/api/groups/{}/invites/'.format(group.id), inviter, {'invitee': str(invitee)})
|
||||
reply1 = client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
||||
{'invitee': str(invitee)})
|
||||
self.assertEqual(reply1.status_code, 201)
|
||||
secret = reply1.json()['secret']
|
||||
self.assertEqual(GroupInvite.objects.count(), 1)
|
||||
|
|
@ -152,7 +144,7 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
|
||||
def test_invite_non_member_denied(self):
|
||||
group = self.f['group1']
|
||||
reply = client.post('/api/groups/{}/invites/'.format(group.id), self.f['local_user2'],
|
||||
reply = client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), self.f['local_user2'],
|
||||
{'invitee': str(self.f['ext_user1'])})
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
self.assertEqual(GroupInvite.objects.count(), 0)
|
||||
|
|
@ -160,7 +152,7 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
def test_invite_already_member(self):
|
||||
group = self.f['group1']
|
||||
group.members.add(self.f['local_user2'].public_identity)
|
||||
reply = client.post('/api/groups/{}/invites/'.format(group.id), self.f['local_user1'],
|
||||
reply = client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), self.f['local_user1'],
|
||||
{'invitee': str(self.f['local_user2'])})
|
||||
self.assertEqual(reply.status_code, 208)
|
||||
|
||||
|
|
@ -177,7 +169,7 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
group = self.f['group1']
|
||||
inviter = self.f['local_user1']
|
||||
invitee = self.f['local_user2']
|
||||
client.post('/api/groups/{}/invites/'.format(group.id), inviter, {'invitee': str(invitee)})
|
||||
client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), inviter, {'invitee': str(invitee)})
|
||||
reply = client.post('/api/group_invites/accept/', invitee, {
|
||||
'group': str(group), 'invitee': str(invitee), 'invitee_key': invitee.public_key(),
|
||||
'secret': 'not-the-right-secret'
|
||||
|
|
@ -189,7 +181,8 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
group = self.f['group1']
|
||||
inviter = self.f['local_user1']
|
||||
invitee = DummyExternalUser('newmember', 'remote.example', known=False)
|
||||
reply1 = client.post('/api/groups/{}/invites/'.format(group.id), inviter, {'invitee': str(invitee)})
|
||||
reply1 = client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
||||
{'invitee': str(invitee)})
|
||||
secret = reply1.json()['secret']
|
||||
|
||||
reply = client.post('/api/group_invites/accept/', invitee, {
|
||||
|
|
@ -206,7 +199,8 @@ class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|||
group = self.f['group1']
|
||||
inviter = self.f['local_user1']
|
||||
invitee = self.f['local_user2']
|
||||
reply1 = client.post('/api/groups/{}/invites/'.format(group.id), inviter, {'invitee': str(invitee)})
|
||||
reply1 = client.post('/api/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
||||
{'invitee': str(invitee)})
|
||||
secret = reply1.json()['secret']
|
||||
bad_signature_client = SignatureAuthClient(bad_signature=True)
|
||||
reply = bad_signature_client.post('/api/group_invites/accept/', invitee, {
|
||||
|
|
|
|||
|
|
@ -353,7 +353,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
def create_group_item(self, name='drill'):
|
||||
return client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': name, 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'owner_group': self.f['group1'].id,
|
||||
'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
|
||||
def test_create_group_owned_item(self):
|
||||
|
|
@ -368,7 +368,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
def test_create_group_owned_item_non_member_denied(self):
|
||||
reply = client.post('/api/inventory_items/', self.f['ext_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'owner_group': self.f['group1'].id,
|
||||
'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
self.assertEqual(InventoryItem.objects.count(), 0)
|
||||
|
|
@ -413,14 +413,14 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
|
||||
def test_group_items_listed_by_group_query_param(self):
|
||||
self.create_group_item()
|
||||
reply = client.get('/api/inventory_items/?group={}'.format(self.f['group1'].id), self.f['local_user2'])
|
||||
reply = client.get('/api/inventory_items/?group={}'.format(str(self.f['group1'])[1:]), self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 1)
|
||||
self.assertEqual(reply.json()[0]['name'], 'drill')
|
||||
|
||||
def test_group_items_not_listed_for_non_member_query_param(self):
|
||||
self.create_group_item()
|
||||
reply = client.get('/api/inventory_items/?group={}'.format(self.f['group1'].id), self.f['ext_user1'])
|
||||
reply = client.get('/api/inventory_items/?group={}'.format(str(self.f['group1'])[1:]), self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 0)
|
||||
|
||||
|
|
@ -435,7 +435,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
'owned_quantity': 3,
|
||||
'tags': ['tag1', 'tag2'],
|
||||
'properties': [{'name': 'prop1', 'value': 'value1'}, {'name': 'prop2', 'value': 'value2'}],
|
||||
'owner_group': self.f['group1'].id,
|
||||
'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
item = InventoryItem.objects.get(name='drill')
|
||||
|
|
@ -451,20 +451,21 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
def test_create_group_owned_item_empty_fails(self):
|
||||
# Parity with InventoryApiTestCase.test_post_new_item_empty: clean()'s name-or-files validation still applies.
|
||||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'availability_policy': 'private', 'owned_quantity': 1, 'owner_group': self.f['group1'].id,
|
||||
'availability_policy': 'private', 'owned_quantity': 1, 'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 400)
|
||||
self.assertEqual(InventoryItem.objects.count(), 0)
|
||||
|
||||
def test_create_group_owned_item_nonexistent_group(self):
|
||||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private', 'owner_group': 999999,
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'owner_group': 'nonexistent@example.com',
|
||||
})
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
self.assertEqual(InventoryItem.objects.count(), 0)
|
||||
|
||||
def test_group_items_listed_for_nonexistent_group_query_param(self):
|
||||
reply = client.get('/api/inventory_items/?group=999999', self.f['local_user1'])
|
||||
reply = client.get('/api/inventory_items/?group=nonexistent@example.com', self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 0)
|
||||
|
||||
|
|
@ -495,7 +496,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'category': 'cat1', 'tags': ['tag1'],
|
||||
'owner_group': self.f['group1'].id,
|
||||
'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
item_id = reply.json()['id']
|
||||
reply = client.patch('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'], {
|
||||
|
|
@ -509,7 +510,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
def test_group_item_storage_location(self):
|
||||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'storage_location': self.f['loc1'].id, 'owner_group': self.f['group1'].id,
|
||||
'storage_location': self.f['loc1'].id, 'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
item = InventoryItem.objects.get(name='drill')
|
||||
|
|
@ -519,7 +520,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
# Parity with TestInventoryItemWithFileApiTestCase.test_post_item_with_file_id.
|
||||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'files': [self.f['test_file1'].id], 'owner_group': self.f['group1'].id,
|
||||
'files': [self.f['test_file1'].id], 'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
item = InventoryItem.objects.get(name='drill')
|
||||
|
|
@ -529,7 +530,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
|
|||
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
|
||||
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
|
||||
'files': [{'data': self.f['encoded_content3'], 'mime_type': 'text/plain'}],
|
||||
'owner_group': self.f['group1'].id,
|
||||
'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
item = InventoryItem.objects.get(name='drill')
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
from authentication.tests import SignatureAuthClient, UserTestMixin, ToolshedTestCase
|
||||
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
|
||||
from files.tests import FilesTestMixin
|
||||
from toolshed.models import InventoryItem, Category, StorageLocation
|
||||
from toolshed.tests import InventoryTestMixin, LocationTestMixin
|
||||
|
|
@ -129,3 +129,110 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
|
|||
self.assertEqual(reply.status_code, 404)
|
||||
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc1'].id).count(), 1)
|
||||
|
||||
|
||||
class GroupOwnedLocationApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.prepare_users()
|
||||
self.prepare_groups()
|
||||
self.f['group1'].members.add(self.f['local_user2'].public_identity)
|
||||
|
||||
def create_group_location(self, name='shelf'):
|
||||
return client.post('/api/storage_locations/', self.f['local_user1'], {
|
||||
'name': name, 'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
|
||||
def test_create_group_owned_location(self):
|
||||
reply = self.create_group_location()
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
location = StorageLocation.objects.get(name='shelf')
|
||||
self.assertIsNone(location.owner)
|
||||
self.assertEqual(location.owner_group, self.f['group1'])
|
||||
self.assertEqual(reply.json()['owner_group'], str(self.f['group1']))
|
||||
self.assertIsNone(reply.json()['owner'])
|
||||
|
||||
def test_create_group_owned_location_non_member_denied(self):
|
||||
reply = client.post('/api/storage_locations/', self.f['ext_user1'], {
|
||||
'name': 'shelf', 'owner_group': str(self.f['group1'])[1:],
|
||||
})
|
||||
self.assertEqual(reply.status_code, 403)
|
||||
self.assertEqual(StorageLocation.objects.count(), 0)
|
||||
|
||||
def test_other_member_can_edit(self):
|
||||
location_id = self.create_group_location().json()['id']
|
||||
reply = client.patch('/api/storage_locations/{}/'.format(location_id), self.f['local_user2'], {
|
||||
'name': 'shelf-renamed'
|
||||
})
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(StorageLocation.objects.get(id=location_id).name, 'shelf-renamed')
|
||||
|
||||
def test_other_member_can_delete(self):
|
||||
location_id = self.create_group_location().json()['id']
|
||||
reply = client.delete('/api/storage_locations/{}/'.format(location_id), self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 204)
|
||||
self.assertEqual(StorageLocation.objects.filter(id=location_id).count(), 0)
|
||||
|
||||
def test_remote_member_without_local_account_can_edit(self):
|
||||
# A remote member (KnownIdentity, no ToolshedUser row) must still act on group-owned
|
||||
# locations - not unauthorized just because .user.exists() is False.
|
||||
self.f['group1'].members.add(self.f['ext_user1'].public_identity)
|
||||
location_id = self.create_group_location().json()['id']
|
||||
reply = client.get('/api/storage_locations/{}/'.format(location_id), self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
reply = client.patch('/api/storage_locations/{}/'.format(location_id), self.f['ext_user1'], {
|
||||
'name': 'shelf-renamed-by-remote-member'
|
||||
})
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(StorageLocation.objects.get(id=location_id).name, 'shelf-renamed-by-remote-member')
|
||||
|
||||
def test_non_member_cannot_see_or_edit(self):
|
||||
location_id = self.create_group_location().json()['id']
|
||||
reply = client.get('/api/storage_locations/{}/'.format(location_id), self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 404)
|
||||
|
||||
def test_group_locations_excluded_from_personal_list(self):
|
||||
self.create_group_location()
|
||||
reply = client.get('/api/storage_locations/', self.f['local_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 0)
|
||||
|
||||
def test_group_locations_listed_by_group_query_param(self):
|
||||
self.create_group_location()
|
||||
reply = client.get('/api/storage_locations/?group={}'.format(str(self.f['group1'])[1:]), self.f['local_user2'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 1)
|
||||
self.assertEqual(reply.json()[0]['name'], 'shelf')
|
||||
|
||||
def test_group_locations_not_listed_for_non_member_query_param(self):
|
||||
self.create_group_location()
|
||||
reply = client.get('/api/storage_locations/?group={}'.format(str(self.f['group1'])[1:]), self.f['ext_user1'])
|
||||
self.assertEqual(reply.status_code, 200)
|
||||
self.assertEqual(len(reply.json()), 0)
|
||||
|
||||
def test_group_location_can_be_parent_for_group_member(self):
|
||||
parent_id = self.create_group_location('shelf').json()['id']
|
||||
reply = client.post('/api/storage_locations/', self.f['local_user2'], {
|
||||
'name': 'bin', 'owner_group': str(self.f['group1'])[1:], 'parent': parent_id,
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
self.assertEqual(reply.json()['path'], 'shelf/bin')
|
||||
|
||||
def test_own_personal_location_can_be_parent_of_group_location(self):
|
||||
# OwnerScopedPrimaryKeyRelatedField (see serializers.py) scopes `parent` to every
|
||||
# location the requester can act on - own personal ones plus any group's - with no
|
||||
# further check that it matches the new location's own owner_group. Same behavior
|
||||
# InventoryItem.storage_location already had before group ownership existed here; not
|
||||
# something this feature narrows.
|
||||
personal = StorageLocation.create_for_owner(name='mine', owner=self.f['local_user1'])
|
||||
reply = client.post('/api/storage_locations/', self.f['local_user1'], {
|
||||
'name': 'bin', 'owner_group': str(self.f['group1'])[1:], 'parent': personal.id,
|
||||
})
|
||||
self.assertEqual(reply.status_code, 201)
|
||||
|
||||
def test_group_location_not_valid_parent_for_non_member(self):
|
||||
parent_id = self.create_group_location('shelf').json()['id']
|
||||
reply = client.post('/api/storage_locations/', self.f['ext_user1'], {
|
||||
'name': 'bin', 'parent': parent_id,
|
||||
})
|
||||
self.assertEqual(reply.status_code, 400)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue