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, ) def test_serializer_exposes_nullable_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 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.assertIsNone(no_defect_data["defect_grade"]) self.assertIsNone(no_defect_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, ) 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_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, ) validated = 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, }) self.assertTrue(validated["clear_defect"]) self.assertEqual(validated["count"], 2) 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, "批次追踪物料仅同缺陷等级可清除批次缺陷", ): 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.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") def test_restore_replaces_current_defects_exactly( self, 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)], ) @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])