85 lines
5.4 KiB
Python
85 lines
5.4 KiB
Python
import json
|
|
import uuid
|
|
from datetime import timedelta
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
from django.contrib.auth.models import User
|
|
from django.db import close_old_connections, IntegrityError, transaction
|
|
from django.test import TestCase, TransactionTestCase, Client, skipUnlessDBFeature
|
|
from django.utils import timezone
|
|
from .models import Device, DeviceIdentity, ChoiceValue, InventoryAPIKey, NumberSequence
|
|
|
|
class IdentityTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user('identity-operator', is_superuser=True)
|
|
_, self.key = InventoryAPIKey.objects.create_key(name='identity', service_user=self.user,
|
|
can_read=True, can_write=True, expiry_date=timezone.now()+timedelta(days=1))
|
|
self.status = ChoiceValue.objects.create(kind='status', label='Eingang')
|
|
self.kind = ChoiceValue.objects.create(kind='device_type', label='Laptop')
|
|
self.payload = {'hwid': 'dmi-uuid:'+str(uuid.uuid4()), 'device_kind': 'Laptop', 'model': 'Test Laptop'}
|
|
def post(self, data=None, key=None):
|
|
return self.client.post('/api/v1/devices/resolve/', json.dumps(data or self.payload), content_type='application/json', HTTP_AUTHORIZATION='Api-Key '+(key or self.key))
|
|
def test_repeat_uses_same_number_and_preserves_repairs(self):
|
|
a=self.post(); self.assertEqual(a.status_code,201,a.content)
|
|
device=Device.objects.get(); device.comment='Repaired'; device.save()
|
|
b=self.post(); self.assertEqual(b.status_code,200,b.content)
|
|
self.assertEqual(a.json()['number'],b.json()['number'])
|
|
self.assertEqual(Device.objects.get().comment,'Repaired')
|
|
self.assertEqual(DeviceIdentity.objects.count(),1)
|
|
def test_existing_raw_uuid_is_reused(self):
|
|
d=Device.objects.create(status=self.status,device_type=self.kind,hwid=self.payload['hwid'][9:],model='Test Laptop')
|
|
r=self.post(); self.assertEqual(r.status_code,200,r.content); self.assertEqual(r.json()['number'],d.number)
|
|
def test_invalid_identifier_and_model_conflict(self):
|
|
for hwid in ['','unknown','dmi-uuid:00000000-0000-0000-0000-000000000000','dmi-uuid:ffffffff-ffff-ffff-ffff-ffffffffffff']:
|
|
self.assertEqual(self.post({**self.payload,'hwid':hwid}).status_code,400)
|
|
self.post(); self.assertEqual(self.post({**self.payload,'model':'Different'}).status_code,409)
|
|
self.assertEqual(Device.objects.count(),1)
|
|
def test_existing_ambiguous_or_archived_is_not_duplicated(self):
|
|
d=Device.objects.create(status=self.status,device_type=self.kind,hwid=self.payload['hwid'],archived_at=timezone.now())
|
|
self.assertEqual(self.post().status_code,409)
|
|
Device.objects.create(status=self.status,device_type=self.kind,hwid=self.payload['hwid'])
|
|
self.assertEqual(self.post().status_code,409)
|
|
self.assertEqual(Device.objects.count(),2)
|
|
def test_readonly_cannot_register(self):
|
|
_,key=InventoryAPIKey.objects.create_key(name='read',service_user=self.user,can_read=True,can_write=False,expiry_date=timezone.now()+timedelta(days=1))
|
|
self.assertEqual(self.post(key=key).status_code,403)
|
|
self.assertFalse(Device.objects.exists())
|
|
def test_serial_fallback_then_uuid_does_not_duplicate(self):
|
|
serial='dmi-serial-sha256:'+'a'*64
|
|
first=self.post({**self.payload,'hwid':serial})
|
|
second=self.post({**self.payload,'aliases':[serial]})
|
|
self.assertEqual(first.status_code,201,first.content)
|
|
self.assertEqual(second.status_code,200,second.content)
|
|
self.assertEqual(first.json()['number'],second.json()['number'])
|
|
self.assertEqual(DeviceIdentity.objects.count(),2)
|
|
conflict=self.post({**self.payload,'aliases':['dmi-serial-sha256:'+'b'*64]})
|
|
self.assertEqual(conflict.status_code,409,conflict.content)
|
|
self.assertEqual(Device.objects.count(),1)
|
|
|
|
def test_database_enforces_unique_identity(self):
|
|
self.post()
|
|
with self.assertRaises(IntegrityError), transaction.atomic():
|
|
DeviceIdentity.objects.create(hwid=self.payload['hwid'],device=Device.objects.get())
|
|
|
|
class ConcurrentIdentityTests(TransactionTestCase):
|
|
@skipUnlessDBFeature('has_select_for_update')
|
|
def test_two_clients_allocate_once(self):
|
|
user=User.objects.create_user('concurrent-identity',is_superuser=True)
|
|
_,key=InventoryAPIKey.objects.create_key(name='identity',service_user=user,can_read=True,can_write=True,expiry_date=timezone.now()+timedelta(days=1))
|
|
ChoiceValue.objects.create(kind='status',label='Eingang')
|
|
ChoiceValue.objects.create(kind='device_type',label='Laptop')
|
|
NumberSequence.objects.update_or_create(name='device',defaults={'last_value':0})
|
|
payload={'hwid':'dmi-uuid:'+str(uuid.uuid4()),'device_kind':'Laptop'}
|
|
barrier=Barrier(2)
|
|
def submit(_):
|
|
close_old_connections()
|
|
try:
|
|
barrier.wait(timeout=10)
|
|
r=Client().post('/api/v1/devices/resolve/',json.dumps(payload),content_type='application/json',HTTP_AUTHORIZATION='Api-Key '+key)
|
|
return r.status_code,r.json()['number']
|
|
finally: close_old_connections()
|
|
with ThreadPoolExecutor(max_workers=2) as pool: results=list(pool.map(submit,range(2)))
|
|
self.assertEqual(sorted(x[0] for x in results),[200,201])
|
|
self.assertEqual(len({x[1] for x in results}),1)
|
|
self.assertEqual(Device.objects.count(),1)
|