This commit is contained in:
j3d1 2026-08-23 14:07:52 +02:00
parent bbe52e4a78
commit c3b43379a3
11 changed files with 362 additions and 56 deletions

View file

@ -4,12 +4,31 @@ from rest_framework.decorators import api_view, permission_classes, authenticati
from rest_framework.permissions import IsAuthenticated from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response from rest_framework.response import Response
from authentication.signature_auth import SignatureAuthenticationLocal from authentication.signature_auth import SignatureAuthentication, SignatureAuthenticationLocal
from files.models import File from files.models import File
from files.serializers import FileSerializer from files.serializers import FileSerializer
from toolshed.models import InventoryItem, WorkflowInstance from toolshed.models import InventoryItem, WorkflowInstance
def _get_authorized_item(identity, item_id):
"""Look up an item by its owner-scoped id and confirm identity may act on it - either as its
personal owner (requires a local ToolshedUser account) or as a current member of its owning
group (works for a remote member too, since group membership is identity-level, see
docs/design-in-progress/groups-mvp.md). id is only unique within one owner/group's own items,
so the lookup itself must be scoped rather than a bare global get. Returns None if not found
or not authorized, the same shape InventoryItem.DoesNotExist handling around it already
expects."""
if identity.user.exists():
try:
return InventoryItem.objects.get(owner=identity.user.get(), id=item_id)
except InventoryItem.DoesNotExist:
pass
try:
return InventoryItem.objects.get(owner_group__in=identity.member_of_groups.all(), id=item_id)
except InventoryItem.DoesNotExist:
return None
@api_view(['GET']) @api_view(['GET'])
@permission_classes([IsAuthenticated]) @permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthenticationLocal]) @authentication_classes([SignatureAuthenticationLocal])
@ -19,35 +38,37 @@ def list_all_files(request, format=None): # /files/
def get_item_files(request, item_id): def get_item_files(request, item_id):
try: item = _get_authorized_item(request.user, item_id)
item = InventoryItem.objects.get(id=item_id, owner=request.user) if item is None:
files = item.files.all()
return Response(FileSerializer(files, many=True).data)
except InventoryItem.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND) return Response(status=status.HTTP_404_NOT_FOUND)
files = item.files.all()
return Response(FileSerializer(files, many=True).data)
def post_item_file(request, item_id): def post_item_file(request, item_id):
try: item = _get_authorized_item(request.user, item_id)
item = InventoryItem.objects.get(id=item_id, owner=request.user) if item is None:
if 'file_hash' in request.data:
# Attach a file the caller already staged on one of their own workflows, identified
# by its content hash (which the client already computed before ever uploading it),
# instead of re-uploading bytes that are already stored server-side.
try:
file = File.objects.get(hash=request.data['file_hash'], staged_by_workflows__owner=request.user)
except File.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
item.files.add(file)
return Response(FileSerializer(file).data, status=status.HTTP_201_CREATED)
serializer = FileSerializer(data=request.data)
if serializer.is_valid():
file = serializer.save()
item.files.add(file)
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
except InventoryItem.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND) return Response(status=status.HTTP_404_NOT_FOUND)
if 'file_hash' in request.data:
# Attach a file the caller already staged on one of their own workflows, identified
# by its content hash (which the client already computed before ever uploading it),
# instead of re-uploading bytes that are already stored server-side. Workflows are
# always personally owned, so this only applies to a caller with a local account.
if not request.user.user.exists():
return Response(status=status.HTTP_404_NOT_FOUND)
try:
file = File.objects.get(hash=request.data['file_hash'],
staged_by_workflows__owner=request.user.user.get())
except File.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
item.files.add(file)
return Response(FileSerializer(file).data, status=status.HTTP_201_CREATED)
serializer = FileSerializer(data=request.data)
if serializer.is_valid():
file = serializer.save()
item.files.add(file)
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
def get_staged_files(request, workflow_id): def get_staged_files(request, workflow_id):
@ -77,7 +98,7 @@ def post_staged_file(request, workflow_id):
@api_view(['POST', 'GET']) @api_view(['POST', 'GET'])
@permission_classes([IsAuthenticated]) @permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthenticationLocal]) @authentication_classes([SignatureAuthentication])
def item_files(request, item_id, format=None): # /item_files/ def item_files(request, item_id, format=None): # /item_files/
if request.method == 'GET': if request.method == 'GET':
return get_item_files(request, item_id) return get_item_files(request, item_id)
@ -87,21 +108,21 @@ def item_files(request, item_id, format=None): # /item_files/
@api_view(['DELETE']) @api_view(['DELETE'])
@permission_classes([IsAuthenticated]) @permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthenticationLocal]) @authentication_classes([SignatureAuthentication])
def delete_item_file(request, item_id, file_id, format=None): # /item_files/ def delete_item_file(request, item_id, file_id, format=None): # /item_files/
try: item = _get_authorized_item(request.user, item_id)
item = InventoryItem.objects.get(id=item_id, owner=request.user) if item is None:
file = item.files.get(id=file_id)
item.files.remove(file_id)
if file.connected_items.count() == 0 and file.profile_picture_users.count() == 0 \
and file.staged_by_workflows.count() == 0:
file.file.delete(save=False)
file.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
except InventoryItem.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND) return Response(status=status.HTTP_404_NOT_FOUND)
try:
file = item.files.get(id=file_id)
except File.DoesNotExist: except File.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND) return Response(status=status.HTTP_404_NOT_FOUND)
item.files.remove(file_id)
if file.connected_items.count() == 0 and file.profile_picture_users.count() == 0 \
and file.staged_by_workflows.count() == 0:
file.file.delete(save=False)
file.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
@api_view(['POST', 'GET']) @api_view(['POST', 'GET'])

