121 lines
6.2 KiB
Python
121 lines
6.2 KiB
Python
import json
|
|
import uuid
|
|
from datetime import timedelta
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.db import close_old_connections
|
|
from django.test import Client, TestCase, TransactionTestCase, skipUnlessDBFeature
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from .models import AuditLog, ChoiceValue, Device, DeviceRegistration, InventoryAPIKey, NumberSequence
|
|
|
|
|
|
class RegistrationTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user('usb-operator', is_superuser=True)
|
|
self.status = ChoiceValue.objects.create(kind='status', label='Neu')
|
|
self.kind = ChoiceValue.objects.create(kind='device_type', label='Laptop')
|
|
_, self.key = InventoryAPIKey.objects.create_key(
|
|
name='usb-intake', service_user=self.user, can_read=True, can_write=True,
|
|
expiry_date=timezone.now() + timedelta(days=1))
|
|
self.payload = {'request_id': str(uuid.uuid4()), 'status': self.status.pk,
|
|
'device_type': self.kind.pk, 'cpu': 'Test CPU', 'ram_gb': 16,
|
|
'hwid': str(uuid.uuid4()), 'comment': 'CPU/RAM: PASS'}
|
|
|
|
def post(self, payload=None, key=None):
|
|
return self.client.post(reverse('inventory-api:device-list'),
|
|
json.dumps(self.payload if payload is None else payload),
|
|
content_type='application/json',
|
|
HTTP_AUTHORIZATION=f'Api-Key {key or self.key}')
|
|
|
|
def test_create_specs_audit_and_retry(self):
|
|
NumberSequence.objects.update_or_create(name='device', defaults={'last_value': 42})
|
|
response = self.post()
|
|
self.assertEqual(response.status_code, 201, response.content)
|
|
device = Device.objects.get()
|
|
self.assertEqual(device.number, 43)
|
|
self.assertEqual(device.ram_gb, 16)
|
|
self.assertEqual(device.stage, Device.Stage.INTAKE)
|
|
self.assertIsNone(device.ssd_gb)
|
|
self.assertEqual(device.wifi, 'UNKNOWN')
|
|
audit = AuditLog.objects.get(device=device)
|
|
self.assertEqual(audit.api_client, 'usb-intake')
|
|
self.assertEqual(audit.user, self.user)
|
|
self.assertEqual(audit.action, AuditLog.Action.CREATE)
|
|
retry = self.post()
|
|
self.assertEqual(retry.status_code, 200)
|
|
self.assertEqual(retry.json()['number'], device.number)
|
|
self.assertEqual(Device.objects.count(), 1)
|
|
self.assertEqual(AuditLog.objects.count(), 1)
|
|
|
|
def test_changed_retry_conflicts(self):
|
|
self.post()
|
|
self.assertEqual(self.post({**self.payload, 'ram_gb': 32}).status_code, 409)
|
|
self.assertEqual(Device.objects.get().ram_gb, 16)
|
|
|
|
def test_rejects_missing_auth_readonly_and_viewer(self):
|
|
self.assertEqual(self.client.post(reverse('inventory-api:device-list'), self.payload).status_code, 401)
|
|
_, key = InventoryAPIKey.objects.create_key(name='read', service_user=self.user, can_read=True,
|
|
expiry_date=timezone.now() + timedelta(days=1))
|
|
self.assertEqual(self.post(key=key).status_code, 403)
|
|
self.user.is_superuser = False
|
|
self.user.save()
|
|
self.user.profile.role = 'VIEWER'
|
|
self.user.profile.save()
|
|
self.assertEqual(self.post().status_code, 403)
|
|
self.assertFalse(Device.objects.exists())
|
|
|
|
def test_strict_validation_and_active_choices(self):
|
|
for changes in ({'status': self.kind.pk}, {'status': None}, {'device_type': None},
|
|
{'ram_gb': -1}, {'number': 99}, {'stage': 'STORAGE'}, {'request_id': 'bad'}):
|
|
with self.subTest(changes=changes):
|
|
self.assertEqual(self.post({**self.payload, **changes}).status_code, 400)
|
|
for missing in ('status', 'device_type', 'request_id'):
|
|
self.assertEqual(self.post({k: v for k, v in self.payload.items() if k != missing}).status_code, 400)
|
|
self.kind.active = False
|
|
self.kind.save()
|
|
self.assertEqual(self.post().status_code, 400)
|
|
self.assertFalse(Device.objects.exists())
|
|
|
|
def test_failure_rolls_back_device_receipt_and_number(self):
|
|
NumberSequence.objects.update_or_create(name='device', defaults={'last_value': 42})
|
|
with patch('inventory.api.add_audit', side_effect=RuntimeError('audit unavailable')):
|
|
with self.assertRaises(RuntimeError):
|
|
self.post()
|
|
self.assertFalse(Device.objects.exists())
|
|
self.assertFalse(DeviceRegistration.objects.exists())
|
|
self.assertEqual(NumberSequence.objects.get(name='device').last_value, 42)
|
|
|
|
|
|
class ConcurrentRegistrationTests(TransactionTestCase):
|
|
@skipUnlessDBFeature('has_select_for_update')
|
|
def test_simultaneous_retry_allocates_once(self):
|
|
user = User.objects.create_user('concurrent-usb', is_superuser=True)
|
|
_, key = InventoryAPIKey.objects.create_key(name='usb', service_user=user, can_read=True, can_write=True,
|
|
expiry_date=timezone.now() + timedelta(days=1))
|
|
status = ChoiceValue.objects.create(kind='status', label='Neu')
|
|
kind = ChoiceValue.objects.create(kind='device_type', label='Desktop')
|
|
NumberSequence.objects.update_or_create(name='device', defaults={'last_value': 0})
|
|
payload = {'request_id': str(uuid.uuid4()), 'status': status.pk, 'device_type': kind.pk}
|
|
barrier = Barrier(2)
|
|
|
|
def submit(_):
|
|
close_old_connections()
|
|
try:
|
|
barrier.wait(timeout=10)
|
|
response = Client().post(reverse('inventory-api:device-list'), json.dumps(payload),
|
|
content_type='application/json', HTTP_AUTHORIZATION=f'Api-Key {key}')
|
|
return response.status_code, response.json()['number']
|
|
finally:
|
|
close_old_connections()
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(submit, range(2)))
|
|
self.assertEqual(sorted(code for code, _ in results), [200, 201])
|
|
self.assertEqual(len({number for _, number in results}), 1)
|
|
self.assertEqual(Device.objects.count(), 1)
|