TermiNexus
1a73483db5
- _DoReconnect 互斥锁移入内部，修复 force_deregister 路径锁协议破坏 - 订单 OGM 键名 GlobalID→ObjectGlobalID（映射不再丢失） - RemoteCallDemo 对 requests.Response 先解析 JSON，避免 500 - 补充 Compensate 状态守卫回归用例（20/20 通过）
249 lines
9.3 KiB
Python
249 lines
9.3 KiB
Python
"""MSYNC Saga 状态机回归测试（独立运行，不依赖主库）
|
||
|
||
用法：
|
||
python gvsds/tests/test_msync_saga.py
|
||
|
||
覆盖：
|
||
- 正向全部成功 → Committed，且写入 TransactionResult 汇总
|
||
- 步骤失败 → 逆序补偿已完成步骤 → Compensated
|
||
- 自动重试：步骤失败后按 CanRetry 内联重试成功 → Committed
|
||
- 补偿失败 → 事务 FAILED，RetryCompensate 成功后 → COMPENSATED
|
||
- 本地事务处理器幂等去重（同 Idempotency-Key 返回首次结果）
|
||
|
||
说明：远程调用通过 mock 注入，不发起真实 HTTP；模型建于 sqlite 内存库。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
|
||
# 将项目根加入 sys.path（脚本可能从任意 cwd 运行）
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||
|
||
import django
|
||
from django.conf import settings
|
||
|
||
_TEST_SETTINGS = {
|
||
'DEBUG': True,
|
||
'SECRET_KEY': 'msync-test-key',
|
||
'USE_TZ': False,
|
||
'DATABASES': {
|
||
'default': {
|
||
'ENGINE': 'django.db.backends.sqlite3',
|
||
'NAME': ':memory:',
|
||
},
|
||
},
|
||
'INSTALLED_APPS': [
|
||
'django.contrib.contenttypes',
|
||
'django.contrib.auth',
|
||
'gvsdsdk',
|
||
'gvsdsdk.msync',
|
||
],
|
||
}
|
||
|
||
|
||
def _Setup():
|
||
settings.configure(**_TEST_SETTINGS)
|
||
django.setup()
|
||
from django.core.management import call_command
|
||
call_command('migrate', run_syncdb=True, verbosity=0, interactive=False)
|
||
|
||
|
||
def _Run():
|
||
_Setup()
|
||
|
||
from unittest import mock
|
||
import uuid
|
||
from gvsdsdk.msync.transaction import TransactionManager, StepExecutionError, TransactionError
|
||
from gvsdsdk.msync.local_tx import local_tx
|
||
from gvsdsdk.msync.models import ServiceNode
|
||
|
||
# 注册参与服务节点（Saga 步骤的 ParticipantService）
|
||
ServiceNode.objects.create(
|
||
ServiceUUID=uuid.uuid4().bytes,
|
||
ServiceName='svc',
|
||
ServiceDomain='http://svc.local',
|
||
ServiceSecretHash='x',
|
||
ServiceStatus=1,
|
||
)
|
||
|
||
results = []
|
||
|
||
def check(name, cond):
|
||
results.append((name, cond))
|
||
print(f' {"PASS" if cond else "FAIL"} {name}')
|
||
|
||
# ─── 1. 正向全部成功 → Committed ───
|
||
print('[1] 正向全部成功')
|
||
dtx = TransactionManager.Create(
|
||
name='t1',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc', 'ActionPayload': {'a': 1}},
|
||
{'name': 's2', 'ParticipantService': 'svc', 'ActionPayload': {'b': 2}},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
{'user_uuid': 'u1'}, # s1 结果 → 数据流注入 s2
|
||
{'ok': True},
|
||
]) as call:
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
check('状态为 Committed', dtx.TransactionStatus == 2)
|
||
check('TransactionResult 已写入', bool(dtx.TransactionResult and dtx.TransactionResult.get('Steps')))
|
||
# 数据流：s2 的 payload 应包含 s1 结果（user_uuid）
|
||
s2_payload = call.call_args_list[1].kwargs['payload']['payload']
|
||
check('步骤数据流（s1 结果注入 s2）', s2_payload.get('user_uuid') == 'u1')
|
||
|
||
# ─── 2. 步骤失败 → 补偿已完成步骤 → Compensated ───
|
||
print('[2] 步骤失败触发补偿')
|
||
dtx = TransactionManager.Create(
|
||
name='t2',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc'},
|
||
{'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0},
|
||
{'name': 's3', 'ParticipantService': 'svc'},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
{'ok': 1}, # s1 成功
|
||
StepExecutionError('s2 失败'), # s2 失败（自动重试耗尽）
|
||
{'ok': 'compensated'}, # 补偿 s1
|
||
]):
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
check('状态为 Compensated', dtx.TransactionStatus == 4)
|
||
steps = {s.StepIndex: s for s in dtx.steps.all()}
|
||
check('s1 已补偿', steps[0].StepStatus == 4)
|
||
check('s2 正向失败', steps[1].StepStatus == 5)
|
||
|
||
# ─── 3. 自动重试：首次失败后成功 → Committed ───
|
||
print('[3] 自动重试')
|
||
dtx = TransactionManager.Create(
|
||
name='t3',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc', 'MaxRetries': 2},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
StepExecutionError('临时失败'),
|
||
{'ok': True},
|
||
]):
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
step = dtx.steps.first()
|
||
check('状态为 Committed', dtx.TransactionStatus == 2)
|
||
check('RetryCount=1', step.RetryCount == 1)
|
||
|
||
# ─── 4. 补偿失败 → Failed；RetryCompensate 成功后 → Compensated ───
|
||
print('[4] 补偿失败可重试')
|
||
dtx = TransactionManager.Create(
|
||
name='t4',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc'},
|
||
{'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
{'ok': 1},
|
||
StepExecutionError('s2 失败'),
|
||
StepExecutionError('补偿 s1 失败'),
|
||
]):
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
check('补偿失败后事务 Failed', dtx.TransactionStatus == 5)
|
||
# RetryCompensate：先成功补偿 s1
|
||
with mock.patch.object(TransactionManager, '_call_service', return_value={'ok': True}):
|
||
TransactionManager.RetryCompensate(dtx.TransactionCode, 0)
|
||
dtx.refresh_from_db()
|
||
check('RetryCompensate 后 Compensated', dtx.TransactionStatus == 4)
|
||
|
||
# ─── 5. 本地事务幂等去重 ───
|
||
print('[5] local_tx 幂等')
|
||
r1 = local_tx.ExecuteStep('no_such', {}, 'K:1:0')
|
||
r2 = local_tx.ExecuteStep('no_such', {}, 'K:1:0')
|
||
check('同键幂等命中', r1 is r2)
|
||
r3 = local_tx.ExecuteStep('no_such', {}, 'K:1:1')
|
||
check('不同键不命中', r1 is not r3)
|
||
|
||
# ─── 6. RetryStep 手动重试（自动重试耗尽后仍可重试，计数分离）───
|
||
print('[6] RetryStep 手动重试')
|
||
dtx = TransactionManager.Create(
|
||
name='t6',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc'},
|
||
{'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
{'ok': 1},
|
||
StepExecutionError('s2 永久失败'), # s2 自动重试耗尽
|
||
StepExecutionError('补偿 s1 也失败'), # 补偿失败 → 事务 FAILED
|
||
]):
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
check('自动重试耗尽后事务 Failed', dtx.TransactionStatus == 5)
|
||
# s2 MaxRetries=0 已耗尽，原逻辑（CanRetry）拒绝；现手动重试不受配额限制
|
||
with mock.patch.object(TransactionManager, '_call_service', return_value={'ok': True}):
|
||
step = TransactionManager.RetryStep(dtx.TransactionCode, 1)
|
||
check('手动重试成功 Completed', step.StepStatus == 2)
|
||
check('手动重试计数独立', step.RetryCount == 1)
|
||
|
||
# ─── 7. RetryStep 拒绝已补偿事务（事务状态守卫）───
|
||
print('[7] RetryStep 已补偿事务拒绝')
|
||
dtx = TransactionManager.Create(
|
||
name='t7',
|
||
steps_data=[
|
||
{'name': 's1', 'ParticipantService': 'svc'},
|
||
{'name': 's2', 'ParticipantService': 'svc', 'MaxRetries': 0},
|
||
],
|
||
)
|
||
with mock.patch.object(TransactionManager, '_call_service', side_effect=[
|
||
{'ok': 1},
|
||
StepExecutionError('s2 失败'),
|
||
{'ok': True}, # s1 补偿成功 → 事务 COMPENSATED
|
||
]):
|
||
TransactionManager.Execute(dtx.TransactionCode)
|
||
dtx.refresh_from_db()
|
||
check('事务已补偿', dtx.TransactionStatus == 4)
|
||
rejected = False
|
||
try:
|
||
TransactionManager.RetryStep(dtx.TransactionCode, 1) # s2 正向失败步骤
|
||
except TransactionError:
|
||
rejected = True
|
||
check('已补偿事务 RetryStep 被拒绝', rejected)
|
||
|
||
# ─── 8. Compensate 状态守卫：PENDING/COMMITTED 拒绝补偿 ───
|
||
print('[8] Compensate 状态守卫')
|
||
dtx8 = TransactionManager.Create(
|
||
name='t8',
|
||
steps_data=[{'name': 's1', 'ParticipantService': 'svc'}],
|
||
)
|
||
rejected_pending = False
|
||
try:
|
||
TransactionManager.Compensate(dtx8.TransactionCode)
|
||
except TransactionError:
|
||
rejected_pending = True
|
||
check('PENDING 事务 Compensate 被拒绝', rejected_pending)
|
||
with mock.patch.object(TransactionManager, '_call_service', return_value={'ok': True}):
|
||
TransactionManager.Execute(dtx8.TransactionCode)
|
||
dtx8.refresh_from_db()
|
||
check('事务已提交', dtx8.TransactionStatus == 2)
|
||
rejected_committed = False
|
||
try:
|
||
TransactionManager.Compensate(dtx8.TransactionCode)
|
||
except TransactionError:
|
||
rejected_committed = True
|
||
check('COMMITTED 事务 Compensate 被拒绝', rejected_committed)
|
||
|
||
# ─── 汇总 ───
|
||
passed = sum(1 for _, ok in results if ok)
|
||
print(f'\n结果: {passed}/{len(results)} 通过')
|
||
if passed != len(results):
|
||
for name, ok in results:
|
||
if not ok:
|
||
print(f' 未通过: {name}')
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
_Run()
|