View file

@ -35,6 +35,12 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
serializer_class = InventoryItemSerializer serializer_class = InventoryItemSerializer
authentication_classes = [SignatureAuthentication] authentication_classes = [SignatureAuthentication]
permission_classes = [IsAuthenticated] permission_classes = [IsAuthenticated]
# Detail routes address an item by its owner-scoped id, not the internal row id - the
# router still names the URL capture group 'pk', so keep that as lookup_url_kwarg and just
# change which model field it's matched against. get_queryset() below is always already
# scoped to the requester's own items/groups, so this can't cross into another owner's ids.
lookup_field = 'id'
lookup_url_kwarg = 'pk'
def get_queryset(self): def get_queryset(self):
# A KnownIdentity acting purely as a group member (e.g. a remote member on a group # A KnownIdentity acting purely as a group member (e.g. a remote member on a group

View file

@ -1,4 +1,4 @@
from django.db import models from django.db import models, transaction
from django.core.validators import MinValueValidator, MaxValueValidator from django.core.validators import MinValueValidator, MaxValueValidator
from django_softdelete.models import SoftDeleteModel from django_softdelete.models import SoftDeleteModel
from rest_framework.exceptions import ValidationError from rest_framework.exceptions import ValidationError
@ -82,6 +82,31 @@ class Tag(models.Model):
return f"{self.origin}#tag:{self.name}" return f"{self.origin}#tag:{self.name}"
class OwnerItemSequence(models.Model):
"""Tracks the last InventoryItem id handed out to a given owner or owner_group, so ids can
be allocated sequentially and without gaps within that scope (see InventoryItem.create_for_owner).
Exactly one of owner/owner_group is set, mirroring InventoryItem's own owner/owner_group 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='owneritemsequence_unique_owner'),
models.UniqueConstraint(fields=['owner_group'], condition=models.Q(owner_group__isnull=False),
name='owneritemsequence_unique_owner_group'),
]
@classmethod
def allocate(cls, *, owner=None, owner_group=None):
with transaction.atomic():
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
class InventoryItem(SoftDeleteModel): class InventoryItem(SoftDeleteModel):
AVAILABILITY_POLICY_CHOICES = ( AVAILABILITY_POLICY_CHOICES = (
('sell', 'Sell'), ('sell', 'Sell'),
@ -91,6 +116,11 @@ class InventoryItem(SoftDeleteModel):
('private', 'Private'), ('private', 'Private'),
) )
internal_id = models.AutoField(primary_key=True)
# The externally visible identifier: sequential and gapless within owner/owner_group's own
# items (see OwnerItemSequence), never the internal_id above. Always allocate through
# create_for_owner rather than InventoryItem.objects.create() directly.
id = models.PositiveIntegerField(editable=False)
published = models.BooleanField(default=False) published = models.BooleanField(default=False)
name = models.CharField(max_length=255, null=True, blank=True) name = models.CharField(max_length=255, null=True, blank=True)
description = models.TextField(null=True, blank=True) description = models.TextField(null=True, blank=True)
@ -108,12 +138,26 @@ class InventoryItem(SoftDeleteModel):
storage_location = models.ForeignKey('StorageLocation', on_delete=models.SET_NULL, null=True, blank=True, storage_location = models.ForeignKey('StorageLocation', on_delete=models.SET_NULL, null=True, blank=True,
related_name='inventory_items') related_name='inventory_items')
class Meta:
constraints = [
models.UniqueConstraint(fields=['owner', 'owner_group', 'id'],
name='inventoryitem_unique_owner_scoped_id'),
]
def clean(self): def clean(self):
if (self.name is None or self.name == "") and self.files.count() == 0: if (self.name is None or self.name == "") and self.files.count() == 0:
raise ValidationError("Name or at least one file must be set") raise ValidationError("Name or at least one file must be set")
if (self.owner is None) == (self.owner_group is None): if (self.owner is None) == (self.owner_group is None):
raise ValidationError("Exactly one of owner or owner_group must be set") raise ValidationError("Exactly one of owner or owner_group must be set")
@classmethod
def create_for_owner(cls, *, owner=None, owner_group=None, **kwargs):
"""The only supported way to create an InventoryItem: allocates the next id for this
owner/owner_group scope and creates the item with it, atomically."""
with transaction.atomic():
next_id = OwnerItemSequence.allocate(owner=owner, owner_group=owner_group)
return cls.objects.create(owner=owner, owner_group=owner_group, id=next_id, **kwargs)
class ItemProperty(models.Model): class ItemProperty(models.Model):
property = models.ForeignKey(Property, on_delete=models.CASCADE) property = models.ForeignKey(Property, on_delete=models.CASCADE)

View file

@ -638,7 +638,7 @@ def import_inventory(user, data, available_files):
except (TypeError, ValueError): except (TypeError, ValueError):
owned_quantity = 1 owned_quantity = 1
item = InventoryItem.objects.create( item = InventoryItem.create_for_owner(
owner=user, owner=user,
name=name or None, name=name or None,
description=row.get('description', '') or '', description=row.get('description', '') or '',

View file

@ -200,6 +200,7 @@ class InventoryItemSerializer(serializers.ModelSerializer):
model = InventoryItem model = InventoryItem
fields = ['id', 'name', 'description', 'owner', 'owner_group', 'category', 'availability_policy', fields = ['id', 'name', 'description', 'owner', 'owner_group', 'category', 'availability_policy',
'owned_quantity', 'tags', 'tags_input', 'properties', 'files', 'storage_location'] 'owned_quantity', 'tags', 'tags_input', 'properties', 'files', 'storage_location']
read_only_fields = ['id']
def get_tags(self, obj): def get_tags(self, obj):
return [tag.name for tag in obj.tags.all()] return [tag.name for tag in obj.tags.all()]
@ -216,7 +217,7 @@ class InventoryItemSerializer(serializers.ModelSerializer):
tags = validated_data.pop('tags', []) tags = validated_data.pop('tags', [])
props = validated_data.pop('itemproperty_set', []) props = validated_data.pop('itemproperty_set', [])
files = validated_data.pop('files', []) files = validated_data.pop('files', [])
item = InventoryItem.objects.create(**validated_data) item = InventoryItem.create_for_owner(**validated_data)
for tag in tags: for tag in tags:
item.tags.add(tag, through_defaults={}) item.tags.add(tag, through_defaults={})
for prop in props: for prop in props:

View file

@ -35,10 +35,10 @@ class InventoryTestMixin(CategoryTestMixin, TagTestMixin, PropertyTestMixin):
def prepare_inventory(self): def prepare_inventory(self):
self.f['local_user1'].friends.add(self.f['local_user2'].public_identity) self.f['local_user1'].friends.add(self.f['local_user2'].public_identity)
self.f['item1'] = InventoryItem.objects.create( self.f['item1'] = InventoryItem.create_for_owner(
owner=self.f['local_user1'], owned_quantity=1, name='test1', description='test', category=self.f['cat1'], owner=self.f['local_user1'], owned_quantity=1, name='test1', description='test', category=self.f['cat1'],
availability_policy='friends') availability_policy='friends')
self.f['item2'] = InventoryItem.objects.create( self.f['item2'] = InventoryItem.create_for_owner(
owner=self.f['local_user1'], owned_quantity=1, name='test2', description='test2', category=self.f['cat1'], owner=self.f['local_user1'], owned_quantity=1, name='test2', description='test2', category=self.f['cat1'],
availability_policy='friends') availability_policy='friends')
self.f['item2'].tags.add(self.f['tag1'], through_defaults={}) self.f['item2'].tags.add(self.f['tag1'], through_defaults={})

View file

@ -1,7 +1,7 @@
from django.test import Client from django.test import Client
from authentication.tests import SignatureAuthClient, UserTestMixin, ToolshedTestCase from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
from files.tests import FilesTestMixin from files.tests import FilesTestMixin
from toolshed.models import File from toolshed.models import File, InventoryItem
from toolshed.tests import InventoryTestMixin from toolshed.tests import InventoryTestMixin
@ -156,3 +156,60 @@ class FileApiTestCase(UserTestMixin, FilesTestMixin, InventoryTestMixin, Toolshe
self.assertEqual(reply.json()[0]['files'][0]['mime_type'], 'text/plain') self.assertEqual(reply.json()[0]['files'][0]['mime_type'], 'text/plain')
self.assertEqual(reply.json()[0]['files'][1]['mime_type'], 'text/plain') self.assertEqual(reply.json()[0]['files'][1]['mime_type'], 'text/plain')
self.assertEqual(reply.json()[1]['files'][0]['mime_type'], 'text/plain') self.assertEqual(reply.json()[1]['files'][0]['mime_type'], 'text/plain')
class GroupOwnedFileApiTestCase(UserTestMixin, GroupTestMixin, FilesTestMixin, ToolshedTestCase):
def setUp(self):
super().setUp()
self.prepare_users()
self.prepare_groups()
self.prepare_files()
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', availability_policy='private')
self.f['group_item'].files.add(self.f['test_file1'])
def test_get_group_item_files(self):
response = client.get(f"/api/item_files/{self.f['group_item'].id}/", self.f['local_user1'])
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 1)
def test_other_member_can_post_file(self):
response = client.post(f"/api/item_files/{self.f['group_item'].id}/", self.f['local_user2'],
{'data': self.f['encoded_content4'], 'mime_type': 'text/plain'})
self.assertEqual(response.status_code, 201)
self.assertEqual(self.f['group_item'].files.count(), 2)
def test_remote_member_without_local_account_can_post_file(self):
self.f['group1'].members.add(self.f['ext_user1'].public_identity)
response = client.post(f"/api/item_files/{self.f['group_item'].id}/", self.f['ext_user1'],
{'data': self.f['encoded_content4'], 'mime_type': 'text/plain'})
self.assertEqual(response.status_code, 201)
self.assertEqual(self.f['group_item'].files.count(), 2)
def test_remote_member_without_local_account_can_get_files(self):
self.f['group1'].members.add(self.f['ext_user1'].public_identity)
response = client.get(f"/api/item_files/{self.f['group_item'].id}/", self.f['ext_user1'])
self.assertEqual(response.status_code, 200)
def test_non_member_cannot_post_file(self):
response = client.post(f"/api/item_files/{self.f['group_item'].id}/", self.f['ext_user1'],
{'data': self.f['encoded_content4'], 'mime_type': 'text/plain'})
self.assertEqual(response.status_code, 404)
self.assertEqual(self.f['group_item'].files.count(), 1)
def test_non_member_cannot_get_files(self):
response = client.get(f"/api/item_files/{self.f['group_item'].id}/", self.f['ext_user1'])
self.assertEqual(response.status_code, 404)
def test_other_member_can_delete_file(self):
response = client.delete(f"/api/item_files/{self.f['group_item'].id}/{self.f['test_file1'].id}/",
self.f['local_user2'])
self.assertEqual(response.status_code, 204)
self.assertEqual(self.f['group_item'].files.count(), 0)
def test_non_member_cannot_delete_file(self):
response = client.delete(f"/api/item_files/{self.f['group_item'].id}/{self.f['test_file1'].id}/",
self.f['ext_user1'])
self.assertEqual(response.status_code, 404)
self.assertEqual(self.f['group_item'].files.count(), 1)

View file

@ -1,7 +1,8 @@
from authentication.models import Group
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase
from files.tests import FilesTestMixin from files.tests import FilesTestMixin
from toolshed.models import InventoryItem, Category from toolshed.models import InventoryItem, Category
from toolshed.tests import InventoryTestMixin from toolshed.tests import InventoryTestMixin, CategoryTestMixin, TagTestMixin, PropertyTestMixin, LocationTestMixin
client = SignatureAuthClient() client = SignatureAuthClient()
@ -232,7 +233,7 @@ class InventoryApiTestCase(UserTestMixin, InventoryTestMixin, ToolshedTestCase):
self.assertEqual(reply.status_code, 403) self.assertEqual(reply.status_code, 403)
def test_get_shared_item_private(self): def test_get_shared_item_private(self):
private_item = InventoryItem.objects.create( private_item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], owned_quantity=1, name='secret', availability_policy='private') 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) + '/', reply = client.get('/api/inventory_items/testuser1@example.com/' + str(private_item.id) + '/',
self.f['local_user2']) self.f['local_user2'])
@ -336,11 +337,17 @@ class TestInventoryItemWithFileApiTestCase(UserTestMixin, FilesTestMixin, Invent
self.assertEqual(reply.status_code, 400) self.assertEqual(reply.status_code, 400)
class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase): class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, CategoryTestMixin, TagTestMixin,
PropertyTestMixin, FilesTestMixin, LocationTestMixin, ToolshedTestCase):
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.prepare_users() self.prepare_users()
self.prepare_groups() self.prepare_groups()
self.prepare_categories()
self.prepare_tags()
self.prepare_properties()
self.prepare_files()
self.prepare_locations()
self.f['group1'].members.add(self.f['local_user2'].public_identity) self.f['group1'].members.add(self.f['local_user2'].public_identity)
def create_group_item(self, name='drill'): def create_group_item(self, name='drill'):
@ -416,4 +423,162 @@ class GroupOwnedInventoryApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTest
self.create_group_item() 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(self.f['group1'].id), self.f['ext_user1'])
self.assertEqual(reply.status_code, 200) self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0) 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 must
# attach to a group-owned item exactly the same way they do for a personal one.
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
'availability_policy': 'rent',
'category': 'cat2',
'name': 'drill',
'description': 'test',
'owned_quantity': 3,
'tags': ['tag1', 'tag2'],
'properties': [{'name': 'prop1', 'value': 'value1'}, {'name': 'prop2', 'value': 'value2'}],
'owner_group': self.f['group1'].id,
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')
self.assertIsNone(item.owner)
self.assertEqual(item.owner_group, self.f['group1'])
self.assertEqual(item.availability_policy, 'rent')
self.assertEqual(item.category, Category.objects.get(name='cat2'))
self.assertEqual(item.owned_quantity, 3)
self.assertEqual([t for t in item.tags.all()], [self.f['tag1'], self.f['tag2']])
self.assertEqual([p for p in item.properties.all()], [self.f['prop1'], self.f['prop2']])
self.assertEqual([p.value for p in item.itemproperty_set.all()], ['value1', 'value2'])
def test_create_group_owned_item_empty_fails(self):
# Parity with InventoryApiTestCase.test_post_new_item_empty - clean()'s name-or-files
# validation must still apply to group-owned items.
reply = client.post('/api/inventory_items/', self.f['local_user1'], {
'availability_policy': 'private', 'owned_quantity': 1, 'owner_group': self.f['group1'].id,
})
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,
})
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'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(len(reply.json()), 0)
def test_put_group_item(self):
# Parity with InventoryApiTestCase.test_put_item - full replace, by a different member
# than the one who created it, exercising the group_items_id -> _is_authorized branch
# in perform_update for a PUT (not just PATCH).
item_id = self.create_group_item().json()['id']
reply = client.put('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'], {
'availability_policy': 'sell',
'name': 'drill-4000',
'description': 'new description',
'owned_quantity': 100,
'tags': ['tag1', 'tag3'],
'properties': [{'name': 'prop1', 'value': 'value5'}],
})
self.assertEqual(reply.status_code, 200)
item = InventoryItem.objects.get(id=item_id)
self.assertEqual(item.owner_group, self.f['group1'])
self.assertEqual(item.availability_policy, 'sell')
self.assertEqual(item.name, 'drill-4000')
self.assertEqual(item.description, 'new description')
self.assertEqual(item.owned_quantity, 100)
self.assertEqual([t for t in item.tags.all()], [self.f['tag1'], self.f['tag3']])
self.assertEqual([p.value for p in item.itemproperty_set.all()], ['value5'])
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'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'category': 'cat1', 'tags': ['tag1'],
'owner_group': self.f['group1'].id,
})
item_id = reply.json()['id']
reply = client.patch('/api/inventory_items/{}/'.format(item_id), self.f['local_user2'], {
'category': None, 'tags': [], 'properties': []
})
self.assertEqual(reply.status_code, 200)
item = InventoryItem.objects.get(id=item_id)
self.assertEqual(item.category, None)
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'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'storage_location': self.f['loc1'].id, 'owner_group': self.f['group1'].id,
})
self.assertEqual(reply.status_code, 201)
item = InventoryItem.objects.get(name='drill')
self.assertEqual(item.storage_location, self.f['loc1'])
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'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'files': [self.f['test_file1'].id], 'owner_group': self.f['group1'].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'], {
'name': 'drill', 'owned_quantity': 1, 'availability_policy': 'private',
'files': [{'data': self.f['encoded_content3'], 'mime_type': 'text/plain'}],
'owner_group': self.f['group1'].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_file3']])
def test_group_items_excluded_from_search(self):
# Group-owned items are only ever reachable via the group's own detail page for MVP
# (see docs/design-in-progress/groups-mvp.md) - search must not surface them, same as
# the main Inventory list already doesn't.
self.create_group_item(name='searchable-drill')
InventoryItem.create_for_owner(owner=self.f['local_user1'], owned_quantity=1, name='searchable-personal')
reply = client.get('/api/search/?query=searchable', self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
names = [item['name'] for item in reply.json()]
self.assertEqual(names, ['searchable-personal'])
class InventoryItemIdAllocationTestCase(UserTestMixin, ToolshedTestCase):
"""InventoryItem.id is sequential and gapless within each owner/owner_group's own items,
never reused, and allocated independently per scope - see OwnerItemSequence."""
def setUp(self):
super().setUp()
self.prepare_users()
def test_ids_are_sequential_and_independent_per_owner(self):
user1_items = [InventoryItem.create_for_owner(owner=self.f['local_user1'], name=f'u1-{i}')
for i in range(3)]
user2_items = [InventoryItem.create_for_owner(owner=self.f['local_user2'], name=f'u2-{i}')
for i in range(2)]
self.assertEqual([item.id for item in user1_items], [1, 2, 3])
self.assertEqual([item.id for item in user2_items], [1, 2])
def test_deleted_item_id_is_never_reused(self):
item1 = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='first')
item2 = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='second')
self.assertEqual((item1.id, item2.id), (1, 2))
item2.delete() # soft delete - item2's row (and its id) stays in the table
item3 = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='third')
self.assertEqual(item3.id, 3)
self.assertFalse(InventoryItem.objects.filter(owner=self.f['local_user1'], id=2).exists())
self.assertTrue(InventoryItem.global_objects.filter(owner=self.f['local_user1'], id=2).exists())
def test_group_scope_has_independent_sequence(self):
group = Group.objects.create(name='alloc-test-group', domain=self.f['example_com'].name)
personal_item = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='personal')
group_item = InventoryItem.create_for_owner(owner_group=group, name='group-owned')
self.assertEqual(personal_item.id, 1)
self.assertEqual(group_item.id, 1)

View file

@ -111,7 +111,7 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc4'].id).count(), 0) self.assertEqual(StorageLocation.objects.filter(id=self.f['loc4'].id).count(), 0)
def test_delete_location_with_items_sets_null(self): def test_delete_location_with_items_sets_null(self):
item = InventoryItem.objects.create( item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], name='located_item', storage_location=self.f['loc3']) 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/' + str(self.f['loc3'].id) + '/', self.f['local_user1'])
self.assertEqual(reply.status_code, 204) self.assertEqual(reply.status_code, 204)

View file

@ -28,11 +28,11 @@ class _DeleteTestDataMixin(UserTestMixin, CategoryTestMixin, LocationTestMixin):
self.f['orphan_file'] = File.objects.create( self.f['orphan_file'] = File.objects.create(
file=ContentFile(b'orphan', 'orphan'), mime_type='text/plain', hash='orphan') file=ContentFile(b'orphan', 'orphan'), mime_type='text/plain', hash='orphan')
self.f['item1'] = InventoryItem.objects.create( self.f['item1'] = InventoryItem.create_for_owner(
owner=self.f['local_user1'], owned_quantity=1, name='item1', category=self.f['cat1']) owner=self.f['local_user1'], owned_quantity=1, name='item1', category=self.f['cat1'])
self.f['item1'].files.add(self.f['orphan_file']) self.f['item1'].files.add(self.f['orphan_file'])
self.f['item_other_user'] = InventoryItem.objects.create( self.f['item_other_user'] = InventoryItem.create_for_owner(
owner=self.f['local_user2'], owned_quantity=1, name='item2', category=self.f['cat1']) owner=self.f['local_user2'], owned_quantity=1, name='item2', category=self.f['cat1'])
self.f['item_other_user'].files.add(self.f['shared_file']) self.f['item_other_user'].files.add(self.f['shared_file'])
@ -136,7 +136,7 @@ class ImportInventoryPropertiesTestCase(UserTestMixin, CategoryTestMixin, TagTes
self.prepare_properties() self.prepare_properties()
def test_property_values_with_comma_and_equals_round_trip(self): def test_property_values_with_comma_and_equals_round_trip(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='widget') item = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='widget')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='10cm, 20cm') ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='10cm, 20cm')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop2'], value='a=b') ItemProperty.objects.create(inventory_item=item, property=self.f['prop2'], value='a=b')
@ -167,7 +167,7 @@ class ImportInventoryPropertiesTestCase(UserTestMixin, CategoryTestMixin, TagTes
self.assertEqual(values, {'prop1': 'value1', 'prop2': 'value2'}) self.assertEqual(values, {'prop1': 'value1', 'prop2': 'value2'})
def test_item_without_properties_imports_cleanly(self): def test_item_without_properties_imports_cleanly(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='bare item') item = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='bare item')
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1'])))) csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
@ -179,7 +179,7 @@ class ImportInventoryPropertiesTestCase(UserTestMixin, CategoryTestMixin, TagTes
self.assertEqual(list(new_item.itemproperty_set.all()), []) self.assertEqual(list(new_item.itemproperty_set.all()), [])
def test_category_and_tags_round_trip(self): def test_category_and_tags_round_trip(self):
item = InventoryItem.objects.create( item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], name='cat and tags item', category=self.f['cat1']) owner=self.f['local_user1'], name='cat and tags item', category=self.f['cat1'])
item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={}) item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={})
@ -206,7 +206,7 @@ class ImportInventoryPropertiesTestCase(UserTestMixin, CategoryTestMixin, TagTes
self.assertFalse(InventoryItem.objects.filter(owner=self.f['local_user1'], name='ghost widget').exists()) self.assertFalse(InventoryItem.objects.filter(owner=self.f['local_user1'], name='ghost widget').exists())
def test_property_value_with_quote_character_round_trips(self): def test_property_value_with_quote_character_round_trips(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='quoted widget') item = InventoryItem.create_for_owner(owner=self.f['local_user1'], name='quoted widget')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='12" screen') ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='12" screen')
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1'])))) csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
@ -237,7 +237,7 @@ class ExportImportApiRoundTripTestCase(UserTestMixin, CategoryTestMixin, TagTest
def test_export_then_import_preserves_category_tags_and_properties(self): def test_export_then_import_preserves_category_tags_and_properties(self):
import base64 import base64
item = InventoryItem.objects.create( item = InventoryItem.create_for_owner(
owner=self.f['local_user1'], name='drill', description='cordless drill', owner=self.f['local_user1'], name='drill', description='cordless drill',
category=self.f['cat1'], availability_policy='friends', owned_quantity=2) category=self.f['cat1'], availability_policy='friends', owned_quantity=2)
item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={}) item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={})

View file

@ -197,6 +197,18 @@ function relation(node, ownAxis, wantWidth, pxPerMm) {
if (combinesAsWidth === wantWidth) { if (combinesAsWidth === wantWidth) {
return {a, b}; return {a, b};
} }
if (a === 0) {
// Every child is a fixed size (a === 0) in the combining direction - e.g. a row that's
// just one crisp QR leaf, with no scale-free (text) sibling to invert against. Inverting
// "width = b" for an a of 0 would divide by zero: a constant width genuinely doesn't
// determine a height, since nothing here actually scales with it. Ask each child directly
// for its own size in the wanted direction instead (every one of them must be similarly
// fixed, since only a fixed leaf ever contributes a === 0), and take the largest - the
// shared dimension has to fit whichever child needs the most room, with any child that
// ends up with room to spare centered within it (see drawQrLeaf).
const otherParts = node.map(child => relation(child, axis, wantWidth, pxPerMm));
return {a: 0, b: Math.max(...otherParts.map(p => p.b))};
}
return {a: 1 / a, b: -b / a}; // invert: solve the affine relation the other way return {a: 1 / a, b: -b / a}; // invert: solve the affine relation the other way
} }