83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
from dataclasses import dataclass
|
|
|
|
from parsers.state_machine import End, Start, State, StateMachine, StateResult
|
|
|
|
|
|
@dataclass
|
|
class DummyExecutionContext:
|
|
count: int
|
|
|
|
def to_debug(self):
|
|
return {"count": self.count}
|
|
|
|
|
|
class GenericTestState(State):
|
|
def __init__(self, name, next_state, fork=None):
|
|
super().__init__(name=name, next_states=[next_state])
|
|
self.next_state = next_state
|
|
self.fork = fork
|
|
|
|
def run(self, state_context) -> StateResult:
|
|
return StateResult(self.next_state, self.fork)
|
|
|
|
def __repr__(self):
|
|
return f"(GenericTestState {self.name} -> {self.next_state}, forks={len(self.fork) if self.fork else 0})"
|
|
|
|
|
|
def test_i_can_execute_a_workflow():
|
|
wkf_as_list = [Start("start", ["a"]),
|
|
GenericTestState("a", "b"),
|
|
GenericTestState("b", "c"),
|
|
GenericTestState("c", "end"),
|
|
End("end", None)]
|
|
|
|
wkf = {state.name: state for state in wkf_as_list}
|
|
|
|
state_machine = StateMachine({"#wkf": wkf})
|
|
state_machine.run("#wkf", "start", DummyExecutionContext(0))
|
|
|
|
assert len(state_machine.paths) == 1
|
|
assert state_machine.paths[0].get_audit_trail() == ["#wkf:start", "#wkf:a", "#wkf:b", "#wkf:c", "#wkf:end"]
|
|
|
|
|
|
def test_i_can_change_workflow():
|
|
wkf1_as_list = [Start("start", ["a"]),
|
|
GenericTestState("a", "#wkf2")]
|
|
|
|
wkf2_as_list = [Start("start", ["c"]),
|
|
GenericTestState("c", "end"),
|
|
End("end", None)]
|
|
|
|
wkfs = {
|
|
"#wkf1": {state.name: state for state in wkf1_as_list},
|
|
"#wkf2": {state.name: state for state in wkf2_as_list}
|
|
}
|
|
|
|
state_machine = StateMachine(wkfs)
|
|
state_machine.run("#wkf1", "start", DummyExecutionContext(0))
|
|
|
|
assert len(state_machine.paths) == 1
|
|
assert state_machine.paths[0].get_audit_trail() == ["#wkf1:start", "#wkf1:a", "#wkf2:start", "#wkf2:c", "#wkf2:end"]
|
|
|
|
|
|
def test_i_can_fork_path():
|
|
wkf_as_list = [Start("start", ["a"]),
|
|
GenericTestState("a", "end", [("b", DummyExecutionContext(i)) for i in range(3)]),
|
|
GenericTestState("b", "end"),
|
|
End("end", None)]
|
|
|
|
wkf = {state.name: state for state in wkf_as_list}
|
|
|
|
state_machine = StateMachine({"#wkf": wkf})
|
|
state_machine.run("#wkf", "start", DummyExecutionContext(0))
|
|
|
|
assert len(state_machine.paths) == 4
|
|
assert state_machine.paths[0].get_audit_trail() == ["#wkf:start", "#wkf:a", "#wkf:end"]
|
|
assert state_machine.paths[0].history[1].forks == [1, 2, 3]
|
|
assert state_machine.paths[1].get_audit_trail() == ["#wkf:start", "#wkf:a", "#wkf:b", "#wkf:end"]
|
|
assert state_machine.paths[1].history[0].parents == [0]
|
|
assert state_machine.paths[2].get_audit_trail() == ["#wkf:start", "#wkf:a", "#wkf:b", "#wkf:end"]
|
|
assert state_machine.paths[2].history[0].parents == [0]
|
|
assert state_machine.paths[3].get_audit_trail() == ["#wkf:start", "#wkf:a", "#wkf:b", "#wkf:end"]
|
|
assert state_machine.paths[3].history[0].parents == [0]
|