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)