feat: new modul (stateflow)

stateflow/tests.py

@@ -1,3 +1,158 @@
 from django.test import TestCase
+from django.contrib.auth import get_user_model
+from . import models, services
 
-# Create your tests here.
+User = get_user_model()
+
+
+class OrderStateFlowTestCase(TestCase):
+    """测试订单状态流转逻辑"""
+
+    def setUp(self):
+        """设置测试数据"""
+        # 创建测试用户
+        self.user = User.objects.create_user(username='testuser', password='testpass')
+        
+        # 创建状态节点
+        self.state1 = models.State.objects.create(name='状态1', description='第一个状态')
+        self.state2 = models.State.objects.create(name='状态2', description='第二个状态')
+        self.state3 = models.State.objects.create(name='状态3', description='第三个状态')
+        
+        # 创建流程
+        self.process = models.Process.objects.create(name='测试流程', description='用于测试的流程')
+        
+        # 添加流程节点
+        models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0)
+        models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1)
+        models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2)
+        
+        # 创建订单（业务对象）
+        from django.contrib.contenttypes.models import ContentType
+        # 使用 Process 作为临时的关联对象（实际使用时应该关联真实的业务对象）
+        ct = ContentType.objects.get_for_model(models.Process)
+        self.business_object = models.BusinessObject.objects.create(
+            name='测试订单',
+            process=self.process,
+            description='测试订单描述',
+            content_type=ct,
+            object_id=self.process.id
+        )
+
+    def test_initial_state(self):
+        """测试初始状态 - 应该是 None（未开始）"""
+        current_state = self.business_object.get_current_state()
+        self.assertIsNone(current_state, '初始状态应该是 None（未开始）')
+        self.assertEqual(self.business_object.get_progress_percentage(), 0.0)
+        
+        # 整体状态应该是 not_started
+        status = services.get_overall_status(self.business_object)
+        self.assertEqual(status, 'not_started')
+
+    def test_advance_to_next_state(self):
+        """测试推进到下一个状态"""
+        # 第一次推进应该完成第一个状态
+        success, message = services.advance_to_next_state(self.business_object, self.user)
+        self.assertTrue(success)
+        
+        current_state = self.business_object.get_current_state()
+        self.assertEqual(current_state.id, self.state2.id)
+        self.assertAlmostEqual(self.business_object.get_progress_percentage(), 33.33, places=1)
+        
+        # 推进到状态3
+        success, message = services.advance_to_next_state(self.business_object, self.user)
+        self.assertTrue(success)
+        
+        current_state = self.business_object.get_current_state()
+        self.assertEqual(current_state.id, self.state3.id)
+        self.assertAlmostEqual(self.business_object.get_progress_percentage(), 66.67, places=1)
+        
+        # 完成最后一个状态
+        success, message = services.advance_to_next_state(self.business_object, self.user)
+        self.assertTrue(success)
+        self.assertEqual(self.business_object.get_progress_percentage(), 100.0)
+        
+        # 应该返回 None（所有状态都已完成）
+        current_state = self.business_object.get_current_state()
+        self.assertIsNone(current_state)
+        
+        # 整体状态应该是 completed
+        status = services.get_overall_status(self.business_object)
+        self.assertEqual(status, 'completed')
+
+    def test_state_status(self):
+        """测试状态的状态"""
+        # 初始状态，所有状态都应该是 not_started
+        self.assertEqual(services.get_business_object_state_status(self.business_object, self.state1), 'not_started')
+        self.assertEqual(services.get_business_object_state_status(self.business_object, self.state2), 'not_started')
+        
+        # 完成第一个状态
+        services.advance_to_next_state(self.business_object, self.user)
+        self.assertEqual(services.get_business_object_state_status(self.business_object, self.state1), 'completed')
+        self.assertEqual(services.get_business_object_state_status(self.business_object, self.state2), 'in_progress')
+        self.assertEqual(services.get_business_object_state_status(self.business_object, self.state3), 'not_started')
+
+    def test_timeline(self):
+        """测试时间线"""
+        timeline = self.business_object.get_timeline()
+        self.assertEqual(len(timeline), 3)
+        
+        # 所有状态都应该是 not_started
+        for item in timeline:
+            self.assertEqual(item['status'], 'not_started')
+        
+        # 完成第一个状态
+        services.advance_to_next_state(self.business_object, self.user)
+        
+        timeline = self.business_object.get_timeline()
+        self.assertEqual(timeline[0]['status'], 'completed')
+        self.assertEqual(timeline[1]['status'], 'in_progress')
+        self.assertEqual(timeline[2]['status'], 'not_started')
+        self.assertIsNotNone(timeline[0]['completed_by'])
+        self.assertEqual(timeline[0]['completed_by'].id, self.user.id)
+        self.assertFalse(timeline[0]['is_cancelled'])
+
+    def test_reset_progress(self):
+        """测试重置进度 - 不删除记录，而是标记为已撤销"""
+        # 完成两个状态
+        services.advance_to_next_state(self.business_object, self.user)
+        services.advance_to_next_state(self.business_object, self.user)
+        
+        # 确认有2条日志记录
+        self.assertEqual(self.business_object.state_logs.count(), 2)
+        
+        # 重置进度
+        services.reset_order_progress(self.business_object)
+        
+        # 日志记录应该还在（不删除）
+        self.assertEqual(self.business_object.state_logs.count(), 2)
+        
+        # 但所有记录都应该标记为已撤销
+        cancelled_count = self.business_object.state_logs.filter(is_cancelled=True).count()
+        self.assertEqual(cancelled_count, 2)
+        
+        # 应该回到初始状态（未开始）
+        current_state = self.business_object.get_current_state()
+        self.assertIsNone(current_state)
+        
+        # 所有撤销的记录都应该有撤销时间
+        for log in self.business_object.state_logs.all():
+            self.assertTrue(log.is_cancelled)
+            self.assertIsNotNone(log.cancelled_at)
+    
+    def test_timeline_with_cancelled(self):
+        """测试包含撤销记录的时间线"""
+        # 推进并完成所有状态
+        services.advance_to_next_state(self.business_object, self.user)
+        services.advance_to_next_state(self.business_object, self.user)
+        services.advance_to_next_state(self.business_object, self.user)
+        
+        # 重置进度
+        services.reset_order_progress(self.business_object)
+        
+        timeline = self.business_object.get_timeline()
+        
+        # 所有状态都应该显示为 cancelled
+        for item in timeline:
+            self.assertEqual(item['status'], 'cancelled')
+            self.assertTrue(item['is_cancelled'])
+            self.assertIsNotNone(item['cancelled_at'])