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_paths_none_converted_to_null():
task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
result_path=None,
input_path=None,
output_path=None)
assert '"ResultPath": null' in task_state.to_json()
assert '"InputPath": null' in task_state.to_json()
assert '"OutputPath": null' in task_state.to_json()
def test_default_paths_not_included():
task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
assert 'ResultPath' not in task_state.to_dict()
assert 'InputPath' not in task_state.to_dict()
assert 'OutputPath' not in task_state.to_dict()
def test_choice_state_with_placeholders():
first_state = Task('FirstState', resource='arn:aws:lambda:us-east-1:1234567890:function:FirstState')
retry = Chain([Pass('Retry'), Pass('Cleanup'), first_state])
choice_state = Choice('Is Completed?')
choice_state.add_choice(
ChoiceRule.BooleanEquals(choice_state.output()["Completed"], True),
Succeed('Complete')
)
choice_state.add_choice(
ChoiceRule.BooleanEquals(choice_state.output()["Completed"], False),
retry
)
first_state.next(choice_state)
result = Graph(first_state).to_dict()
def test_paths_none():
task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
result_path=None,
input_path=None,
output_path=None)
assert 'ResultPath' in task_state.to_dict()
assert task_state.to_dict()['ResultPath'] is None
assert 'InputPath' in task_state.to_dict()
assert task_state.to_dict()['InputPath'] is None
assert 'OutputPath' in task_state.to_dict()
assert task_state.to_dict()['OutputPath'] is None
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
},
def test_default_paths_not_converted_to_null():
task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
assert '"ResultPath": null' not in task_state.to_json()
assert '"InputPath": null' not in task_state.to_json()
assert '"OutputPath": null' not in task_state.to_json()