This commit is contained in:
j3d1 2026-07-23 04:41:04 +02:00
parent 7c91661be2
commit 796fef81be
37 changed files with 3404 additions and 108 deletions

View file

@ -11,6 +11,8 @@ from rest_framework.response import Response
from authentication.models import ToolshedUser
from authentication.signature_auth import SignatureAuthenticationLocal
from files.models import File
from files.serializers import FileSerializer
from hostadmin.models import Domain
router = routers.SimpleRouter()
@ -53,15 +55,43 @@ class UserViewSet(viewsets.ModelViewSet):
permission_classes = [IsAuthenticated, IsAdminUser]
@api_view(['GET'])
@api_view(['GET', 'PATCH'])
@permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthenticationLocal])
def getUserInfo(request):
user = request.user
if request.method == 'PATCH':
old_file = user.profile_picture
if 'profile_picture' in request.data:
profile_picture = request.data.get('profile_picture')
if profile_picture is None:
user.profile_picture = None
elif type(profile_picture) == dict:
serializer = FileSerializer(data=profile_picture)
if not serializer.is_valid():
return Response(serializer.errors, status=400)
user.profile_picture = serializer.save()
else:
return Response({'profile_picture': 'Must be null or an object with data and mime_type.'}, status=400)
elif 'profile_picture_id' in request.data:
profile_picture_id = request.data.get('profile_picture_id')
if profile_picture_id is None:
user.profile_picture = None
else:
try:
user.profile_picture = File.objects.get(id=profile_picture_id)
except File.DoesNotExist:
return Response({'profile_picture_id': 'File does not exist.'}, status=400)
user.save()
if old_file and old_file != user.profile_picture and old_file.connected_items.count() == 0 and old_file.profile_picture_users.count() == 0:
old_file.delete()
return Response({
'username': user.username,
'domain': user.domain,
'email': user.email
'email': user.email,
'profile_picture': FileSerializer(user.profile_picture).data if user.profile_picture else None,
})

View file

@ -0,0 +1,20 @@
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('files', '0001_initial'),
('authentication', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='toolsheduser',
name='profile_picture',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL,
related_name='profile_picture_users', to='files.file'),
),
]

View file

@ -86,6 +86,8 @@ class ToolshedUser(AbstractUser):
domain = models.CharField(max_length=255, default='localhost')
private_key = models.CharField(max_length=255)
public_identity = models.ForeignKey(KnownIdentity, on_delete=models.CASCADE, related_name='user')
profile_picture = models.ForeignKey('files.File', on_delete=models.SET_NULL, null=True, blank=True,
related_name='profile_picture_users')
objects = ToolshedUserManager()
class Meta:

View file

@ -1,4 +1,5 @@
import json
import base64
from django.test import Client, RequestFactory
from nacl.encoding import HexEncoder
@ -6,6 +7,7 @@ from nacl.signing import SigningKey
from authentication.models import ToolshedUser, KnownIdentity
from authentication.tests import UserTestMixin, SignatureAuthClient, DummyExternalUser, ToolshedTestCase
from files.models import File
class AuthorizationTestCase(ToolshedTestCase):
@ -240,6 +242,7 @@ class UserApiTestCase(UserTestMixin, ToolshedTestCase):
self.assertEqual(reply.json()['username'], 'testuser1')
self.assertEqual(reply.json()['domain'], 'example.com')
self.assertEqual(reply.json()['email'], 'test1@abc.de')
self.assertIsNone(reply.json()['profile_picture'])
def test_user_info2(self):
target = "/auth/user/"
@ -249,6 +252,50 @@ class UserApiTestCase(UserTestMixin, ToolshedTestCase):
self.assertEqual(reply.status_code, 200)
self.assertEqual(reply.json()['username'], 'testuser1')
self.assertEqual(reply.json()['domain'], 'example.com')
self.assertIsNone(reply.json()['profile_picture'])
def test_user_info_patch_profile_picture(self):
content = base64.b64encode(b'user-profile-image').decode('utf-8')
reply = self.client.patch('/auth/user/', self.f['local_user1'], {
'profile_picture': {
'data': content,
'mime_type': 'image/png'
}
})
self.assertEqual(reply.status_code, 200)
self.assertTrue(reply.json()['profile_picture'])
self.assertEqual(reply.json()['profile_picture']['mime_type'], 'image/png')
self.assertEqual(File.objects.count(), 1)
self.f['local_user1'].refresh_from_db()
self.assertIsNotNone(self.f['local_user1'].profile_picture)
def test_user_info_patch_profile_picture_clear(self):
encoded_content = base64.b64encode(b'user-profile-image').decode('utf-8')
test_file = File.objects.create(mime_type='image/png', data=encoded_content)
self.f['local_user1'].profile_picture = test_file
self.f['local_user1'].save()
reply = self.client.patch('/auth/user/', self.f['local_user1'], {'profile_picture': None})
self.assertEqual(reply.status_code, 200)
self.assertIsNone(reply.json()['profile_picture'])
self.f['local_user1'].refresh_from_db()
self.assertIsNone(self.f['local_user1'].profile_picture)
self.assertFalse(File.objects.filter(id=test_file.id).exists())
def test_user_info_patch_profile_picture_invalid(self):
reply = self.client.patch('/auth/user/', self.f['local_user1'], {'profile_picture': 'invalid'})
self.assertEqual(reply.status_code, 400)
def test_user_info_patch_profile_picture_id(self):
encoded_content = base64.b64encode(b'user-profile-image-by-id').decode('utf-8')
test_file = File.objects.create(mime_type='image/jpeg', data=encoded_content)
reply = self.client.patch('/auth/user/', self.f['local_user1'], {'profile_picture_id': test_file.id})
self.assertEqual(reply.status_code, 200)
self.assertEqual(reply.json()['profile_picture']['id'], test_file.id)
def test_user_info_patch_profile_picture_id_not_found(self):
reply = self.client.patch('/auth/user/', self.f['local_user1'], {'profile_picture_id': 999999})
self.assertEqual(reply.status_code, 400)
def test_user_info_fail(self):
reply = self.anonymous_client.get('/auth/user/')

