Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_catch_creation():
catch = Catch(error_equals=['States.ALL'], next_step=Fail('End'))
assert catch.to_dict() == {
'ErrorEquals': ['States.ALL'],
'Next': 'End'
}
def test_retry_fail_for_unsupported_state():
c1 = Choice('My Choice')
with pytest.raises(ValueError):
c1.add_catch(Catch(error_equals=["States.NoChoiceMatched"], next_step=Fail("ChoiceFailed")))
def test_task_state_creation():
task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
task_state.add_retry(Retry(error_equals=['ErrorA', 'ErrorB'], interval_seconds=1, max_attempts=2, backoff_rate=2))
task_state.add_retry(Retry(error_equals=['ErrorC'], interval_seconds=5))
task_state.add_catch(Catch(error_equals=['States.ALL'], next_step=Pass('End State')))
assert task_state.type == 'Task'
assert len(task_state.retries) == 2
assert len(task_state.catches) == 1
assert task_state.to_dict() == {
'Type': 'Task',
'Resource': 'arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
'Retry': [
{
'ErrorEquals': ['ErrorA', 'ErrorB'],
'IntervalSeconds': 1,
'BackoffRate': 2,
'MaxAttempts': 2
},
{
'ErrorEquals': ['ErrorC'],
'IntervalSeconds': 5