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_verify_succeed_state_fields():
with pytest.raises(TypeError):
Succeed('Succeed', unknown_field='Unknown Field')
def test_append_states_after_terminal_state_will_fail():
with pytest.raises(ValueError):
chain = Chain()
chain.append(Pass('Pass'))
chain.append(Fail('Fail'))
chain.append(Pass('Pass2'))
with pytest.raises(ValueError):
chain = Chain()
chain.append(Pass('Pass'))
chain.append(Succeed('Succeed'))
chain.append(Pass('Pass2'))
with pytest.raises(ValueError):
chain = Chain()
chain.append(Pass('Pass'))
chain.append(Choice('Choice'))
chain.append(Pass('Pass2'))
def test_succeed_state_creation():
succeed_state = Succeed(
state_id='Succeed',
comment='This is a comment'
)
assert succeed_state.state_id == 'Succeed'
assert succeed_state.comment == 'This is a comment'
assert succeed_state.to_dict() == {
'Type': 'Succeed',
'Comment': 'This is a comment'
}
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()
expected_repr = {
"StartAt": "FirstState",
"States": {
"FirstState": {
"Resource": "arn:aws:lambda:us-east-1:1234567890:function:FirstState",
"Type": "Task",