forked from erp-dev/erp
237 lines
10 KiB
Python
237 lines
10 KiB
Python
"""
|
||
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,
|
||
)
|
||
|
||
|