from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal
from threading import Barrier
from unittest import skipUnless

from django.db import connection, connections, transaction
from django.test import SimpleTestCase, TransactionTestCase

from apps.inm.models import MaterialBatch, WareHouse
from apps.mtm.models import Material


class MaterialBatchInventoryKeyTests(SimpleTestCase):
    def setUp(self):
        self.material = Material(id='100', name='测试物料')
        self.warehouse = WareHouse(
            id='200',
            number='TEST',
            name='测试仓库',
            place='测试地点',
        )

    def test_inventory_key_normalizes_omitted_optional_fields(self):
        omitted = MaterialBatch._normalize_inventory_lookup(
            material=self.material,
            batch='BATCH-001',
            warehouse=self.warehouse,
        )
        explicit = MaterialBatch._normalize_inventory_lookup(
            material=self.material,
            batch='BATCH-001',
            warehouse=self.warehouse,
            state=10,
            defect=None,
        )

        self.assertEqual(omitted, explicit)
        self.assertEqual(
            MaterialBatch._inventory_advisory_lock_payload(omitted),
            MaterialBatch._inventory_advisory_lock_payload(explicit),
        )

    def test_inventory_key_requires_material_batch_and_warehouse(self):
        required_values = {
            'material': self.material,
            'batch': 'BATCH-001',
            'warehouse': self.warehouse,
        }

        for field in required_values:
            with self.subTest(field=field):
                lookup = required_values.copy()
                lookup[field] = None
                with self.assertRaisesRegex(ValueError, field):
                    MaterialBatch._normalize_inventory_lookup(**lookup)

    def test_inventory_key_rejects_unknown_fields(self):
        with self.assertRaisesRegex(TypeError, 'supplier'):
            MaterialBatch._normalize_inventory_lookup(
                material=self.material,
                batch='BATCH-001',
                warehouse=self.warehouse,
                supplier=None,
            )


@skipUnless(
    connection.vendor == 'postgresql',
    'advisory lock concurrency test requires PostgreSQL',
)
class MaterialBatchConcurrencyTests(TransactionTestCase):
    def test_concurrent_first_create_uses_one_inventory_record(self):
        material = Material.objects.create(name='仓库并发测试物料')
        warehouse = WareHouse.objects.create(
            number='CONCURRENT',
            name='并发测试仓库',
            place='测试地点',
        )
        barrier = Barrier(2)

        def create_inventory():
            connections.close_all()
            try:
                barrier.wait()
                with transaction.atomic():
                    mb, created = (
                        MaterialBatch.locked_get_or_create_inventory(
                            material=material,
                            batch='CONCURRENT-001',
                            warehouse=warehouse,
                            defaults={'count': Decimal('0')},
                        )
                    )
                    mb.count += Decimal('1')
                    mb.save(update_fields=['count'])
                    return created
            finally:
                connections.close_all()

        with ThreadPoolExecutor(max_workers=2) as executor:
            created_results = list(executor.map(
                lambda _: create_inventory(),
                range(2),
            ))

        queryset = MaterialBatch.objects.filter(
            material=material,
            batch='CONCURRENT-001',
            warehouse=warehouse,
            state=10,
            defect=None,
        )
        self.assertEqual(queryset.count(), 1)
        self.assertEqual(queryset.get().count, Decimal('2'))
        self.assertCountEqual(created_results, [True, False])