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_chaining_steps():
s1 = Pass('Step - One')
s2 = Pass('Step - Two')
s3 = Pass('Step - Three')
Chain([s1, s2])
assert s1.next_step == s2
assert s2.next_step is None
chain1 = Chain([s2, s3])
assert s2.next_step == s3
chain2 = Chain([s1, s3])
assert s1.next_step == s3
assert s2.next_step == s1.next_step
with pytest.raises(DuplicateStatesInChain):
chain2.append(s3)
with pytest.raises(DuplicateStatesInChain):
chain3 = Chain([chain1, chain2])
s1.next(s2)
chain3 = Chain([s3, s1])
assert chain3.steps == [s3, s1]
assert s3.next_step == s1
assert s1.next_step == s2
assert s2.next_step == s3
Chain([Chain([s3]), Chain([s1])])
with pytest.raises(DuplicateStatesInChain):
Chain([Chain([s1, s2, s1]), s3])
s3 = Pass('Step - Three')
Chain([s1, s2])
assert s1.next_step == s2
assert s2.next_step is None
chain1 = Chain([s2, s3])
assert s2.next_step == s3
chain2 = Chain([s1, s3])
assert s1.next_step == s3
assert s2.next_step == s1.next_step
with pytest.raises(DuplicateStatesInChain):
chain2.append(s3)
with pytest.raises(DuplicateStatesInChain):
chain3 = Chain([chain1, chain2])
s1.next(s2)
chain3 = Chain([s3, s1])
assert chain3.steps == [s3, s1]
assert s3.next_step == s1
assert s1.next_step == s2
assert s2.next_step == s3
Chain([Chain([s3]), Chain([s1])])
with pytest.raises(DuplicateStatesInChain):
Chain([Chain([s1, s2, s1]), s3])
Chain([s1, s2, s1, s3])
Chain([Chain([s1, s2]), s3])
assert s1.next_step == s2
def append(self, step):
"""Add a state at the tail end of the chain.
Args:
step (State): State to insert at the tail end of the chain.
"""
if len(self.steps) == 0:
self.steps.append(step)
else:
if step in self.steps:
raise DuplicateStatesInChain("State '{step_name}' is already inside this chain. A chain cannot have duplicate states.".format(step_name=step.state_id))
last_step = self.steps[-1]
last_step.next(step)
self.steps.append(step)
def __init__(self, steps=[]):
"""
Args:
steps (list(State), optional): List of states to be chained in-order. (default: [])
"""
if not isinstance(steps, list):
raise TypeError("Chain takes a 'list' of steps. You provided an input that is not a list.")
self.steps = []
steps_expanded = []
[steps_expanded.extend(step) if isinstance(step, Chain) else steps_expanded.append(step) for step in steps]
for step in steps_expanded:
if steps_expanded.count(step) > 1:
raise DuplicateStatesInChain("Duplicate states in the chain.")
list(map(self.append, steps_expanded))