41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
from flashops_control.engine.recovery import RecoveryContext, next_recovery_level
|
|
from flashops_control.enums import RecoveryLevel, RecoveryTrigger
|
|
|
|
|
|
def test_integrity_failure_always_freezes():
|
|
level = next_recovery_level(
|
|
RecoveryContext(
|
|
trigger=RecoveryTrigger.DATA_INTEGRITY,
|
|
attempt=0,
|
|
agent_online=True,
|
|
host_network_online=True,
|
|
oob_online=True,
|
|
)
|
|
)
|
|
assert level == RecoveryLevel.FREEZE
|
|
|
|
|
|
def test_offline_oob_removes_destructive_recovery_levels():
|
|
level = next_recovery_level(
|
|
RecoveryContext(
|
|
trigger=RecoveryTrigger.HEARTBEAT_LOST,
|
|
attempt=2,
|
|
agent_online=True,
|
|
host_network_online=True,
|
|
oob_online=False,
|
|
)
|
|
)
|
|
assert level == RecoveryLevel.FREEZE
|
|
|
|
|
|
def test_recovery_escalates_in_order():
|
|
context = dict(
|
|
trigger=RecoveryTrigger.HEARTBEAT_LOST,
|
|
agent_online=True,
|
|
host_network_online=True,
|
|
oob_online=True,
|
|
)
|
|
assert next_recovery_level(RecoveryContext(attempt=0, **context)) == RecoveryLevel.AGENT_SOFT
|
|
assert next_recovery_level(RecoveryContext(attempt=1, **context)) == RecoveryLevel.OS_REBOOT
|
|
assert next_recovery_level(RecoveryContext(attempt=2, **context)) == RecoveryLevel.OOB_RESET
|