From e95e61f3ae03c17641ab028e6e42895da495c14b Mon Sep 17 00:00:00 2001 From: colaftc Date: Fri, 5 Dec 2025 22:52:16 +0800 Subject: [PATCH] fix: has_started was not exclude is_cancelled=True status node --- printing/models.py | 14 +- printing/test_has_started.py | 268 +++++++++++++++++++++++++++++++++++ 2 files changed, 276 insertions(+), 6 deletions(-) create mode 100644 printing/test_has_started.py diff --git a/printing/models.py b/printing/models.py index f9c4949..f2809c7 100644 --- a/printing/models.py +++ b/printing/models.py @@ -189,14 +189,15 @@ class PlateOrder(ModelBase): @property def has_started(self) -> bool: - """判断是否已开始""" + """判断是否已开始（有未撤销的状态记录）""" if not self.business_object: return False - # 检查是否有状态流转记录 + # 只检查未撤销的记录 from stateflow.models import StateFlowRecord has_records = StateFlowRecord.objects.filter( - business_object=self.business_object + business_object=self.business_object, + is_cancelled=False ).exists() return has_records @@ -351,14 +352,15 @@ class PrintingJob(ModelBase): @property def has_started(self) -> bool: - """判断是否已开始（有BusinessObject且有状态记录，或不在初始状态）""" + """判断是否已开始（有未撤销的状态记录）""" if not self.business_object: return False - # 检查是否有状态流转记录 + # 只检查未撤销的记录 from stateflow.models import StateFlowRecord has_records = StateFlowRecord.objects.filter( - business_object=self.business_object + business_object=self.business_object, + is_cancelled=False ).exists() return has_records diff --git a/printing/test_has_started.py b/printing/test_has_started.py new file mode 100644 index 0000000..9979767 --- /dev/null +++ b/printing/test_has_started.py @@ -0,0 +1,268 @@ +""" +测试 PlateOrder 和 PrintingJob 的 has_started 属性 +""" +from django.test import TestCase +from django.contrib.auth import get_user_model +from printing import models as printing_models +from basic_info import models as basic_info_models +from stateflow import models as stateflow_models +from stateflow import services as stateflow_services + +User = get_user_model() + + +class HasStartedTestCase(TestCase): + """测试 has_started 属性的各种情况""" + + def setUp(self): + """设置测试数据""" + self.user = User.objects.create_user(username='testuser', password='testpass') + + # 创建商户 + self.merchant = basic_info_models.Merchant.objects.create( + name='测试商户', + type=basic_info_models.MerchantTypeEnum.STORE + ) + + # 创建客户 + self.customer = basic_info_models.Customer.objects.create( + name='测试客户', + merchant=self.merchant, + mobile='13800138000' + ) + + # 创建状态和流程 + self.state1 = stateflow_models.State.objects.create(name='状态1') + self.state2 = stateflow_models.State.objects.create(name='状态2') + self.state3 = stateflow_models.State.objects.create(name='状态3') + + self.process = stateflow_models.Process.objects.create(name='测试流程') + stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) + stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) + stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) + + def test_plate_order_has_started_without_business_object(self): + """测试没有 business_object 时 has_started 为 False""" + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST001', + customer=self.customer + ) + + # 没有 business_object，has_started 应该是 False + self.assertFalse(plate_order.has_started) + + def test_plate_order_has_started_with_business_object_no_records(self): + """测试有 business_object 但没有状态流转记录时 has_started 为 False""" + # 手动创建 business_object + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST002', + customer=self.customer, + business_object=business_object + ) + + # 有 business_object 但没有状态流转记录，has_started 应该是 False + self.assertFalse(plate_order.has_started) + + # 验证确实没有记录 + self.assertEqual(business_object.state_logs.count(), 0) + + def test_plate_order_has_started_after_advance(self): + """测试推进状态后 has_started 为 True""" + # 手动创建 business_object + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST003', + customer=self.customer, + business_object=business_object + ) + + # 推进一次 + stateflow_services.advance_to_next_state(business_object, self.user) + + # 现在应该有记录了 + self.assertEqual(business_object.state_logs.count(), 1) + + # has_started 应该是 True + self.assertTrue(plate_order.has_started) + + def test_plate_order_has_started_after_step_back(self): + """测试回退后 has_started 变为 False（因为只有已撤销的记录）""" + # 手动创建 business_object + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST004', + customer=self.customer, + business_object=business_object + ) + + # 推进一次 + stateflow_services.advance_to_next_state(business_object, self.user) + self.assertTrue(plate_order.has_started) + + # 回退 + stateflow_services.step_back_one_state(business_object, self.user) + + # 验证有已撤销的记录 + self.assertEqual(business_object.state_logs.count(), 1) + self.assertTrue(business_object.state_logs.first().is_cancelled) + + # has_started 应该变为 False（因为只检查未撤销的记录） + self.assertFalse(plate_order.has_started) + + def test_plate_order_has_started_after_reset(self): + """测试重置后 has_started 变为 False（因为记录都被标记为已撤销）""" + # 手动创建 business_object + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST005', + customer=self.customer, + business_object=business_object + ) + + # 推进两次 + stateflow_services.advance_to_next_state(business_object, self.user) + stateflow_services.advance_to_next_state(business_object, self.user) + self.assertTrue(plate_order.has_started) + + # 重置 + stateflow_services.reset_business_object_progress(business_object) + + # 验证记录还在但都被撤销了 + self.assertEqual(business_object.state_logs.count(), 2) + self.assertTrue(all(log.is_cancelled for log in business_object.state_logs.all())) + + # has_started 应该变为 False（因为没有未撤销的记录） + self.assertFalse(plate_order.has_started) + + def test_plate_order_serializer_includes_has_started(self): + """测试序列化器包含 has_started 字段""" + # 手动创建 business_object + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + plate_order = printing_models.PlateOrder.objects.create( + design_code='TEST200', + customer=self.customer, + business_object=business_object + ) + + # 验证 has_started 是一个属性 + self.assertFalse(plate_order.has_started) + + # 推进后 + stateflow_services.advance_to_next_state(business_object, self.user) + + # 重新获取以确保属性被重新计算 + plate_order.refresh_from_db() + self.assertTrue(plate_order.has_started) + + def test_printing_order_can_change_process(self): + """测试 PrintingOrder.can_change_process() 使用正确的 has_started 逻辑""" + # 创建 PrintingOrder + printing_order = printing_models.PrintingOrder.objects.create( + customer=self.customer, + fabric='测试布料', + width='150cm' + ) + + # 创建 Product 和 Category + category = basic_info_models.ProductCategory.objects.create( + name='测试分类', + merchant=self.merchant + ) + product = basic_info_models.Product.objects.create( + name='测试产品', + merchant=self.merchant, + category=category + ) + + # 创建 PrintingJob + business_object = stateflow_models.BusinessObject.objects.create( + name='测试业务对象', + process=self.process + ) + + printing_job = printing_models.PrintingJob.objects.create( + printing_order=printing_order, + product=product, + quantity=100, + unit='米', + business_object=business_object + ) + + # 未推进时，可以修改流程 + self.assertTrue(printing_order.can_change_process()) + self.assertFalse(printing_job.has_started) + + # 推进后，不能修改流程 + stateflow_services.advance_to_next_state(business_object, self.user) + self.assertFalse(printing_order.can_change_process()) + self.assertTrue(printing_job.has_started) + + # 回退后，又可以修改流程了（因为 has_started 变为 False） + stateflow_services.step_back_one_state(business_object, self.user) + self.assertTrue(printing_order.can_change_process()) + self.assertFalse(printing_job.has_started) + + def test_has_started_query_performance(self): + """测试 has_started 的查询逻辑（排除已撤销记录）""" + # 创建多个 PlateOrder + business_object1 = stateflow_models.BusinessObject.objects.create( + name='测试业务对象1', + process=self.process + ) + business_object2 = stateflow_models.BusinessObject.objects.create( + name='测试业务对象2', + process=self.process + ) + + plate_order1 = printing_models.PlateOrder.objects.create( + design_code='TEST101', + customer=self.customer, + business_object=business_object1 + ) + plate_order2 = printing_models.PlateOrder.objects.create( + design_code='TEST102', + customer=self.customer, + business_object=business_object2 + ) + + # 只推进第一个 + stateflow_services.advance_to_next_state(business_object1, self.user) + + # 验证 + self.assertTrue(plate_order1.has_started) + self.assertFalse(plate_order2.has_started) + + # 推进第二个 + stateflow_services.advance_to_next_state(business_object2, self.user) + + # 现在都应该是 True + self.assertTrue(plate_order1.has_started) + self.assertTrue(plate_order2.has_started) + + # 回退第一个 + stateflow_services.step_back_one_state(business_object1, self.user) + + # 第一个应该变为 False，第二个保持 True + self.assertFalse(plate_order1.has_started) + self.assertTrue(plate_order2.has_started)