c3lf/c3lf-system-3
9
0
Fork 0
Code Issues 26 Pull requests Releases Wiki Activity

create inventory API v2 und update tests

Browse source
This commit is contained in:
j3d1 2024-01-07 21:13:44 +01:00
parent 77828295f8
commit fc05c7c1d8
12 changed files with 486 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
core/core/urls.py
View file

@ -24,6 +24,7 @@ urlpatterns = [
path('api/1/', include('inventory.api_v1')),
path('api/1/', include('files.api_v1')),
path('api/1/', include('files.media_v1')),
path('api/2/', include('inventory.api_v2')),
path('api/2/', include('files.api_v2')),
path('media/2/', include('files.media_v2')),
path('api/2/', include('authentication.api_v2')),

181
core/inventory/api_v2.py Normal file
View file

@ -0,0 +1,181 @@
from datetime import datetime
from django.urls import path, re_path
from django.contrib.auth.decorators import permission_required
from rest_framework import routers, viewsets, serializers
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from files.models import File
from inventory.models import Event, Container, Item
class EventSerializer(serializers.ModelSerializer):
class Meta:
model = Event
fields = ['eid', 'slug', 'name', 'start', 'end', 'pre_start', 'post_end']
read_only_fields = ['eid']
class EventViewSet(viewsets.ModelViewSet):
serializer_class = EventSerializer
queryset = Event.objects.all()
permission_classes = []
class ContainerSerializer(serializers.ModelSerializer):
itemCount = serializers.SerializerMethodField()
class Meta:
model = Container
fields = ['cid', 'name', 'itemCount']
read_only_fields = ['cid', 'itemCount']
def get_itemCount(self, instance):
return Item.objects.filter(container=instance.cid).count()
class ContainerViewSet(viewsets.ModelViewSet):
serializer_class = ContainerSerializer
queryset = Container.objects.all()
class ItemSerializer(serializers.ModelSerializer):
dataImage = serializers.CharField(write_only=True, required=False)
cid = serializers.SerializerMethodField()
box = serializers.SerializerMethodField()
file = serializers.SerializerMethodField()
class Meta:
model = Item
fields = ['cid', 'box', 'uid', 'description', 'file', 'dataImage']
read_only_fields = ['uid']
def get_cid(self, instance):
return instance.container.cid
def get_box(self, instance):
return instance.container.name
def get_file(self, instance):
if len(instance.files.all()) > 0:
return instance.files.all().order_by('-created_at')[0].hash
return None
def to_internal_value(self, data):
if 'cid' in data:
container = Container.objects.get(cid=data['cid'])
internal = super().to_internal_value(data)
internal['container'] = container
return internal
return super().to_internal_value(data)
def validate(self, attrs):
return super().validate(attrs)
def create(self, validated_data):
if 'dataImage' in validated_data:
file = File.objects.create(data=validated_data['dataImage'])
validated_data.pop('dataImage')
item = Item.objects.create(**validated_data)
item.files.set([file])
return item
return Item.objects.create(**validated_data)
def update(self, instance, validated_data):
if 'returned' in validated_data:
if validated_data['returned']:
validated_data['returned_at'] = datetime.now()
validated_data.pop('returned')
if 'dataImage' in validated_data:
file = File.objects.create(data=validated_data['dataImage'])
validated_data.pop('dataImage')
instance.files.add(file)
return super().update(instance, validated_data)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
@permission_required('view_item', raise_exception=True)
def search_items(request, event_slug, query):
try:
event = Event.objects.get(slug=event_slug)
query_tokens = query.split(' ')
q = Item.objects.filter(event=event)
for token in query_tokens:
if token:
q = q.filter(description__icontains=token)
return Response(ItemSerializer(q, many=True).data)
except Event.DoesNotExist:
return Response(status=404)
@api_view(['GET', 'POST'])
@permission_classes([IsAuthenticated])
def item(request, event_slug):
try:
event = Event.objects.get(slug=event_slug)
if request.method == 'GET':
if not request.user.has_event_perm(event, 'view_item'):
return Response(status=403)
return Response(ItemSerializer(Item.objects.filter(event=event), many=True).data)
elif request.method == 'POST':
if not request.user.has_event_perm(event, 'add_item'):
return Response(status=403)
validated_data = ItemSerializer(data=request.data)
if validated_data.is_valid():
validated_data.save(event=event)
return Response(validated_data.data, status=201)
except Event.DoesNotExist:
return Response(status=404)
@api_view(['GET', 'PUT', 'DELETE', 'PATCH'])
@permission_classes([IsAuthenticated])
def item_by_id(request, event_slug, id):
try:
event = Event.objects.get(slug=event_slug)
item = Item.objects.get(event=event, uid=id)
if request.method == 'GET':
if not request.user.has_event_perm(event, 'view_item'):
return Response(status=403)
return Response(ItemSerializer(item).data)
elif request.method == 'PUT':
if not request.user.has_event_perm(event, 'change_item'):
return Response(status=403)
validated_data = ItemSerializer(item, data=request.data)
if validated_data.is_valid():
validated_data.save()
return Response(validated_data.data)
return Response(validated_data.errors, status=400)
elif request.method == 'PATCH':
if not request.user.has_event_perm(event, 'change_item'):
return Response(status=403)
validated_data = ItemSerializer(item, data=request.data, partial=True)
if validated_data.is_valid():
validated_data.save()
return Response(validated_data.data)
return Response(validated_data.errors, status=400)
elif request.method == 'DELETE':
if not request.user.has_event_perm(event, 'delete_item'):
return Response(status=403)
item.delete()
return Response(status=204)
except Item.DoesNotExist:
return Response(status=404)
except Event.DoesNotExist:
return Response(status=404)
router = routers.SimpleRouter()
router.register(r'events', EventViewSet, basename='events')
router.register(r'boxes', ContainerViewSet, basename='boxes')
router.register(r'box', ContainerViewSet, basename='boxes')
urlpatterns = router.urls + [
path('<event_slug>/items/', item),
path('<event_slug>/items/<query>/', search_items),
path('<event_slug>/item/', item),
path('<event_slug>/item/<id>/', item_by_id),
]

0
core/inventory/tests/v1/__init__.py Normal file
View file

0
core/inventory/tests/test_api.py → core/inventory/tests/v1/test_api.py
View file

0
core/inventory/tests/test_containers.py → core/inventory/tests/v1/test_containers.py
View file

0
core/inventory/tests/test_events.py → core/inventory/tests/v1/test_events.py
View file

0
core/inventory/tests/test_items.py → core/inventory/tests/v1/test_items.py
View file

0
core/inventory/tests/v2/__init__.py Normal file
View file

44
core/inventory/tests/v2/test_api.py Normal file
View file

@ -0,0 +1,44 @@
from django.test import TestCase, Client
from django.contrib.auth.models import Permission
from knox.models import AuthToken
from authentication.models import ExtendedUser
class ApiTest(TestCase):
def setUp(self):
super().setUp()
self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
self.user.user_permissions.add(*Permission.objects.all())
self.user.save()
self.token = AuthToken.objects.create(user=self.user)
self.client = Client(headers={'Authorization': 'Token ' + self.token[1]})
def test_root(self):
from core.settings import SYSTEM3_VERSION
response = self.client.get('/api/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["framework_version"], SYSTEM3_VERSION)
def test_events(self):
response = self.client.get('/api/2/events/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])
def test_containers(self):
response = self.client.get('/api/2/boxes/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])
def test_files(self):
response = self.client.get('/api/2/files/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])
def test_items(self):
from inventory.models import Event
Event.objects.create(slug='TEST1', name='Event')
response = self.client.get('/api/2/TEST1/items/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])

66
core/inventory/tests/v2/test_containers.py Normal file
View file

@ -0,0 +1,66 @@
from django.test import TestCase, Client
from django.contrib.auth.models import Permission
from knox.models import AuthToken
from authentication.models import ExtendedUser
from inventory.models import Container
class ContainerTestCase(TestCase):
def setUp(self):
self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
self.user.user_permissions.add(*Permission.objects.all())
self.token = AuthToken.objects.create(user=self.user)
self.client = Client(headers={'Authorization': 'Token ' + self.token[1]})
def test_empty(self):
response = self.client.get('/api/2/boxes/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])
def test_members(self):
Container.objects.create(name='BOX')
response = self.client.get('/api/2/boxes/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 1)
self.assertEqual(response.json()[0]['cid'], 1)
self.assertEqual(response.json()[0]['name'], 'BOX')
self.assertEqual(response.json()[0]['itemCount'], 0)
def test_multi_members(self):
Container.objects.create(name='BOX 1')
Container.objects.create(name='BOX 2')
Container.objects.create(name='BOX 3')
response = self.client.get('/api/2/boxes/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 3)
def test_create_container(self):
response = self.client.post('/api/2/box/', {'name': 'BOX'})
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()['cid'], 1)
self.assertEqual(response.json()['name'], 'BOX')
self.assertEqual(response.json()['itemCount'], 0)
self.assertEqual(len(Container.objects.all()), 1)
self.assertEqual(Container.objects.all()[0].cid, 1)
self.assertEqual(Container.objects.all()[0].name, 'BOX')
def test_update_container(self):
box = Container.objects.create(name='BOX 1')
response = self.client.put(f'/api/2/box/{box.cid}/', {'name': 'BOX 2'}, content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['cid'], 1)
self.assertEqual(response.json()['name'], 'BOX 2')
self.assertEqual(response.json()['itemCount'], 0)
self.assertEqual(len(Container.objects.all()), 1)
self.assertEqual(Container.objects.all()[0].cid, 1)
self.assertEqual(Container.objects.all()[0].name, 'BOX 2')
def test_delete_container(self):
box = Container.objects.create(name='BOX 1')
Container.objects.create(name='BOX 2')
self.assertEqual(len(Container.objects.all()), 2)
response = self.client.delete(f'/api/2/box/{box.cid}/')
self.assertEqual(response.status_code, 204)
self.assertEqual(len(Container.objects.all()), 1)

56
core/inventory/tests/v2/test_events.py Normal file
View file

@ -0,0 +1,56 @@
from django.test import TestCase, Client
from inventory.models import Event
client = Client()
class EventTestCase(TestCase):
def test_empty(self):
response = client.get('/api/2/events/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), [])
def test_members(self):
Event.objects.create(slug='EVENT', name='Event')
response = client.get('/api/2/events/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 1)
self.assertEqual(response.json()[0]['slug'], 'EVENT')
self.assertEqual(response.json()[0]['name'], 'Event')
def test_multi_members(self):
Event.objects.create(slug='EVENT1', name='Event 1')
Event.objects.create(slug='EVENT2', name='Event 2')
Event.objects.create(slug='EVENT3', name='Event 3')
response = client.get('/api/2/events/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 3)
def test_create_event(self):
response = client.post('/api/2/events/', {'slug': 'EVENT', 'name': 'Event'})
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()['slug'], 'EVENT')
self.assertEqual(response.json()['name'], 'Event')
self.assertEqual(len(Event.objects.all()), 1)
self.assertEqual(Event.objects.all()[0].slug, 'EVENT')
self.assertEqual(Event.objects.all()[0].name, 'Event')
def test_update_event(self):
from rest_framework.test import APIClient
event = Event.objects.create(slug='EVENT1', name='Event 1')
response = APIClient().put(f'/api/2/events/{event.eid}/', {'slug': 'EVENT2', 'name': 'Event 2 new'})
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['slug'], 'EVENT2')
self.assertEqual(response.json()['name'], 'Event 2 new')
self.assertEqual(len(Event.objects.all()), 1)
self.assertEqual(Event.objects.all()[0].slug, 'EVENT2')
self.assertEqual(Event.objects.all()[0].name, 'Event 2 new')
def test_remove_event(self):
event = Event.objects.create(slug='EVENT1', name='Event 1')
Event.objects.create(slug='EVENT2', name='Event 2')
self.assertEqual(len(Event.objects.all()), 2)
response = client.delete(f'/api/2/events/{event.eid}/')
self.assertEqual(response.status_code, 204)
self.assertEqual(len(Event.objects.all()), 1)

138
core/inventory/tests/v2/test_items.py Normal file
View file

@ -0,0 +1,138 @@
from django.test import TestCase, Client
from django.contrib.auth.models import Permission
from knox.models import AuthToken
from authentication.models import ExtendedUser
from files.models import File
from inventory.models import Event, Container, Item
class ItemTestCase(TestCase):
def setUp(self):
super().setUp()
self.event = Event.objects.create(slug='EVENT', name='Event')
self.box = Container.objects.create(name='BOX')
self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
self.user.user_permissions.add(*Permission.objects.all())
self.token = AuthToken.objects.create(user=self.user)
self.client = Client(headers={'Authorization': 'Token ' + self.token[1]})
def test_empty(self):
response = self.client.get(f'/api/2/{self.event.slug}/item/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content, b'[]')
def test_members(self):
item = Item.objects.create(container=self.box, event=self.event, description='1')
response = self.client.get(f'/api/2/{self.event.slug}/item/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(),
[{'uid': 1, 'description': '1', 'box': 'BOX', 'cid': self.box.cid, 'file': None}])
def test_members_with_file(self):
import base64
item = Item.objects.create(container=self.box, event=self.event, description='1')
file = File.objects.create(item=item, data="data:text/plain;base64," + base64.b64encode(b"foo").decode('utf-8'))
response = self.client.get(f'/api/2/{self.event.slug}/item/')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(),
[{'uid': 1, 'description': '1', 'box': 'BOX', 'cid': self.box.cid, 'file': file.hash}])
def test_multi_members(self):
Item.objects.create(container=self.box, event=self.event, description='1')
Item.objects.create(container=self.box, event=self.event, description='2')
Item.objects.create(container=self.box, event=self.event, description='3')
response = self.client.get(f'/api/2/{self.event.slug}/item/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 3)
def test_create_item(self):
response = self.client.post(f'/api/2/{self.event.slug}/item/', {'cid': self.box.cid, 'description': '1'})
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json(),
{'uid': 1, 'description': '1', 'box': 'BOX', 'cid': self.box.cid, 'file': None})
self.assertEqual(len(Item.objects.all()), 1)
self.assertEqual(Item.objects.all()[0].uid, 1)
self.assertEqual(Item.objects.all()[0].description, '1')
self.assertEqual(Item.objects.all()[0].container.cid, self.box.cid)
def test_create_item_with_file(self):
import base64
response = self.client.post(f'/api/2/{self.event.slug}/item/',
{'cid': self.box.cid, 'description': '1',
'dataImage': "data:text/plain;base64," + base64.b64encode(b"foo").decode(
'utf-8')}, content_type='application/json')
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()['uid'], 1)
self.assertEqual(response.json()['description'], '1')
self.assertEqual(response.json()['box'], 'BOX')
self.assertEqual(response.json()['cid'], self.box.cid)
self.assertEqual(len(response.json()['file']), 64)
self.assertEqual(len(Item.objects.all()), 1)
self.assertEqual(Item.objects.all()[0].uid, 1)
self.assertEqual(Item.objects.all()[0].description, '1')
self.assertEqual(Item.objects.all()[0].container.cid, self.box.cid)
self.assertEqual(len(File.objects.all()), 1)
def test_update_item(self):
item = Item.objects.create(container=self.box, event=self.event, description='1')
response = self.client.put(f'/api/2/{self.event.slug}/item/{item.uid}/', {'description': '2'},
content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(),
{'uid': 1, 'description': '2', 'box': 'BOX', 'cid': self.box.cid, 'file': None})
self.assertEqual(len(Item.objects.all()), 1)
self.assertEqual(Item.objects.all()[0].uid, 1)
self.assertEqual(Item.objects.all()[0].description, '2')
self.assertEqual(Item.objects.all()[0].container.cid, self.box.cid)
def test_update_item_with_file(self):
import base64
item = Item.objects.create(container=self.box, event=self.event, description='1')
response = self.client.put(f'/api/2/{self.event.slug}/item/{item.uid}/',
{'description': '2',
'dataImage': "data:text/plain;base64," + base64.b64encode(b"foo").decode('utf-8')},
content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['uid'], 1)
self.assertEqual(response.json()['description'], '2')
self.assertEqual(response.json()['box'], 'BOX')
self.assertEqual(response.json()['cid'], self.box.cid)
self.assertEqual(len(response.json()['file']), 64)
self.assertEqual(len(Item.objects.all()), 1)
self.assertEqual(Item.objects.all()[0].uid, 1)
self.assertEqual(Item.objects.all()[0].description, '2')
self.assertEqual(Item.objects.all()[0].container.cid, self.box.cid)
self.assertEqual(len(File.objects.all()), 1)
def test_delete_item(self):
item = Item.objects.create(container=self.box, event=self.event, description='1')
Item.objects.create(container=self.box, event=self.event, description='2')
self.assertEqual(len(Item.objects.all()), 2)
response = self.client.delete(f'/api/2/{self.event.slug}/item/{item.uid}/')
self.assertEqual(response.status_code, 204)
self.assertEqual(len(Item.objects.all()), 1)
def test_delete_item2(self):
Item.objects.create(container=self.box, event=self.event, description='1')
item2 = Item.objects.create(container=self.box, event=self.event, description='2')
self.assertEqual(len(Item.objects.all()), 2)
response = self.client.delete(f'/api/2/{self.event.slug}/item/{item2.uid}/')
self.assertEqual(response.status_code, 204)
self.assertEqual(len(Item.objects.all()), 1)
item3 = Item.objects.create(container=self.box, event=self.event, description='3')
self.assertEqual(item3.uid, 3)
self.assertEqual(len(Item.objects.all()), 2)
def test_item_count(self):
Item.objects.create(container=self.box, event=self.event, description='1')
Item.objects.create(container=self.box, event=self.event, description='2')
response = self.client.get('/api/2/boxes/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()), 1)
self.assertEqual(response.json()[0]['itemCount'], 2)
def test_item_nonexistent(self):
response = self.client.get(f'/api/2/NOEVENT/item/')
self.assertEqual(response.status_code, 404)