""" BusinessObject 克隆服务测试 重点验证： - 克隆后新对象与源对象业务数据一致（process/描述/关联对象/日志/参数） - 时间戳（completed_at/cancelled_at/参数记录 created_at 等）保持一致 - 包含已撤销记录 - 克隆对象与原对象互不影响 """ import datetime from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from django.test import TestCase from django.utils import timezone from stateflow import models, services User = get_user_model() class CloneBusinessObjectServiceTestCase(TestCase): def setUp(self): self.user1 = User.objects.create_user(username='u1', password='pass') self.user2 = User.objects.create_user(username='u2', password='pass') self.state1 = models.State.objects.create(name='状态1', description='第一个节点') self.state2 = models.State.objects.create(name='状态2', description='第二个节点') self.process = models.Process.objects.create(name='流程A', 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) ct = ContentType.objects.get_for_model(models.Process) self.business_object = models.BusinessObject.objects.create( name='BO-1', process=self.process, description='BO 描述', content_type=ct, object_id=self.process.id, ) # 固定时间点 tz = timezone.get_default_timezone() self.t1 = timezone.make_aware(datetime.datetime(2025, 12, 1, 10, 0, 0), tz) self.t1a = timezone.make_aware(datetime.datetime(2025, 12, 1, 10, 1, 0), tz) self.t1b = timezone.make_aware(datetime.datetime(2025, 12, 1, 10, 2, 0), tz) self.t2 = timezone.make_aware(datetime.datetime(2025, 12, 2, 9, 0, 0), tz) self.t2a = timezone.make_aware(datetime.datetime(2025, 12, 2, 9, 1, 0), tz) self.t2_cancel = timezone.make_aware(datetime.datetime(2025, 12, 2, 9, 30, 0), tz) # 创建两条状态日志：一条正常，一条已撤销 log1 = models.StateFlowRecord.objects.create( business_object=self.business_object, state=self.state1, completed_by=self.user1, is_cancelled=False, ) models.StateFlowRecord.objects.filter(id=log1.id).update( completed_at=self.t1, created_at=self.t1, updated_at=self.t1, ) rec1 = models.StateLogParameterRecord.objects.create( state_log=log1, parameters={'temperature': '25.5'}, remark='初次测量', ) models.StateLogParameterRecord.objects.filter(id=rec1.id).update( created_at=self.t1a, updated_at=self.t1a, ) rec2 = models.StateLogParameterRecord.objects.create( state_log=log1, parameters={'temperature': '26.0', 'operator': '张三'}, remark='复测', ) models.StateLogParameterRecord.objects.filter(id=rec2.id).update( created_at=self.t1b, updated_at=self.t1b, ) log2 = models.StateFlowRecord.objects.create( business_object=self.business_object, state=self.state2, completed_by=self.user2, is_cancelled=True, cancelled_at=self.t2_cancel, ) models.StateFlowRecord.objects.filter(id=log2.id).update( completed_at=self.t2, created_at=self.t2, updated_at=self.t2, cancelled_at=self.t2_cancel, ) rec3 = models.StateLogParameterRecord.objects.create( state_log=log2, parameters={'humidity': '60%'}, remark='记录湿度', ) models.StateLogParameterRecord.objects.filter(id=rec3.id).update( created_at=self.t2a, updated_at=self.t2a, ) # 重新加载，确保测试数据时间戳生效 self.business_object.refresh_from_db() def test_clone_business_object_copies_all_records_and_parameters(self): new_object_id = (self.business_object.object_id or 0) + 1000 cloned = services.clone_business_object( self.business_object, new_object_id=new_object_id, expected_content_type_id=self.business_object.content_type_id, ) # 基本字段一致 self.assertNotEqual(cloned.id, self.business_object.id) self.assertEqual(cloned.name, self.business_object.name) self.assertEqual(cloned.description, self.business_object.description) self.assertEqual(cloned.process_id, self.business_object.process_id) self.assertEqual(cloned.content_type_id, self.business_object.content_type_id) self.assertEqual(cloned.object_id, new_object_id) self.assertNotEqual(cloned.object_id, self.business_object.object_id) # ModelBase 时间戳字段同步（严格一致） self.assertEqual(cloned.created_at, self.business_object.created_at) self.assertEqual(cloned.updated_at, self.business_object.updated_at) src_logs = list(self.business_object.state_logs.all().order_by('completed_at', 'id')) new_logs = list(cloned.state_logs.all().order_by('completed_at', 'id')) self.assertEqual(len(new_logs), len(src_logs)) for src, new in zip(src_logs, new_logs): self.assertNotEqual(src.id, new.id) self.assertEqual(new.state_id, src.state_id) self.assertEqual(new.completed_by_id, src.completed_by_id) self.assertEqual(new.is_cancelled, src.is_cancelled) self.assertEqual(new.cancelled_at, src.cancelled_at) self.assertEqual(new.completed_at, src.completed_at) self.assertEqual(new.created_at, src.created_at) self.assertEqual(new.updated_at, src.updated_at) src_params = list(src.parameter_records.all().order_by('created_at', 'id')) new_params = list(new.parameter_records.all().order_by('created_at', 'id')) self.assertEqual(len(new_params), len(src_params)) for src_rec, new_rec in zip(src_params, new_params): self.assertNotEqual(src_rec.id, new_rec.id) self.assertEqual(new_rec.parameters, src_rec.parameters) self.assertEqual(new_rec.remark, src_rec.remark) self.assertEqual(new_rec.created_at, src_rec.created_at) self.assertEqual(new_rec.updated_at, src_rec.updated_at) # 已撤销记录的参数默认不可见（克隆也应一致） src_cancelled = src_logs[1] new_cancelled = new_logs[1] self.assertTrue(src_cancelled.is_cancelled) self.assertTrue(new_cancelled.is_cancelled) self.assertEqual(src_cancelled.get_all_parameters_summary(), {}) self.assertEqual(new_cancelled.get_all_parameters_summary(), {}) self.assertEqual( new_cancelled.get_all_parameters_summary(include_cancelled=True), src_cancelled.get_all_parameters_summary(include_cancelled=True), ) def test_clone_is_independent_from_source(self): cloned = services.clone_business_object( self.business_object, new_object_id=(self.business_object.object_id or 0) + 1000, expected_content_type_id=self.business_object.content_type_id, ) # 修改克隆对象的参数记录，不应影响源对象 cloned_first_log = cloned.state_logs.order_by('completed_at', 'id').first() cloned_first_param = cloned_first_log.parameter_records.order_by('created_at', 'id').first() cloned_first_param.parameters['temperature'] = '99.9' cloned_first_param.save(update_fields=['parameters']) src_first_log = self.business_object.state_logs.order_by('completed_at', 'id').first() src_first_param = src_first_log.parameter_records.order_by('created_at', 'id').first() self.assertEqual(src_first_param.parameters['temperature'], '25.5') def test_clone_empty_business_object(self): bo = models.BusinessObject.objects.create( name='BO-empty', process=self.process, description='empty', ) cloned = services.clone_business_object( bo, new_object_id=None, expected_content_type_id=None, ) self.assertNotEqual(cloned.id, bo.id) self.assertIsNone(cloned.content_type_id) self.assertIsNone(cloned.object_id) self.assertEqual(cloned.state_logs.count(), 0) def test_clone_rejects_mismatched_content_type(self): """content_type 不一致应拒绝克隆""" # 用另一个 ContentType（比如 State） ct_state = ContentType.objects.get_for_model(models.State) self.assertNotEqual(ct_state.id, self.business_object.content_type_id) with self.assertRaises(ValueError): services.clone_business_object( self.business_object, new_object_id=(self.business_object.object_id or 0) + 1000, expected_content_type_id=ct_state.id, ) def test_clone_requires_new_object_id_when_content_type_present(self): """源对象有 content_type 时，必须提供新的 object_id""" with self.assertRaises(ValueError): services.clone_business_object( self.business_object, new_object_id=None, expected_content_type_id=self.business_object.content_type_id, ) def test_clone_rejects_same_object_id(self): """object_id 与源对象相同应拒绝克隆""" with self.assertRaises(ValueError): services.clone_business_object( self.business_object, new_object_id=self.business_object.object_id, expected_content_type_id=self.business_object.content_type_id, )