View file

@ -1,5 +1,7 @@
from django.http import HttpResponse
from django.urls import path
from django.db.models import Q
from django.conf import settings
from drf_yasg.utils import swagger_auto_schema
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes, authentication_classes
@ -7,7 +9,6 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from authentication.signature_auth import SignatureAuthentication
from backend import settings
from files.models import File
@ -17,7 +18,10 @@ from files.models import File
@authentication_classes([SignatureAuthentication])
def media_urls(request, hash_path):
try:
file = File.objects.filter(connected_items__owner__in=request.user.friends_or_self()).distinct().get(
file = File.objects.filter(
Q(connected_items__owner__in=request.user.friends_or_self()) |
Q(profile_picture_users__in=request.user.friends_or_self())
).distinct().get(
file=hash_path)
if settings.SERVE_X_ACCEL_REDIRECT:

View file

@ -1,7 +1,7 @@
from django.core.files.base import ContentFile
from django.core.files.storage import DefaultStorage
from django.db import IntegrityError, transaction
from django.test import Client
from django.test import Client, override_settings
from authentication.tests import SignatureAuthClient, ToolshedTestCase, UserTestMixin
from toolshed.tests import InventoryTestMixin
from nacl.hash import sha256
@ -165,3 +165,33 @@ class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, Toolsh
self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)
self.assertTrue('X-Accel-Redirect' not in reply.headers)
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
def test_profile_picture_url(self):
self.f['local_user1'].profile_picture = self.f['test_file3']
self.f['local_user1'].save()
reply = client.get(
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
def test_profile_picture_url_friend(self):
self.f['local_user1'].profile_picture = self.f['test_file3']
self.f['local_user1'].save()
reply = client.get(
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
self.f['local_user2'])
self.assertEqual(reply.status_code, 200)
def test_profile_picture_url_not_friend(self):
self.f['local_user1'].profile_picture = self.f['test_file3']
self.f['local_user1'].save()
reply = client.get(
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)

View file

@ -0,0 +1,19 @@
# Generated by Django 4.2.2 on 2026-07-23 02:02
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('toolshed', '0007_workflowinstance'),
]
operations = [
migrations.AlterField(
model_name='inventoryitem',
name='storage_location',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='inventory_items', to='toolshed.storagelocation'),
),
]

View file

@ -90,7 +90,7 @@ class InventoryItem(SoftDeleteModel):
tags = models.ManyToManyField(Tag, through='ItemTag', related_name='inventory_items')
properties = models.ManyToManyField(Property, through='ItemProperty')
files = models.ManyToManyField(File, related_name='connected_items')
storage_location = models.ForeignKey('StorageLocation', on_delete=models.CASCADE, null=True, blank=True,
storage_location = models.ForeignKey('StorageLocation', on_delete=models.SET_NULL, null=True, blank=True,
related_name='inventory_items')
def clean(self):

View file

@ -1,6 +1,6 @@
from authentication.tests import SignatureAuthClient, UserTestMixin, ToolshedTestCase
from files.tests import FilesTestMixin
from toolshed.models import InventoryItem, Category
from toolshed.models import InventoryItem, Category, StorageLocation
from toolshed.tests import InventoryTestMixin, LocationTestMixin
client = SignatureAuthClient()
@ -69,3 +69,63 @@ class LocationApiTestCase(UserTestMixin, InventoryTestMixin, LocationTestMixin,
self.assertEqual(reply.json()[3]['description'], None)
self.assertEqual(reply.json()[3]['category'], 'cat1')
self.assertEqual(reply.json()[3]['path'], 'loc1/loc4')
def test_post_new_location(self):
reply = client.post('/api/storage_locations/', self.f['local_user1'], {
'name': 'loc5',
'description': 'a new location',
})
self.assertEqual(reply.status_code, 201)
self.assertEqual(StorageLocation.objects.count(), 5)
location = StorageLocation.objects.get(name='loc5')
self.assertEqual(location.description, 'a new location')
self.assertEqual(location.owner, self.f['local_user1'])
self.assertEqual(location.parent, None)
self.assertEqual(reply.json()['path'], 'loc5')
def test_post_new_nested_location(self):
reply = client.post('/api/storage_locations/', self.f['local_user1'], {
'name': 'loc5',
'parent': self.f['loc3'].id,
})
self.assertEqual(reply.status_code, 201)
location = StorageLocation.objects.get(name='loc5')
self.assertEqual(location.parent, self.f['loc3'])
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,
})
self.assertEqual(reply.status_code, 200)
location = StorageLocation.objects.get(id=self.f['loc2'].id)
self.assertEqual(location.name, 'loc2-renamed')
self.assertEqual(location.parent, self.f['loc1'])
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'])
self.assertEqual(reply.status_code, 204)
self.assertEqual(StorageLocation.objects.count(), 3)
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc4'].id).count(), 0)
def test_delete_location_with_items_sets_null(self):
item = InventoryItem.objects.create(
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'])
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'])
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)
self.assertEqual(StorageLocation.objects.filter(id=self.f['loc1'].id).count(), 1)