855 lines
29 KiB
Python
855 lines
29 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from decimal import Decimal
|
|
from threading import Barrier
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.db import connection, connections, transaction
|
|
from django.test import SimpleTestCase, TestCase, TransactionTestCase
|
|
from unittest import skipUnless
|
|
|
|
from apps.mtm.models import Material, Mgroup, WmScope
|
|
from apps.qm.models import Defect
|
|
from apps.system.models import Dept, User
|
|
from apps.wpm.filters import HandoverFilter, WMaterialFilter
|
|
from apps.wpm.models import Handover, WMaterial
|
|
from apps.wpm.serializers import (
|
|
HandoverSerializer,
|
|
WMaterialCreateSerializer,
|
|
WMaterialSerializer,
|
|
)
|
|
from apps.wpm.services import restore_mlog_wpr_defects, snapshot_mlog_wpr_defects
|
|
from apps.wpm.views import MlogbwViewSet
|
|
from apps.wpmw.models import Wpr, WprDefect
|
|
from rest_framework.exceptions import ParseError
|
|
|
|
|
|
class HandoverFilterTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.dept = Dept.objects.create(name="交接筛选测试车间")
|
|
cls.other_dept = Dept.objects.create(name="其他测试车间")
|
|
cls.mgroup = Mgroup.objects.create(
|
|
name="交接筛选测试工段",
|
|
belong_dept=cls.dept,
|
|
)
|
|
cls.peer_mgroup = Mgroup.objects.create(
|
|
name="同车间其他工段",
|
|
belong_dept=cls.dept,
|
|
)
|
|
cls.other_mgroup = Mgroup.objects.create(
|
|
name="其他车间工段",
|
|
belong_dept=cls.other_dept,
|
|
)
|
|
cls.user = User.objects.create_user(username="handover-filter-user")
|
|
cls.material = Material.objects.create(name="交接筛选测试物料")
|
|
|
|
@classmethod
|
|
def create_handover(cls, **kwargs):
|
|
defaults = {
|
|
"send_date": "2026-07-28",
|
|
"send_user": cls.user,
|
|
"material": cls.material,
|
|
}
|
|
defaults.update(kwargs)
|
|
return Handover.objects.create(**defaults)
|
|
|
|
def test_mgroup_only_returns_records_involving_the_selected_mgroup(self):
|
|
sent = self.create_handover(
|
|
send_mgroup=self.mgroup,
|
|
recive_mgroup=self.peer_mgroup,
|
|
)
|
|
received = self.create_handover(
|
|
send_mgroup=self.peer_mgroup,
|
|
recive_mgroup=self.mgroup,
|
|
)
|
|
self.create_handover(
|
|
send_mgroup=self.peer_mgroup,
|
|
recive_mgroup=self.other_mgroup,
|
|
)
|
|
|
|
queryset = HandoverFilter(
|
|
{"mgroup": self.mgroup.id},
|
|
queryset=Handover.objects.all(),
|
|
).qs
|
|
|
|
self.assertCountEqual(queryset.values_list("id", flat=True), [
|
|
sent.id,
|
|
received.id,
|
|
])
|
|
|
|
def test_mgroupx_adds_only_department_level_records_for_same_dept(self):
|
|
direct = self.create_handover(send_mgroup=self.mgroup)
|
|
dept_level = self.create_handover(
|
|
send_dept=self.dept,
|
|
send_mgroup=None,
|
|
)
|
|
self.create_handover(
|
|
send_mgroup=self.peer_mgroup,
|
|
recive_mgroup=self.other_mgroup,
|
|
)
|
|
self.create_handover(
|
|
send_dept=self.other_dept,
|
|
send_mgroup=None,
|
|
)
|
|
|
|
queryset = HandoverFilter(
|
|
{"mgroupx": self.mgroup.id},
|
|
queryset=Handover.objects.all(),
|
|
).qs
|
|
|
|
self.assertCountEqual(queryset.values_list("id", flat=True), [
|
|
direct.id,
|
|
dept_level.id,
|
|
])
|
|
|
|
|
|
class WMaterialDefectGradeTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.material = Material.objects.create(name="缺陷等级测试物料")
|
|
cls.defect_b = Defect.objects.create(
|
|
name="B类缺陷",
|
|
cate=Defect.cate_list[0],
|
|
okcate=Defect.DEFECT_OK_B,
|
|
)
|
|
cls.defect_notok = Defect.objects.create(
|
|
name="不合格缺陷",
|
|
cate=Defect.cate_list[0],
|
|
okcate=Defect.DEFECT_NOTOK,
|
|
)
|
|
cls.normal_with_notok_defect = WMaterial.objects.create(
|
|
material=cls.material,
|
|
batch="NORMAL-NOTOK",
|
|
count=1,
|
|
state=WMaterial.WM_OK,
|
|
defect=cls.defect_notok,
|
|
)
|
|
cls.notok_with_b_defect = WMaterial.objects.create(
|
|
material=cls.material,
|
|
batch="NOTOK-B",
|
|
count=1,
|
|
state=WMaterial.WM_NOTOK,
|
|
defect=cls.defect_b,
|
|
)
|
|
cls.normal_without_defect = WMaterial.objects.create(
|
|
material=cls.material,
|
|
batch="NORMAL-NONE",
|
|
count=1,
|
|
state=WMaterial.WM_OK,
|
|
)
|
|
cls.repair_without_defect = WMaterial.objects.create(
|
|
material=cls.material,
|
|
batch="REPAIR-NONE",
|
|
count=1,
|
|
state=WMaterial.WM_REPAIR,
|
|
)
|
|
cls.notok_with_legacy_sign = WMaterial.objects.create(
|
|
material=cls.material,
|
|
batch="NOTOK-LEGACY",
|
|
count=1,
|
|
state=WMaterial.WM_NOTOK,
|
|
notok_sign="zw",
|
|
)
|
|
|
|
def test_serializer_exposes_effective_defect_grade_without_using_state(self):
|
|
normal_notok_data = WMaterialSerializer(
|
|
self.normal_with_notok_defect
|
|
).data
|
|
notok_b_data = WMaterialSerializer(self.notok_with_b_defect).data
|
|
no_defect_data = WMaterialSerializer(self.normal_without_defect).data
|
|
repair_no_defect_data = WMaterialSerializer(
|
|
self.repair_without_defect
|
|
).data
|
|
legacy_data = WMaterialSerializer(self.notok_with_legacy_sign).data
|
|
|
|
self.assertEqual(
|
|
normal_notok_data["defect_grade"], Defect.DEFECT_NOTOK
|
|
)
|
|
self.assertEqual(normal_notok_data["defect_grade_name"], "不合格")
|
|
self.assertEqual(
|
|
notok_b_data["defect_grade"], Defect.DEFECT_OK_B
|
|
)
|
|
self.assertEqual(notok_b_data["defect_grade_name"], "合格B类")
|
|
self.assertEqual(no_defect_data["defect_grade"], Defect.DEFECT_OK)
|
|
self.assertEqual(no_defect_data["defect_grade_name"], "合格")
|
|
self.assertEqual(
|
|
repair_no_defect_data["defect_grade"], Defect.DEFECT_OK
|
|
)
|
|
self.assertEqual(repair_no_defect_data["defect_grade_name"], "合格")
|
|
self.assertEqual(legacy_data["defect_grade"], Defect.DEFECT_NOTOK)
|
|
self.assertEqual(legacy_data["defect_grade_name"], "不合格")
|
|
|
|
def test_filtering_state_and_defect_grade_are_independent(self):
|
|
normal_notok = WMaterialFilter(
|
|
{
|
|
"state": WMaterial.WM_OK,
|
|
"defect__okcate": Defect.DEFECT_NOTOK,
|
|
},
|
|
queryset=WMaterial.objects.all(),
|
|
).qs
|
|
notok_b = WMaterialFilter(
|
|
{
|
|
"state": WMaterial.WM_NOTOK,
|
|
"defect__okcate": Defect.DEFECT_OK_B,
|
|
},
|
|
queryset=WMaterial.objects.all(),
|
|
).qs
|
|
|
|
self.assertQuerySetEqual(
|
|
normal_notok,
|
|
[self.normal_with_notok_defect],
|
|
transform=lambda item: item,
|
|
)
|
|
self.assertQuerySetEqual(
|
|
notok_b,
|
|
[self.notok_with_b_defect],
|
|
transform=lambda item: item,
|
|
)
|
|
|
|
def test_effective_grade_filter_includes_defaults_and_legacy_signs(self):
|
|
ok_items = WMaterialFilter(
|
|
{"defect_grade": Defect.DEFECT_OK},
|
|
queryset=WMaterial.objects.all(),
|
|
).qs
|
|
notok_items = WMaterialFilter(
|
|
{"defect_grade": Defect.DEFECT_NOTOK},
|
|
queryset=WMaterial.objects.all(),
|
|
).qs
|
|
|
|
self.assertCountEqual(
|
|
ok_items.values_list("id", flat=True),
|
|
[self.normal_without_defect.id, self.repair_without_defect.id],
|
|
)
|
|
self.assertCountEqual(
|
|
notok_items.values_list("id", flat=True),
|
|
[
|
|
self.normal_with_notok_defect.id,
|
|
self.notok_with_legacy_sign.id,
|
|
],
|
|
)
|
|
|
|
|
|
class MlogbwViewSetTests(SimpleTestCase):
|
|
@patch("apps.wpm.views.MlogViewSet.lock_and_check_can_update")
|
|
@patch("apps.wpm.views.Mlogbw.cal_count_notok")
|
|
@patch("apps.wpm.views.Mlogbw.objects.get_or_create")
|
|
@patch("apps.wpm.views.Mlogb.objects.filter")
|
|
def test_perform_create_syncs_fix_output_without_route(
|
|
self,
|
|
mlogb_filter,
|
|
mlogbw_get_or_create,
|
|
cal_count_notok,
|
|
lock_and_check_can_update,
|
|
):
|
|
wm = object()
|
|
wpr = SimpleNamespace(wm=wm)
|
|
material = SimpleNamespace(tracking=Material.MA_TRACKING_SINGLE)
|
|
mlog = SimpleNamespace(
|
|
route=None,
|
|
is_fix=True,
|
|
cal_mlog_count_from_mlogb=MagicMock(),
|
|
)
|
|
mlogb_in = SimpleNamespace(
|
|
mlog=mlog,
|
|
route=None,
|
|
wm_in=wm,
|
|
material_in=material,
|
|
)
|
|
mlogb_out = object()
|
|
mlogb_qs = MagicMock()
|
|
mlogb_qs.exists.return_value = True
|
|
mlogb_qs.__iter__.return_value = iter([mlogb_out])
|
|
mlogb_filter.return_value = mlogb_qs
|
|
ins = SimpleNamespace(
|
|
mlogb=mlogb_in,
|
|
wpr=wpr,
|
|
number="2606P1095-3",
|
|
)
|
|
serializer = MagicMock()
|
|
serializer.save.return_value = ins
|
|
lock_and_check_can_update.return_value = mlog
|
|
|
|
MlogbwViewSet().perform_create(serializer)
|
|
|
|
mlogbw_get_or_create.assert_called_once_with(
|
|
mlogb=mlogb_out,
|
|
wpr=wpr,
|
|
defaults={
|
|
"number": "2606P1095-3",
|
|
"mlogbw_from": ins,
|
|
},
|
|
)
|
|
self.assertEqual(cal_count_notok.call_count, 2)
|
|
mlog.cal_mlog_count_from_mlogb.assert_called_once_with()
|
|
|
|
|
|
class WMaterialScopeTests(SimpleTestCase):
|
|
def test_inventory_key_normalizes_omitted_nullable_fields(self):
|
|
material = Material(id="100", name="测试物料")
|
|
|
|
omitted = WMaterial._normalize_inventory_lookup(
|
|
material=material,
|
|
batch="BATCH-001",
|
|
state=WMaterial.WM_OK,
|
|
)
|
|
explicit = WMaterial._normalize_inventory_lookup(
|
|
material=material,
|
|
batch="BATCH-001",
|
|
mgroup=None,
|
|
belong_dept=None,
|
|
state=WMaterial.WM_OK,
|
|
defect=None,
|
|
notok_sign=None,
|
|
material_origin=None,
|
|
state_origin=None,
|
|
)
|
|
|
|
self.assertEqual(omitted, explicit)
|
|
self.assertEqual(
|
|
WMaterial._inventory_advisory_lock_payload(omitted),
|
|
WMaterial._inventory_advisory_lock_payload(explicit),
|
|
)
|
|
|
|
def test_scope_resolver_returns_consistent_location_fields(self):
|
|
dept = object()
|
|
mgroup = SimpleNamespace(belong_dept=dept)
|
|
|
|
self.assertEqual(
|
|
WmScope.resolve_location(WmScope.MGROUP, mgroup),
|
|
{"mgroup": mgroup, "belong_dept": dept},
|
|
)
|
|
self.assertEqual(
|
|
WmScope.resolve_location(WmScope.DEPT, mgroup),
|
|
{"mgroup": None, "belong_dept": dept},
|
|
)
|
|
self.assertEqual(
|
|
WmScope.resolve_location(WmScope.GLOBAL, mgroup),
|
|
{"mgroup": None, "belong_dept": None},
|
|
)
|
|
|
|
def test_only_mgroup_scope_requires_handover_mgroup(self):
|
|
self.assertTrue(WmScope.requires_mgroup(WmScope.MGROUP))
|
|
self.assertFalse(WmScope.requires_mgroup(WmScope.DEPT))
|
|
self.assertFalse(WmScope.requires_mgroup(WmScope.GLOBAL))
|
|
|
|
def test_global_scope_key_does_not_access_missing_relations(self):
|
|
wm = WMaterial()
|
|
|
|
self.assertEqual(wm.belong_dept_or_mgroup_id, ("global", None))
|
|
|
|
def test_department_and_mgroup_scope_keys_cannot_collide(self):
|
|
dept_wm = WMaterial(belong_dept_id=10)
|
|
mgroup_wm = WMaterial(mgroup_id=10)
|
|
|
|
self.assertNotEqual(
|
|
dept_wm.belong_dept_or_mgroup_id,
|
|
mgroup_wm.belong_dept_or_mgroup_id,
|
|
)
|
|
|
|
def test_manual_create_allows_global_scope(self):
|
|
serializer = WMaterialCreateSerializer()
|
|
attrs = {
|
|
"material": Material(),
|
|
"count": 1,
|
|
"batch": "GLOBAL-001",
|
|
}
|
|
|
|
validated = serializer.validate(attrs)
|
|
|
|
self.assertFalse(serializer.fields["mgroup"].required)
|
|
self.assertFalse(serializer.fields["belong_dept"].required)
|
|
self.assertNotIn("mgroup", validated)
|
|
self.assertNotIn("belong_dept", validated)
|
|
|
|
def test_global_scope_can_be_filtered_explicitly(self):
|
|
self.assertIn(
|
|
"isnull",
|
|
WMaterialFilter.Meta.fields["belong_dept"],
|
|
)
|
|
self.assertIn(
|
|
"isnull",
|
|
WMaterialFilter.Meta.fields["mgroup"],
|
|
)
|
|
|
|
def test_manual_create_derives_department_from_mgroup(self):
|
|
dept = SimpleNamespace(id=20)
|
|
mgroup = SimpleNamespace(id=10, belong_dept=dept)
|
|
attrs = {
|
|
"material": Material(),
|
|
"count": 1,
|
|
"batch": "MGROUP-001",
|
|
"mgroup": mgroup,
|
|
}
|
|
|
|
validated = WMaterialCreateSerializer().validate(attrs)
|
|
|
|
self.assertIs(validated["belong_dept"], dept)
|
|
|
|
def test_global_inventory_cannot_enter_normal_handover(self):
|
|
wm = WMaterial(
|
|
material=Material(tracking=Material.MA_TRACKING_BATCH),
|
|
batch="GLOBAL-001",
|
|
count=1,
|
|
)
|
|
|
|
with self.assertRaisesMessage(ParseError, "全局库存无需正常交接"):
|
|
HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"count": 1,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_NORMAL,
|
|
})
|
|
|
|
def test_global_inventory_split_stays_global(self):
|
|
wm = WMaterial(
|
|
material=Material(tracking=Material.MA_TRACKING_BATCH),
|
|
batch="GLOBAL-001",
|
|
count=1,
|
|
)
|
|
|
|
validated = HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"count": 1,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_DIV,
|
|
})
|
|
|
|
self.assertIsNone(validated["send_dept"])
|
|
self.assertIsNone(validated["recive_dept"])
|
|
self.assertIsNone(validated["recive_mgroup"])
|
|
|
|
def test_global_inventory_cannot_merge_into_scoped_inventory(self):
|
|
material = Material(tracking=Material.MA_TRACKING_BATCH)
|
|
wm = WMaterial(material=material, batch="GLOBAL-001", count=1)
|
|
target = WMaterial(
|
|
material=material,
|
|
batch="GLOBAL-MERGED",
|
|
count=0,
|
|
belong_dept_id=20,
|
|
)
|
|
|
|
with self.assertRaisesMessage(ParseError, "全局库存合批目标必须是全局库存"):
|
|
HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"count": 1,
|
|
"new_wm": target,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
def test_global_inventory_scrap_requires_receiving_mgroup(self):
|
|
wm = WMaterial(
|
|
material=Material(tracking=Material.MA_TRACKING_BATCH),
|
|
batch="GLOBAL-001",
|
|
count=1,
|
|
)
|
|
|
|
with self.assertRaisesMessage(
|
|
ParseError,
|
|
"全局库存返修、报废或改版必须指定接收工段",
|
|
):
|
|
HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"count": 1,
|
|
"type": Handover.H_SCRAP,
|
|
"mtype": Handover.H_NORMAL,
|
|
})
|
|
|
|
def test_handover_keeps_explicit_sending_mgroup(self):
|
|
inventory_dept = Dept(id="20", name="库存车间")
|
|
sending_dept = Dept(id="30", name="送料车间")
|
|
mgroup = Mgroup(id="10", name="测试工段", belong_dept=sending_dept)
|
|
wm = WMaterial(
|
|
material=Material(tracking=Material.MA_TRACKING_BATCH),
|
|
batch="DEPT-001",
|
|
count=1,
|
|
belong_dept=inventory_dept,
|
|
)
|
|
|
|
validated = HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"count": 1,
|
|
"send_mgroup": mgroup,
|
|
"recive_dept": object(),
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_NORMAL,
|
|
})
|
|
|
|
self.assertIs(validated["send_mgroup"], mgroup)
|
|
self.assertIs(validated["send_dept"], sending_dept)
|
|
|
|
def test_batch_tracking_merge_can_clear_ok_b_batch_defects(self):
|
|
material = Material(tracking=Material.MA_TRACKING_BATCH)
|
|
defect_a = Defect(id="1", okcate=Defect.DEFECT_OK_B)
|
|
defect_b = Defect(id="2", okcate=Defect.DEFECT_OK_B)
|
|
wm_a = WMaterial(
|
|
id="10", material=material, batch="B-001", count=1,
|
|
state=WMaterial.WM_OK, defect=defect_a,
|
|
)
|
|
wm_b = WMaterial(
|
|
id="20", material=material, batch="B-002", count=1,
|
|
state=WMaterial.WM_OK, defect=defect_b,
|
|
)
|
|
|
|
validated = HandoverSerializer().validate({
|
|
"wm": wm_a,
|
|
"handoverb": [
|
|
{"wm": wm_a, "count": 1},
|
|
{"wm": wm_b, "count": 1},
|
|
],
|
|
"new_batch": "B-MERGED",
|
|
"clear_defect": True,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
self.assertTrue(validated["clear_defect"])
|
|
self.assertEqual(validated["count"], 2)
|
|
|
|
def test_batch_tracking_merge_can_clear_ok_and_ok_b_defects(self):
|
|
material = Material(tracking=Material.MA_TRACKING_BATCH)
|
|
defect_b = Defect(id="1", okcate=Defect.DEFECT_OK_B)
|
|
wm_ok = WMaterial(
|
|
id="10", material=material, batch="OK-001", count=1,
|
|
state=WMaterial.WM_OK, defect=None,
|
|
)
|
|
wm_b = WMaterial(
|
|
id="20", material=material, batch="B-001", count=1,
|
|
state=WMaterial.WM_OK, defect=defect_b,
|
|
)
|
|
|
|
validated = HandoverSerializer().validate({
|
|
"wm": wm_ok,
|
|
"handoverb": [
|
|
{"wm": wm_ok, "count": 1},
|
|
{"wm": wm_b, "count": 1},
|
|
],
|
|
"new_batch": "OK-MERGED",
|
|
"clear_defect": True,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
self.assertTrue(validated["clear_defect"])
|
|
self.assertEqual(validated["count"], 2)
|
|
|
|
def test_batch_tracking_merge_cannot_clear_same_grade_notok_defects(self):
|
|
material = Material(tracking=Material.MA_TRACKING_BATCH)
|
|
defect_a = Defect(id="1", okcate=Defect.DEFECT_NOTOK)
|
|
defect_b = Defect(id="2", okcate=Defect.DEFECT_NOTOK)
|
|
wm_a = WMaterial(
|
|
id="10", material=material, batch="N-001", count=1,
|
|
state=WMaterial.WM_NOTOK, defect=defect_a,
|
|
)
|
|
wm_b = WMaterial(
|
|
id="20", material=material, batch="N-002", count=1,
|
|
state=WMaterial.WM_NOTOK, defect=defect_b,
|
|
)
|
|
|
|
with self.assertRaisesMessage(
|
|
ParseError,
|
|
"批次追踪物料仅合格品和合格B类可清除批次缺陷",
|
|
):
|
|
HandoverSerializer().validate({
|
|
"wm": wm_a,
|
|
"handoverb": [
|
|
{"wm": wm_a, "count": 1},
|
|
{"wm": wm_b, "count": 1},
|
|
],
|
|
"new_batch": "N-MERGED",
|
|
"clear_defect": True,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
def test_batch_tracking_merge_cannot_clear_mixed_defect_grades(self):
|
|
material = Material(tracking=Material.MA_TRACKING_BATCH)
|
|
defect_b = Defect(id="1", okcate=Defect.DEFECT_OK_B)
|
|
defect_notok = Defect(id="2", okcate=Defect.DEFECT_NOTOK)
|
|
wm_a = WMaterial(
|
|
id="10", material=material, batch="B-001", count=1,
|
|
state=WMaterial.WM_OK, defect=defect_b,
|
|
)
|
|
wm_b = WMaterial(
|
|
id="20", material=material, batch="N-001", count=1,
|
|
state=WMaterial.WM_OK, defect=defect_notok,
|
|
)
|
|
|
|
with self.assertRaisesMessage(
|
|
ParseError,
|
|
"批次追踪物料仅合格品和合格B类可清除批次缺陷",
|
|
):
|
|
HandoverSerializer().validate({
|
|
"wm": wm_a,
|
|
"handoverb": [
|
|
{"wm": wm_a, "count": 1},
|
|
{"wm": wm_b, "count": 1},
|
|
],
|
|
"new_batch": "MIXED-MERGED",
|
|
"clear_defect": True,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
def test_single_tracking_merge_can_clear_notok_batch_defect(self):
|
|
material = Material(tracking=Material.MA_TRACKING_SINGLE)
|
|
defect = Defect(id="1", okcate=Defect.DEFECT_NOTOK)
|
|
wm = WMaterial(
|
|
id="10", material=material, batch="S-001", count=1,
|
|
state=WMaterial.WM_NOTOK, defect=defect,
|
|
)
|
|
wm_b = WMaterial(
|
|
id="20", material=material, batch="S-002", count=1,
|
|
state=WMaterial.WM_NOTOK, defect=defect,
|
|
)
|
|
|
|
validated = HandoverSerializer().validate({
|
|
"wm": wm,
|
|
"handoverb": [
|
|
{"wm": wm, "count": 1},
|
|
{"wm": wm_b, "count": 1},
|
|
],
|
|
"new_batch": "S-MERGED",
|
|
"clear_defect": True,
|
|
"type": Handover.H_NORMAL,
|
|
"mtype": Handover.H_MERGE,
|
|
})
|
|
|
|
self.assertTrue(validated["clear_defect"])
|
|
self.assertEqual(validated["count"], 2)
|
|
|
|
|
|
class WprDefectLifecycleTests(SimpleTestCase):
|
|
@patch("apps.wpmw.models.WprDefect.objects.bulk_create")
|
|
@patch("apps.wpmw.models.WprDefect.objects.filter")
|
|
@patch(
|
|
"apps.wpmw.models.idWorker.get_id",
|
|
side_effect=["100", "101"],
|
|
)
|
|
def test_replace_defects_assigns_unique_ids_before_bulk_create(
|
|
self,
|
|
get_id,
|
|
defect_filter,
|
|
bulk_create,
|
|
):
|
|
wpr = Wpr(id="1")
|
|
|
|
wpr.replace_defects([
|
|
{"defect_id": "10", "is_main": True},
|
|
{"defect_id": "11", "is_main": False},
|
|
])
|
|
|
|
defect_filter.return_value.delete.assert_called_once_with()
|
|
created = bulk_create.call_args.args[0]
|
|
self.assertEqual([str(item.id) for item in created], ["100", "101"])
|
|
self.assertEqual(get_id.call_count, 2)
|
|
|
|
@patch("apps.wpmw.models.WprDefect.objects.filter")
|
|
def test_change_inventory_never_changes_individual_defects(self, defect_filter):
|
|
material = Material(id="1", tracking=Material.MA_TRACKING_SINGLE)
|
|
old_wm = WMaterial(
|
|
id="10", material=material, state=WMaterial.WM_OK,
|
|
)
|
|
target_wm = WMaterial(
|
|
id="20", material=material, state=WMaterial.WM_OK,
|
|
defect=None,
|
|
)
|
|
wpr = SimpleNamespace(
|
|
wm=old_wm,
|
|
mb=None,
|
|
material=material,
|
|
state=WMaterial.WM_OK,
|
|
version=1,
|
|
save=MagicMock(),
|
|
)
|
|
|
|
Wpr.change_or_new(
|
|
wpr=wpr,
|
|
wm=target_wm,
|
|
old_wm=old_wm,
|
|
)
|
|
|
|
defect_filter.assert_not_called()
|
|
self.assertIs(wpr.wm, target_wm)
|
|
|
|
@patch.object(Wpr, "replace_defects")
|
|
@patch("apps.wpmw.models.FtestDefect.objects.filter")
|
|
@patch("apps.wpmw.models.FtestItem.objects.filter")
|
|
def test_apply_ftest_replaces_defects_and_carries_info(
|
|
self,
|
|
ftest_item_filter,
|
|
ftest_defect_filter,
|
|
replace_defects,
|
|
):
|
|
ftest = object()
|
|
ftest_defect_filter.return_value.values.return_value = [
|
|
{"defect_id": "30", "is_main": False},
|
|
{"defect_id": "31", "is_main": True},
|
|
]
|
|
ftest_item_filter.return_value.select_related.return_value = [
|
|
SimpleNamespace(
|
|
id="40",
|
|
test_val_json={"value": 1},
|
|
testitem=SimpleNamespace(name="尺寸"),
|
|
),
|
|
]
|
|
wpr = Wpr(
|
|
oinfo={"existing": {"value": 2}},
|
|
)
|
|
wpr.save = MagicMock()
|
|
|
|
wpr.apply_ftest(ftest)
|
|
|
|
ftest_defect_filter.assert_called_once_with(ftest=ftest, has=True)
|
|
replace_defects.assert_called_once_with([
|
|
{"defect_id": "30", "is_main": False},
|
|
{"defect_id": "31", "is_main": True},
|
|
])
|
|
self.assertEqual(
|
|
wpr.oinfo["40"],
|
|
{"name": "尺寸", "val": {"value": 1}},
|
|
)
|
|
|
|
@patch.object(Wpr, "replace_defects")
|
|
def test_apply_no_ftest_clears_defects(self, replace_defects):
|
|
wpr = Wpr(oinfo={"existing": True})
|
|
|
|
wpr.apply_ftest(None)
|
|
|
|
replace_defects.assert_called_once_with([])
|
|
self.assertEqual(wpr.oinfo, {"existing": True})
|
|
|
|
@patch("apps.wpmw.models.FtestItem.objects.filter")
|
|
def test_revert_ftest_info_only_removes_matching_keys(self, item_filter):
|
|
item_filter.return_value.values_list.return_value = ["10", "11"]
|
|
wpr = Wpr(oinfo={"10": {"v": 1}, "other": {"v": 2}})
|
|
wpr.save = MagicMock()
|
|
|
|
wpr.revert_ftest_info(object())
|
|
|
|
self.assertEqual(wpr.oinfo, {"other": {"v": 2}})
|
|
wpr.save.assert_called_once()
|
|
|
|
|
|
class MlogWprDefectSnapshotTests(SimpleTestCase):
|
|
@patch("apps.wpm.services.WprDefect.objects.filter")
|
|
@patch("apps.wpm.services.Wpr.objects.select_for_update")
|
|
@patch("apps.wpm.services.Mlogbw.objects.filter")
|
|
def test_snapshot_keeps_multiple_defects_and_empty_wpr(
|
|
self,
|
|
mlogbw_filter,
|
|
select_for_update,
|
|
defect_filter,
|
|
):
|
|
mlogbw_filter.return_value.values_list.return_value.distinct.return_value = [
|
|
"1",
|
|
"2",
|
|
]
|
|
select_for_update.return_value.filter.return_value.values_list.return_value = [
|
|
"1",
|
|
"2",
|
|
]
|
|
defect_filter.return_value.values.return_value = [
|
|
{"wpr_id": "1", "defect_id": "10", "is_main": True},
|
|
{"wpr_id": "1", "defect_id": "11", "is_main": False},
|
|
]
|
|
mlog = SimpleNamespace(id="100", save=MagicMock())
|
|
|
|
snapshot = snapshot_mlog_wpr_defects(mlog)
|
|
|
|
self.assertEqual(snapshot["2"], [])
|
|
self.assertEqual(len(snapshot["1"]), 2)
|
|
select_for_update.return_value.filter.return_value.values_list.assert_called_once_with(
|
|
"id",
|
|
flat=True,
|
|
)
|
|
|
|
@patch("apps.wpm.services.WprDefect.objects.bulk_create")
|
|
@patch("apps.wpm.services.WprDefect.objects.filter")
|
|
@patch("apps.wpm.services.Wpr.objects.select_for_update")
|
|
@patch(
|
|
"apps.wpm.services.idWorker.get_id",
|
|
side_effect=["100", "101"],
|
|
)
|
|
def test_restore_replaces_current_defects_exactly(
|
|
self,
|
|
get_id,
|
|
select_for_update,
|
|
defect_filter,
|
|
bulk_create,
|
|
):
|
|
wpr = Wpr(id="1")
|
|
select_for_update.return_value.filter.return_value = [wpr]
|
|
mlog = SimpleNamespace(
|
|
id="100",
|
|
wpr_defect_snapshot={
|
|
"1": [
|
|
{"defect_id": "10", "is_main": True},
|
|
{"defect_id": "11", "is_main": False},
|
|
],
|
|
},
|
|
)
|
|
|
|
restored = restore_mlog_wpr_defects(mlog)
|
|
|
|
self.assertTrue(restored)
|
|
defect_filter.return_value.delete.assert_called_once_with()
|
|
created = bulk_create.call_args.args[0]
|
|
self.assertEqual(
|
|
[(str(item.defect_id), item.is_main) for item in created],
|
|
[("10", True), ("11", False)],
|
|
)
|
|
self.assertEqual([str(item.id) for item in created], ["100", "101"])
|
|
self.assertEqual(get_id.call_count, 2)
|
|
|
|
|
|
@skipUnless(
|
|
connection.vendor == "postgresql",
|
|
"advisory lock concurrency test requires PostgreSQL",
|
|
)
|
|
class WMaterialConcurrencyTests(TransactionTestCase):
|
|
def test_concurrent_first_create_uses_one_inventory_record(self):
|
|
material = Material.objects.create(name="并发库存测试物料")
|
|
barrier = Barrier(2)
|
|
|
|
def create_inventory():
|
|
connections.close_all()
|
|
try:
|
|
barrier.wait()
|
|
with transaction.atomic():
|
|
wm, created = (
|
|
WMaterial.locked_get_or_create_inventory(
|
|
material=material,
|
|
batch="CONCURRENT-001",
|
|
state=WMaterial.WM_OK,
|
|
defaults={"count": Decimal("0")},
|
|
)
|
|
)
|
|
wm.count += Decimal("1")
|
|
wm.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 = WMaterial.objects.filter(
|
|
material=material,
|
|
batch="CONCURRENT-001",
|
|
mgroup=None,
|
|
belong_dept=None,
|
|
state=WMaterial.WM_OK,
|
|
defect=None,
|
|
notok_sign=None,
|
|
material_origin=None,
|
|
state_origin=None,
|
|
)
|
|
self.assertEqual(queryset.count(), 1)
|
|
self.assertEqual(queryset.get().count, Decimal("2"))
|
|
self.assertCountEqual(created_results, [True, False])
|