116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
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])
|