from django.test import TestCase from django.contrib.auth import get_user_model from . import models, services 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_business_object_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_business_object_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'])