This commit is contained in:
j3d1 2026-08-27 01:34:34 +02:00
parent f46eff65fd
commit d56784eb8d
8 changed files with 335 additions and 93 deletions

View file

@ -25,9 +25,11 @@ THUMBNAIL_SIZES = (32, 64, 256)
def _accessible_files(request):
# Shared by media_urls and thumbnail_urls: a file is visible if the requester is
# friends-or-self with whatever references it (item, profile picture), or it's their own staged photo.
# friends-or-self with whatever references it (item, profile picture), a member of the group
# that owns the item it's attached to, or it's their own staged photo.
return File.objects.filter(
Q(connected_items__owner__in=request.user.friends_or_self()) |
Q(connected_items__owner_group__in=request.user.member_of_groups.all()) |
Q(profile_picture_users__in=request.user.friends_or_self()) |
Q(staged_by_workflows__owner__in=request.user.user.all())
).distinct()

View file

@ -7,7 +7,8 @@ from django.core.files.base import ContentFile
from django.core.files.storage import DefaultStorage, default_storage
from django.db import IntegrityError, transaction
from django.test import Client, override_settings
from authentication.tests import SignatureAuthClient, ToolshedTestCase, UserTestMixin
from authentication.tests import SignatureAuthClient, ToolshedTestCase, UserTestMixin, GroupTestMixin
from toolshed.models import InventoryItem
from toolshed.tests import InventoryTestMixin
from nacl.hash import sha256
from nacl.encoding import HexEncoder
@ -216,6 +217,35 @@ class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, Toolsh
self.assertEqual(reply.status_code, 404)
class GroupOwnedMediaUrlTestCase(FilesTestMixin, UserTestMixin, GroupTestMixin, ToolshedTestCase):
"""_accessible_files() only checked connected_items__owner (personal items) before, never
connected_items__owner_group - a group-owned item's own files were unreachable via /media/ or
/thumbnails/ for every member, including ones who could see and edit the item itself."""
def setUp(self):
super().setUp()
self.prepare_files()
self.prepare_users()
self.prepare_groups()
self.f['group1'].members.add(self.f['local_user2'].public_identity)
self.f['group_item'] = InventoryItem.create_for_owner(
owner_group=self.f['group1'], owned_quantity=1, name='group-drill')
self.f['group_item'].files.add(self.f['test_file1'])
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
def test_group_member_can_view_group_item_file(self):
reply = client.get(
f"/media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}",
self.f['local_user2'])
self.assertEqual(reply.status_code, 200)
def test_non_member_cannot_view_group_item_file(self):
reply = client.get(
f"/media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}",
self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)
class ThumbnailUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, ToolshedTestCase):
def setUp(self):
super().setUp()

View file

@ -1,4 +1,5 @@
from django.test import Client
from authentication.models import Group
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
from files.tests import FilesTestMixin
from toolshed.models import File, InventoryItem
@ -140,7 +141,7 @@ class FileApiTestCase(UserTestMixin, FilesTestMixin, InventoryTestMixin, Toolshe
self.assertEqual(self.f['item1'].files.count(), 2)
def test_get_inventory(self):
reply = client.get('/api/inventory_items/', self.f['local_user1'])
reply = client.get('/api/inventory_items/{}/'.format(self.f['local_user1']), self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 2)
self.assertEqual(reply.json()[0]['name'], 'test1')
@ -213,3 +214,20 @@ class GroupOwnedFileApiTestCase(UserTestMixin, GroupTestMixin, FilesTestMixin, T
self.f['ext_user1'])
self.assertEqual(response.status_code, 404)
self.assertEqual(self.f['group_item'].files.count(), 1)
def test_item_files_when_id_collides_across_two_groups(self):
group2 = Group.objects.create(name='group2', domain=self.f['example_com'].name)
group2.members.add(self.f['local_user2'].public_identity)
group2_item = InventoryItem.create_for_owner(
owner_group=group2, owned_quantity=1, name='group2-drill', availability_policy='private')
self.assertEqual(self.f['group_item'].id, group2_item.id)
response = client.get(f"/api/item_files/{self.f['group_item'].id}/", self.f['local_user2'])
self.assertEqual(response.status_code, 200)
def test_group_item_files_listed_in_all_files(self):
response = client.get('/api/files/', self.f['local_user2'])
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 1)
self.assertEqual(response.json()[0]['id'], self.f['test_file1'].id)

View file

@ -1,7 +1,7 @@
from authentication.models import Group
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
from files.tests import FilesTestMixin
from toolshed.models import InventoryItem, Category
from toolshed.models import InventoryItem, Category, StorageLocation
from toolshed.tests import InventoryTestMixin, CategoryTestMixin, TagTestMixin, PropertyTestMixin, LocationTestMixin
client = SignatureAuthClient()
@ -16,9 +16,10 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.prepare_tags()
self.prepare_properties()
self.prepare_inventory()
self.own_handle = str(self.f['local_user1'])
def test_get_inventory(self):
reply = client.get('/api/inventory_items/', self.f['local_user1'])
reply = client.get('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 2)
self.assertEqual(reply.json()[0]['name'], 'test1')
@ -38,7 +39,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual(reply.json()[1]['availability_policy'], 'friends')
def test_post_new_item(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'availability_policy': 'rent',
'category': 'cat2',
'name': 'test3',
@ -61,7 +62,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p.value for p in item.itemproperty_set.all()], ['value3', 'value4'])
def test_post_new_item2(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'availability_policy': 'share',
'name': 'test3',
'description': 'test',
@ -80,7 +81,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p for p in item.properties.all()], [])
def test_post_new_item_empty(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'availability_policy': 'rent',
'owned_quantity': 1,
'image': '',
@ -89,7 +90,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual(InventoryItem.objects.count(), 2)
def test_post_new_item3(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'availability_policy': 'private',
'name': 'test3',
'description': 'test',
@ -109,7 +110,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p for p in item.properties.all()], [])
def test_put_item(self):
reply = client.put('/api/inventory_items/1/', self.f['local_user1'], {
reply = client.put('/api/inventory_items/{}/1/'.format(self.own_handle), self.f['local_user1'], {
'availability_policy': 'sell',
'name': 'test4',
'description': 'new description',
@ -132,7 +133,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p.value for p in item.itemproperty_set.all()], ['value5', 'value6', 'value7'])
def test_patch_item(self):
reply = client.patch('/api/inventory_items/1/', self.f['local_user1'], {
reply = client.patch('/api/inventory_items/{}/1/'.format(self.own_handle), self.f['local_user1'], {
'description': 'new description2',
'category': 'cat1',
'owned_quantity': 100,
@ -152,7 +153,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p.value for p in item.itemproperty_set.all()], ['value8'])
def test_patch_item2(self):
reply = client.patch('/api/inventory_items/1/', self.f['local_user1'], {
reply = client.patch('/api/inventory_items/{}/1/'.format(self.own_handle), self.f['local_user1'], {
'description': 'new description2',
'category': None,
'owned_quantity': 100,
@ -171,7 +172,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual([p for p in item.properties.all()], [])
def test_delete_item(self):
reply = client.delete('/api/inventory_items/1/', self.f['local_user1'])
reply = client.delete('/api/inventory_items/{}/1/'.format(self.own_handle), self.f['local_user1'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(InventoryItem.objects.count(), 1)
self.assertEqual(InventoryItem.objects.get(id=2).name, 'test2')
@ -228,16 +229,19 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual(reply.json()['name'], 'test1')
def test_get_shared_item_not_friend(self):
# Non-visible resolves the same way a non-member's group access does now (see
# GroupOwnedInventoryApiTestCase.test_non_member_cannot_see_or_edit) - 404, not a
# separate 403 special case for "not a friend".
reply = client.get('/api/inventory_items/testuser1@example.com/' + str(self.f['item1'].id) + '/',
self.f['ext_user1'])
self.assertEqual(reply.status_code, 403)
self.assertEqual(reply.status_code, 404)
def test_get_shared_item_private(self):
private_item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], owned_quantity=1, name='secret', availability_policy='private')
reply = client.get('/api/inventory_items/testuser1@example.com/' + str(private_item.id) + '/',
self.f['local_user2'])
self.assertEqual(reply.status_code, 403)
self.assertEqual(reply.status_code, 404)
reply = client.get('/api/inventory_items/testuser1@example.com/' + str(private_item.id) + '/',
self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
@ -256,6 +260,98 @@ 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()
@ -265,9 +361,10 @@ class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, Invent
self.prepare_properties()
self.prepare_files()
self.prepare_inventory()
self.own_handle = str(self.f['local_user1'])
def test_post_item_with_file_id(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'test4',
'description': 'test',
'category': 'cat1',
@ -290,7 +387,7 @@ class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, Invent
self.assertEqual([f for f in item.files.all()], [self.f['test_file1']])
def test_post_item_with_encoded_file(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'test4',
'description': 'test',
'category': 'cat1',
@ -313,7 +410,7 @@ class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, Invent
self.assertEqual([f for f in item.files.all()], [self.f['test_file3']])
def test_post_item_with_file_id_fail(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'test4',
'description': 'test',
'category': 'cat1',
@ -325,7 +422,7 @@ class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, Invent
self.assertEqual(reply.status_code, 400)
def test_post_item_with_encoded_file_fail(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'test4',
'description': 'test',
'category': 'cat1',
@ -349,11 +446,11 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
self.prepare_files()
self.prepare_locations()
self.f['group1'].members.add(self.f['local_user2'].public_identity)
self.group_handle = '+' + str(self.f['group1'])[1:]
def create_group_item(self, name='drill'):
return client.post('/api/inventory_items/', self.f['local_user1'], {
return client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': name, 'owned_quantity': 1, 'availability_policy': 'private',
'owner_group': str(self.f['group1'])[1:],
})
def test_create_group_owned_item(self):
@ -366,16 +463,15 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
self.assertIsNone(reply.json()['owner'])
def test_create_group_owned_item_non_member_denied(self):
reply = client.post('/api/inventory_items/', self.f['ext_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['ext_user1'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'owner_group': str(self.f['group1'])[1:],
})
self.assertEqual(reply.status_code, 403)
self.assertEqual(InventoryItem.objects.count(), 0)
def test_other_member_can_edit(self):
item_id = self.create_group_item().json()['id']
reply = client.patch('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'], {
reply = client.patch('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['local_user2'], {
'name': 'drill-renamed'
})
self.assertEqual(reply.status_code, 200)
@ -383,18 +479,33 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
def test_other_member_can_delete(self):
item_id = self.create_group_item().json()['id']
reply = client.delete('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'])
reply = client.delete('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['local_user2'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(InventoryItem.objects.filter(id=item_id).count(), 0)
def test_delete_group_item_when_id_collides_with_own_personal_item(self):
own_handle = str(self.f['local_user2'])
personal_reply = client.post('/api/inventory_items/{}/'.format(own_handle), self.f['local_user2'], {
'name': 'personal-drill', 'owned_quantity': 1, 'availability_policy': 'private',
})
self.assertEqual(personal_reply.json()['id'], 1)
item_id = self.create_group_item().json()['id']
self.assertEqual(item_id, 1)
reply = client.delete('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['local_user2'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(InventoryItem.objects.filter(name='drill').count(), 0)
self.assertEqual(InventoryItem.objects.filter(name='personal-drill').count(), 1)
def test_remote_member_without_local_account_can_edit(self):
# A remote member (KnownIdentity, no ToolshedUser row) must still act on group-owned
# items - not unauthorized just because .user.exists() is False.
self.f['group1'].members.add(self.f['ext_user1'].public_identity)
item_id = self.create_group_item().json()['id']
reply = client.get('/api/inventory_items/{}/'.format(item_id), self.f['ext_user1'])
reply = client.get('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['ext_user1'])
self.assertEqual(reply.status_code, 200)
reply = client.patch('/api/inventory_items/{}/'.format(item_id), self.f['ext_user1'], {
reply = client.patch('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['ext_user1'], {
'name': 'drill-renamed-by-remote-member'
})
self.assertEqual(reply.status_code, 200)
@ -402,32 +513,32 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
def test_non_member_cannot_see_or_edit(self):
item_id = self.create_group_item().json()['id']
reply = client.get('/api/inventory_items/{}/'.format(item_id), self.f['ext_user1'])
reply = client.get('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)
def test_group_items_excluded_from_personal_list(self):
self.create_group_item()
reply = client.get('/api/inventory_items/', self.f['local_user1'])
reply = client.get('/api/inventory_items/{}/'.format(str(self.f['local_user1'])), self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0)
def test_group_items_listed_by_group_query_param(self):
def test_group_items_listed_by_owner_handle(self):
self.create_group_item()
reply = client.get('/api/inventory_items/?group={}'.format(str(self.f['group1'])[1:]), self.f['local_user2'])
reply = client.get('/api/inventory_items/{}/'.format(self.group_handle), 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):
def test_group_items_not_listed_for_non_member(self):
self.create_group_item()
reply = client.get('/api/inventory_items/?group={}'.format(str(self.f['group1'])[1:]), self.f['ext_user1'])
reply = client.get('/api/inventory_items/{}/'.format(self.group_handle), self.f['ext_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0)
def test_create_group_owned_item_with_full_fields(self):
# Parity with InventoryApiTestCase.test_post_new_item: tags/properties/category attach
# to a group-owned item the same way as a personal one.
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'availability_policy': 'rent',
'category': 'cat2',
'name': 'drill',
@ -435,7 +546,6 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
'owned_quantity': 3,
'tags': ['tag1', 'tag2'],
'properties': [{'name': 'prop1', 'value': 'value1'}, {'name': 'prop2', 'value': 'value2'}],
'owner_group': str(self.f['group1'])[1:],
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')
@ -450,22 +560,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': str(self.f['group1'])[1:],
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'availability_policy': 'private', 'owned_quantity': 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'], {
reply = client.post('/api/inventory_items/+nonexistent@example.com/', self.f['local_user1'], {
'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=nonexistent@example.com', self.f['local_user1'])
def test_group_items_listed_for_nonexistent_group(self):
reply = client.get('/api/inventory_items/+nonexistent@example.com/', self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0)
@ -473,7 +582,7 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
# Parity with InventoryApiTestCase.test_put_item, but as a PUT by a different member than
# the creator, to exercise the _is_authorized branch in perform_update for PUT too.
item_id = self.create_group_item().json()['id']
reply = client.put('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'], {
reply = client.put('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['local_user2'], {
'availability_policy': 'sell',
'name': 'drill-4000',
'description': 'new description',
@ -493,13 +602,12 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
def test_patch_group_item_clears_fields(self):
# Parity with InventoryApiTestCase.test_patch_item2 - clearing category/tags/properties.
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'category': 'cat1', 'tags': ['tag1'],
'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'], {
reply = client.patch('/api/inventory_items/{}/{}/'.format(self.group_handle, item_id), self.f['local_user2'], {
'category': None, 'tags': [], 'properties': []
})
self.assertEqual(reply.status_code, 200)
@ -508,9 +616,9 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
self.assertEqual([t for t in item.tags.all()], [])
def test_group_item_storage_location(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'storage_location': self.f['loc1'].id, 'owner_group': str(self.f['group1'])[1:],
'storage_location': self.f['loc1'].id,
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')
@ -518,19 +626,18 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTest
def test_post_group_item_with_file_id(self):
# Parity with TestInventoryItemWithFileApiTestCase.test_post_item_with_file_id.
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'files': [self.f['test_file1'].id], 'owner_group': str(self.f['group1'])[1:],
'files': [self.f['test_file1'].id],
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')
self.assertEqual([f for f in item.files.all()], [self.f['test_file1']])
def test_post_group_item_with_encoded_file(self):
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
reply = client.post('/api/inventory_items/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'files': [{'data': self.f['encoded_content3'], 'mime_type': 'text/plain'}],
'owner_group': str(self.f['group1'])[1:],
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')

View file

@ -16,6 +16,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.prepare_properties()
self.prepare_locations()
self.prepare_inventory()
self.own_handle = str(self.f['local_user1'])
def test_locations(self):
self.assertEqual("loc1", str(self.f['loc1']))
@ -30,7 +31,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(self.f['loc1'], self.f['loc4'].parent)
def test_get_inventory(self):
reply = client.get('/api/inventory_items/', self.f['local_user1'])
reply = client.get('/api/inventory_items/{}/'.format(self.own_handle), self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 2)
self.assertEqual(reply.json()[0]['name'], 'test1')
@ -50,7 +51,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()[1]['availability_policy'], 'friends')
def test_get_inventory_item(self):
reply = client.get('/api/storage_locations/', self.f['local_user1'])
reply = client.get('/api/storage_locations/{}/'.format(self.own_handle), self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 4)
self.assertEqual(reply.json()[0]['name'], 'loc1')
@ -71,7 +72,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()[3]['path'], 'loc1/loc4')
def test_post_new_location(self):
reply = client.post('/api/storage_locations/', self.f['local_user1'], {
reply = client.post('/api/storage_locations/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'loc5',
'description': 'a new location',
})
@ -84,7 +85,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()['path'], 'loc5')
def test_post_new_nested_location(self):
reply = client.post('/api/storage_locations/', self.f['local_user1'], {
reply = client.post('/api/storage_locations/{}/'.format(self.own_handle), self.f['local_user1'], {
'name': 'loc5',
'parent': self.f['loc3'].id,
})
@ -94,10 +95,11 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()['path'], 'loc1/loc3/loc5')
def test_patch_location(self):
reply = client.patch('/api/storage_locations/' + str(self.f['loc2'].id) + '/', self.f['local_user1'], {
'name': 'loc2-renamed',
'parent': self.f['loc1'].id,
})
reply = client.patch(
'/api/storage_locations/{}/{}/'.format(self.own_handle, self.f['loc2'].id), self.f['local_user1'], {
'name': 'loc2-renamed',
'parent': self.f['loc1'].id,
})
self.assertEqual(reply.status_code, 200)
location = StorageLocation.objects.get(id=self.f['loc2'].id)
self.assertEqual(location.name, 'loc2-renamed')
@ -105,7 +107,8 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()['path'], 'loc1/loc2-renamed')
def test_delete_location(self):
reply = client.delete('/api/storage_locations/' + str(self.f['loc4'].id) + '/', self.f['local_user1'])
reply = client.delete(
'/api/storage_locations/{}/{}/'.format(self.own_handle, self.f['loc4'].id), self.f['local_user1'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(StorageLocation.objects.count(), 3)
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc4'].id).count(), 0)
@ -113,20 +116,25 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
def test_delete_location_with_items_sets_null(self):
item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], name='located_item', storage_location=self.f['loc3'])
reply = client.delete('/api/storage_locations/' + str(self.f['loc3'].id) + '/', self.f['local_user1'])
reply = client.delete(
'/api/storage_locations/{}/{}/'.format(self.own_handle, self.f['loc3'].id), self.f['local_user1'])
self.assertEqual(reply.status_code, 204)
item.refresh_from_db()
self.assertIsNone(item.storage_location)
self.assertEqual(InventoryItem.objects.filter(id=item.id).count(), 1)
def test_locations_are_owner_scoped(self):
reply = client.get('/api/storage_locations/', self.f['local_user2'])
reply = client.get('/api/storage_locations/{}/'.format(str(self.f['local_user2'])), self.f['local_user2'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0)
def test_cannot_delete_other_users_location(self):
reply = client.delete('/api/storage_locations/' + str(self.f['loc1'].id) + '/', self.f['local_user2'])
self.assertEqual(reply.status_code, 404)
# local_user2 is a friend of local_user1 (see prepare_inventory), so local_user1's own
# handle resolves and the location is visible - friends can read but never write, so this
# is a 403 (found, not authorized), not a 404.
reply = client.delete(
'/api/storage_locations/{}/{}/'.format(self.own_handle, self.f['loc1'].id), self.f['local_user2'])
self.assertEqual(reply.status_code, 403)
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc1'].id).count(), 1)
@ -136,10 +144,13 @@ class GroupOwnedLocationApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestC
self.prepare_users()
self.prepare_groups()
self.f['group1'].members.add(self.f['local_user2'].public_identity)
# The '+' prefix is the '#'->'+' escape resolve_owner_handle expects for a group handle in
# the owner_handle URL path parameter - the same escape encodeHandleForUrl uses.
self.group_handle = '+' + str(self.f['group1'])[1:]
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:],
return client.post('/api/storage_locations/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': name,
})
def test_create_group_owned_location(self):
@ -152,67 +163,91 @@ class GroupOwnedLocationApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestC
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:],
reply = client.post('/api/storage_locations/{}/'.format(self.group_handle), self.f['ext_user1'], {
'name': 'shelf',
})
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'
})
reply = client.patch(
'/api/storage_locations/{}/{}/'.format(self.group_handle, 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'])
reply = client.delete(
'/api/storage_locations/{}/{}/'.format(self.group_handle, location_id), self.f['local_user2'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(StorageLocation.objects.filter(id=location_id).count(), 0)
def test_delete_group_location_when_id_collides_with_own_personal_location(self):
# Mirrors InventoryItemViewSet's own regression test - id is only unique within its own
# owner/owner_group scope, so a member's personal location and their group's location can
# land on the same id. Every route being scoped by an explicit owner_handle - never a
# bare, unscoped id - means this can no longer raise MultipleObjectsReturned.
own_handle = str(self.f['local_user2'])
personal_reply = client.post('/api/storage_locations/{}/'.format(own_handle), self.f['local_user2'],
{'name': 'personal-shelf'})
self.assertEqual(personal_reply.json()['id'], 1)
location_id = self.create_group_location().json()['id']
self.assertEqual(location_id, 1)
reply = client.delete(
'/api/storage_locations/{}/{}/'.format(self.group_handle, location_id), self.f['local_user2'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(StorageLocation.objects.filter(name='shelf').count(), 0)
self.assertEqual(StorageLocation.objects.filter(name='personal-shelf').count(), 1)
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'])
reply = client.get(
'/api/storage_locations/{}/{}/'.format(self.group_handle, 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'
})
reply = client.patch(
'/api/storage_locations/{}/{}/'.format(self.group_handle, 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'])
reply = client.get(
'/api/storage_locations/{}/{}/'.format(self.group_handle, 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'])
reply = client.get('/api/storage_locations/{}/'.format(str(self.f['local_user1'])), 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):
def test_group_locations_listed_by_owner_handle(self):
self.create_group_location()
reply = client.get('/api/storage_locations/?group={}'.format(str(self.f['group1'])[1:]), self.f['local_user2'])
reply = client.get('/api/storage_locations/{}/'.format(self.group_handle), 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):
def test_group_locations_not_listed_for_non_member(self):
self.create_group_location()
reply = client.get('/api/storage_locations/?group={}'.format(str(self.f['group1'])[1:]), self.f['ext_user1'])
reply = client.get('/api/storage_locations/{}/'.format(self.group_handle), 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,
reply = client.post('/api/storage_locations/{}/'.format(self.group_handle), self.f['local_user2'], {
'name': 'bin', 'parent': parent_id,
})
self.assertEqual(reply.status_code, 201)
self.assertEqual(reply.json()['path'], 'shelf/bin')
@ -224,15 +259,15 @@ class GroupOwnedLocationApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestC
# 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,
reply = client.post('/api/storage_locations/{}/'.format(self.group_handle), self.f['local_user1'], {
'name': 'bin', '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,
})
reply = client.post(
'/api/storage_locations/{}/'.format(str(self.f['ext_user1'])), self.f['ext_user1'], {
'name': 'bin', 'parent': parent_id,
})
self.assertEqual(reply.status_code, 400)

View file

@ -204,6 +204,22 @@ export function deserializeShortId(ints) {
return schema.interpret(fieldValues)
}
export function encodeDomainQualifiedShortId(domain, ints) {
return domain + ':' + encodeShortId(ints)
}
export function decodeDomainQualifiedShortId(text) {
const i = text.indexOf(':~')
if (i === -1) {
throw new Error("not a domain-qualified short id: expected '<domain>:~<token>'")
}
return {domain: text.slice(0, i), token: text.slice(i + 1), ints: decodeShortId(text.slice(i + 1))}
}
export function isDomainQualifiedShortId(text) {
return typeof text === 'string' && /^[^\s~]+:~/.test(text)
}
export function serializeShortId({kind, ...fieldValues}) {
const entry = Object.entries(SCHEMAS).find(([, s]) => s.name === kind)
if (!entry) {

View file

@ -1,4 +1,7 @@
import {encodeShortId, decodeShortId, deserializeShortId, serializeShortId} from '../short-id.js'
import {
encodeShortId, decodeShortId, deserializeShortId, serializeShortId,
encodeDomainQualifiedShortId, decodeDomainQualifiedShortId, isDomainQualifiedShortId
} from '../short-id.js'
test('encodes the worked example from docs/handles-and-shortids.md', () => {
const token = encodeShortId([0, 7, 42])
@ -108,3 +111,34 @@ test('rejects a missing field', () => {
test('rejects a negative field value', () => {
expect(() => encodeShortId([0, -1, 42])).toThrow()
})
test('encodes a Domain-Qualified Short ID as "<domain>:<token>"', () => {
expect(encodeDomainQualifiedShortId('toolsheddomain.tld', [0, 7, 42])).toBe('toolsheddomain.tld:~DyU')
})
test('decodes a Domain-Qualified Short ID back into its domain and ints', () => {
const decoded = decodeDomainQualifiedShortId('toolsheddomain.tld:~DyU')
expect(decoded.domain).toBe('toolsheddomain.tld')
expect(decoded.token).toBe('~DyU')
expect(decoded.ints).toEqual([0, 7, 42])
})
test('round-trips a Domain-Qualified Short ID through encode/decode', () => {
const named = {kind: 'storage_location', owner_identity_id: 3, storage_location_id: 1000}
const encoded = encodeDomainQualifiedShortId('example.com', serializeShortId(named))
const {domain, ints} = decodeDomainQualifiedShortId(encoded)
expect(domain).toBe('example.com')
expect(deserializeShortId(ints)).toEqual(named)
})
test('rejects a Domain-Qualified Short ID missing the "~" after the domain', () => {
expect(() => decodeDomainQualifiedShortId('toolsheddomain.tldDyU')).toThrow()
expect(() => decodeDomainQualifiedShortId('~DyU')).toThrow()
})
test('isDomainQualifiedShortId tells a "<domain>:~token" apart from a bare token', () => {
expect(isDomainQualifiedShortId('toolsheddomain.tld:~DyU')).toBe(true)
expect(isDomainQualifiedShortId('~DyU')).toBe(false)
expect(isDomainQualifiedShortId('alice@example.com:i42')).toBe(false)
expect(isDomainQualifiedShortId('not a short id at all')).toBe(false)
})

View file

@ -73,7 +73,7 @@
</div>
<div class="btn-group mt-2">
<button class="btn btn-danger btn-sm"
@click="deleteStorageLocation(location.id)">Delete
@click="deleteStorageLocation(location)">Delete
</button>
<router-link :to="`${locationRoute(location)}/edit`"
class="btn btn-primary btn-sm">Edit