From d0f330e5ee71282f7ce5ccdcda79b12316807d08 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Tue, 3 Mar 2026 14:37:05 -0300 Subject: [PATCH] refactor: extract test StateChart definitions to tests/machines/ Move reusable StateChart class definitions from inline test code to dedicated modules under tests/machines/, organized by feature: - workflow/: CampaignMachine variants, ReverseTrafficLightMachine - error/: ErrorInGuardSC, ErrorInActionSC, ErrorInErrorHandlerSC, etc. - validators/: OrderValidation, MultiValidator, ValidatorWithCond, etc. - compound/: ShireToRivendell, MoriaExpedition, MiddleEarthJourney, etc. - history/: GollumPersonality variants, DeepMemoryOfMoria, ShallowMoria - parallel/: WarOfTheRing, TwoTowers, Session variants - eventless/: RingCorruption variants, BeaconChain, AutoAdvance, etc. - in_condition/: Fellowship, GateOfMoria, CombinedGuard, etc. - donedata/: DestroyTheRing, NestedQuestDoneData, QuestForErebor variants Test files now import from these modules instead of defining classes inline. Machines that test invalid definitions or use closures remain inline in test files. --- tests/conftest.py | 83 +-------- tests/machines/compound/__init__.py | 0 .../machines/compound/middle_earth_journey.py | 25 +++ .../middle_earth_journey_two_compounds.py | 18 ++ .../middle_earth_journey_with_finals.py | 25 +++ tests/machines/compound/moria_expedition.py | 15 ++ .../compound/moria_expedition_with_escape.py | 18 ++ tests/machines/compound/quest_for_erebor.py | 13 ++ tests/machines/compound/shire_to_rivendell.py | 13 ++ tests/machines/donedata/__init__.py | 0 tests/machines/donedata/destroy_the_ring.py | 20 +++ .../donedata/destroy_the_ring_simple.py | 17 ++ .../donedata/nested_quest_donedata.py | 22 +++ .../quest_for_erebor_done_convention.py | 13 ++ .../donedata/quest_for_erebor_explicit_id.py | 14 ++ .../donedata/quest_for_erebor_multi_word.py | 13 ++ .../donedata/quest_for_erebor_with_event.py | 14 ++ tests/machines/error/__init__.py | 0 .../machines/error/error_convention_event.py | 16 ++ .../error/error_convention_transition_list.py | 15 ++ tests/machines/error/error_in_action_sc.py | 15 ++ .../error/error_in_action_sm_with_flag.py | 17 ++ tests/machines/error/error_in_after_sc.py | 15 ++ .../error/error_in_error_handler_sc.py | 24 +++ tests/machines/error/error_in_guard_sc.py | 14 ++ tests/machines/error/error_in_guard_sm.py | 15 ++ tests/machines/error/error_in_on_enter_sc.py | 15 ++ tests/machines/eventless/__init__.py | 0 tests/machines/eventless/auto_advance.py | 15 ++ tests/machines/eventless/beacon_chain.py | 13 ++ .../eventless/beacon_chain_lighting.py | 18 ++ .../machines/eventless/coordinated_advance.py | 18 ++ tests/machines/eventless/ring_corruption.py | 18 ++ .../ring_corruption_with_bear_ring.py | 18 ++ .../eventless/ring_corruption_with_tick.py | 15 ++ tests/machines/history/__init__.py | 0 .../machines/history/deep_memory_of_moria.py | 21 +++ tests/machines/history/gollum_personality.py | 17 ++ .../gollum_personality_default_gollum.py | 17 ++ .../gollum_personality_with_default.py | 17 ++ tests/machines/history/shallow_moria.py | 21 +++ tests/machines/in_condition/__init__.py | 0 tests/machines/in_condition/combined_guard.py | 18 ++ .../machines/in_condition/descendant_check.py | 15 ++ tests/machines/in_condition/eventless_in.py | 18 ++ tests/machines/in_condition/fellowship.py | 18 ++ .../in_condition/fellowship_coordination.py | 18 ++ tests/machines/in_condition/gate_of_moria.py | 13 ++ tests/machines/parallel/__init__.py | 0 tests/machines/parallel/session.py | 17 ++ .../parallel/session_with_done_state.py | 20 +++ tests/machines/parallel/two_towers.py | 20 +++ tests/machines/parallel/war_of_the_ring.py | 25 +++ tests/machines/parallel/war_with_exit.py | 20 +++ tests/machines/validators/__init__.py | 0 tests/machines/validators/multi_validator.py | 19 ++ tests/machines/validators/order_validation.py | 17 ++ .../order_validation_no_error_events.py | 19 ++ .../validators/validator_fallthrough.py | 20 +++ .../validators/validator_with_cond.py | 17 ++ .../validator_with_error_transition.py | 25 +++ tests/machines/workflow/__init__.py | 0 tests/machines/workflow/campaign_machine.py | 14 ++ .../campaign_machine_with_validator.py | 18 ++ .../workflow/campaign_machine_with_values.py | 14 ++ .../workflow/reverse_traffic_light.py | 13 ++ tests/test_error_execution.py | 130 +------------- tests/test_mixins.py | 2 +- tests/test_statechart_compound.py | 150 ++-------------- tests/test_statechart_donedata.py | 121 ++----------- tests/test_statechart_eventless.py | 114 +----------- tests/test_statechart_history.py | 130 +------------- tests/test_statechart_in_condition.py | 96 +--------- tests/test_statechart_parallel.py | 167 +++--------------- tests/test_validators.py | 122 +------------ 75 files changed, 1067 insertions(+), 1020 deletions(-) create mode 100644 tests/machines/compound/__init__.py create mode 100644 tests/machines/compound/middle_earth_journey.py create mode 100644 tests/machines/compound/middle_earth_journey_two_compounds.py create mode 100644 tests/machines/compound/middle_earth_journey_with_finals.py create mode 100644 tests/machines/compound/moria_expedition.py create mode 100644 tests/machines/compound/moria_expedition_with_escape.py create mode 100644 tests/machines/compound/quest_for_erebor.py create mode 100644 tests/machines/compound/shire_to_rivendell.py create mode 100644 tests/machines/donedata/__init__.py create mode 100644 tests/machines/donedata/destroy_the_ring.py create mode 100644 tests/machines/donedata/destroy_the_ring_simple.py create mode 100644 tests/machines/donedata/nested_quest_donedata.py create mode 100644 tests/machines/donedata/quest_for_erebor_done_convention.py create mode 100644 tests/machines/donedata/quest_for_erebor_explicit_id.py create mode 100644 tests/machines/donedata/quest_for_erebor_multi_word.py create mode 100644 tests/machines/donedata/quest_for_erebor_with_event.py create mode 100644 tests/machines/error/__init__.py create mode 100644 tests/machines/error/error_convention_event.py create mode 100644 tests/machines/error/error_convention_transition_list.py create mode 100644 tests/machines/error/error_in_action_sc.py create mode 100644 tests/machines/error/error_in_action_sm_with_flag.py create mode 100644 tests/machines/error/error_in_after_sc.py create mode 100644 tests/machines/error/error_in_error_handler_sc.py create mode 100644 tests/machines/error/error_in_guard_sc.py create mode 100644 tests/machines/error/error_in_guard_sm.py create mode 100644 tests/machines/error/error_in_on_enter_sc.py create mode 100644 tests/machines/eventless/__init__.py create mode 100644 tests/machines/eventless/auto_advance.py create mode 100644 tests/machines/eventless/beacon_chain.py create mode 100644 tests/machines/eventless/beacon_chain_lighting.py create mode 100644 tests/machines/eventless/coordinated_advance.py create mode 100644 tests/machines/eventless/ring_corruption.py create mode 100644 tests/machines/eventless/ring_corruption_with_bear_ring.py create mode 100644 tests/machines/eventless/ring_corruption_with_tick.py create mode 100644 tests/machines/history/__init__.py create mode 100644 tests/machines/history/deep_memory_of_moria.py create mode 100644 tests/machines/history/gollum_personality.py create mode 100644 tests/machines/history/gollum_personality_default_gollum.py create mode 100644 tests/machines/history/gollum_personality_with_default.py create mode 100644 tests/machines/history/shallow_moria.py create mode 100644 tests/machines/in_condition/__init__.py create mode 100644 tests/machines/in_condition/combined_guard.py create mode 100644 tests/machines/in_condition/descendant_check.py create mode 100644 tests/machines/in_condition/eventless_in.py create mode 100644 tests/machines/in_condition/fellowship.py create mode 100644 tests/machines/in_condition/fellowship_coordination.py create mode 100644 tests/machines/in_condition/gate_of_moria.py create mode 100644 tests/machines/parallel/__init__.py create mode 100644 tests/machines/parallel/session.py create mode 100644 tests/machines/parallel/session_with_done_state.py create mode 100644 tests/machines/parallel/two_towers.py create mode 100644 tests/machines/parallel/war_of_the_ring.py create mode 100644 tests/machines/parallel/war_with_exit.py create mode 100644 tests/machines/validators/__init__.py create mode 100644 tests/machines/validators/multi_validator.py create mode 100644 tests/machines/validators/order_validation.py create mode 100644 tests/machines/validators/order_validation_no_error_events.py create mode 100644 tests/machines/validators/validator_fallthrough.py create mode 100644 tests/machines/validators/validator_with_cond.py create mode 100644 tests/machines/validators/validator_with_error_transition.py create mode 100644 tests/machines/workflow/__init__.py create mode 100644 tests/machines/workflow/campaign_machine.py create mode 100644 tests/machines/workflow/campaign_machine_with_validator.py create mode 100644 tests/machines/workflow/campaign_machine_with_values.py create mode 100644 tests/machines/workflow/reverse_traffic_light.py diff --git a/tests/conftest.py b/tests/conftest.py index 105dad84..f229e518 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,86 +22,32 @@ def current_time(): @pytest.fixture() def campaign_machine(): - "Define a new class for each test" - from statemachine import State - from statemachine import StateChart - - class CampaignMachine(StateChart): - "A workflow machine" - - draft = State(initial=True) - producing = State("Being produced") - closed = State(final=True) - - add_job = draft.to(draft) | producing.to(producing) - produce = draft.to(producing) - deliver = producing.to(closed) + from tests.machines.workflow.campaign_machine import CampaignMachine return CampaignMachine @pytest.fixture() def campaign_machine_with_validator(): - "Define a new class for each test" - from statemachine import State - from statemachine import StateChart - - class CampaignMachine(StateChart): - "A workflow machine" - - draft = State(initial=True) - producing = State("Being produced") - closed = State(final=True) - - add_job = draft.to(draft) | producing.to(producing) - produce = draft.to(producing, validators="can_produce") - deliver = producing.to(closed) + from tests.machines.workflow.campaign_machine_with_validator import ( + CampaignMachineWithValidator, + ) - def can_produce(*args, **kwargs): - if "goods" not in kwargs: - raise LookupError("Goods not found.") - - return CampaignMachine + return CampaignMachineWithValidator @pytest.fixture() def campaign_machine_with_final_state(): - "Define a new class for each test" - from statemachine import State - from statemachine import StateChart - - class CampaignMachine(StateChart): - "A workflow machine" - - draft = State(initial=True) - producing = State("Being produced") - closed = State(final=True) - - add_job = draft.to(draft) | producing.to(producing) - produce = draft.to(producing) - deliver = producing.to(closed) + from tests.machines.workflow.campaign_machine import CampaignMachine return CampaignMachine @pytest.fixture() def campaign_machine_with_values(): - "Define a new class for each test" - from statemachine import State - from statemachine import StateChart - - class CampaignMachineWithKeys(StateChart): - "A workflow machine" - - draft = State(initial=True, value=1) - producing = State("Being produced", value=2) - closed = State(value=3, final=True) - - add_job = draft.to(draft) | producing.to(producing) - produce = draft.to(producing) - deliver = producing.to(closed) + from tests.machines.workflow.campaign_machine_with_values import CampaignMachineWithValues - return CampaignMachineWithKeys + return CampaignMachineWithValues @pytest.fixture() @@ -153,18 +99,7 @@ def classic_traffic_light_machine_allow_event(classic_traffic_light_machine): @pytest.fixture() def reverse_traffic_light_machine(): - from statemachine import State - from statemachine import StateChart - - class ReverseTrafficLightMachine(StateChart): - "A traffic light machine" - - green = State(initial=True) - yellow = State() - red = State() - - stop = red.from_(yellow, green, red) - cycle = green.from_(red) | yellow.from_(green) | red.from_(yellow) | red.from_.itself() + from tests.machines.workflow.reverse_traffic_light import ReverseTrafficLightMachine return ReverseTrafficLightMachine diff --git a/tests/machines/compound/__init__.py b/tests/machines/compound/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/compound/middle_earth_journey.py b/tests/machines/compound/middle_earth_journey.py new file mode 100644 index 00000000..6f087e5c --- /dev/null +++ b/tests/machines/compound/middle_earth_journey.py @@ -0,0 +1,25 @@ +from statemachine import State +from statemachine import StateChart + + +class MiddleEarthJourney(StateChart): + class rivendell(State.Compound): + council = State(initial=True) + preparing = State() + + get_ready = council.to(preparing) + + class moria(State.Compound): + gates = State(initial=True) + bridge = State(final=True) + + cross = gates.to(bridge) + + class lothlorien(State.Compound): + mirror = State(initial=True) + departure = State(final=True) + + leave = mirror.to(departure) + + march_to_moria = rivendell.to(moria) + march_to_lorien = moria.to(lothlorien) diff --git a/tests/machines/compound/middle_earth_journey_two_compounds.py b/tests/machines/compound/middle_earth_journey_two_compounds.py new file mode 100644 index 00000000..f1719527 --- /dev/null +++ b/tests/machines/compound/middle_earth_journey_two_compounds.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class MiddleEarthJourneyTwoCompounds(StateChart): + class rivendell(State.Compound): + council = State(initial=True) + preparing = State() + + get_ready = council.to(preparing) + + class moria(State.Compound): + gates = State(initial=True) + bridge = State(final=True) + + cross = gates.to(bridge) + + march_to_moria = rivendell.to(moria) diff --git a/tests/machines/compound/middle_earth_journey_with_finals.py b/tests/machines/compound/middle_earth_journey_with_finals.py new file mode 100644 index 00000000..e5cc0148 --- /dev/null +++ b/tests/machines/compound/middle_earth_journey_with_finals.py @@ -0,0 +1,25 @@ +from statemachine import State +from statemachine import StateChart + + +class MiddleEarthJourneyWithFinals(StateChart): + class rivendell(State.Compound): + council = State(initial=True) + preparing = State(final=True) + + get_ready = council.to(preparing) + + class moria(State.Compound): + gates = State(initial=True) + bridge = State(final=True) + + cross = gates.to(bridge) + + class lothlorien(State.Compound): + mirror = State(initial=True) + departure = State(final=True) + + leave = mirror.to(departure) + + march_to_moria = rivendell.to(moria) + march_to_lorien = moria.to(lothlorien) diff --git a/tests/machines/compound/moria_expedition.py b/tests/machines/compound/moria_expedition.py new file mode 100644 index 00000000..dc287da7 --- /dev/null +++ b/tests/machines/compound/moria_expedition.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class MoriaExpedition(StateChart): + class moria(State.Compound): + class upper_halls(State.Compound): + entrance = State(initial=True) + bridge = State(final=True) + + cross = entrance.to(bridge) + + assert isinstance(upper_halls, State) + depths = State(final=True) + descend = upper_halls.to(depths) diff --git a/tests/machines/compound/moria_expedition_with_escape.py b/tests/machines/compound/moria_expedition_with_escape.py new file mode 100644 index 00000000..4fc4d4d9 --- /dev/null +++ b/tests/machines/compound/moria_expedition_with_escape.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class MoriaExpeditionWithEscape(StateChart): + class moria(State.Compound): + class upper_halls(State.Compound): + entrance = State(initial=True) + bridge = State() + + cross = entrance.to(bridge) + + assert isinstance(upper_halls, State) + depths = State(final=True) + descend = upper_halls.to(depths) + + daylight = State(final=True) + escape = moria.to(daylight) diff --git a/tests/machines/compound/quest_for_erebor.py b/tests/machines/compound/quest_for_erebor.py new file mode 100644 index 00000000..b5a96b70 --- /dev/null +++ b/tests/machines/compound/quest_for_erebor.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class QuestForErebor(StateChart): + class lonely_mountain(State.Compound): + approach = State(initial=True) + inside = State(final=True) + + enter_mountain = approach.to(inside) + + victory = State(final=True) + done_state_lonely_mountain = lonely_mountain.to(victory) diff --git a/tests/machines/compound/shire_to_rivendell.py b/tests/machines/compound/shire_to_rivendell.py new file mode 100644 index 00000000..70de8951 --- /dev/null +++ b/tests/machines/compound/shire_to_rivendell.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class ShireToRivendell(StateChart): + class shire(State.Compound): + bag_end = State(initial=True) + green_dragon = State() + + visit_pub = bag_end.to(green_dragon) + + road = State(final=True) + depart = shire.to(road) diff --git a/tests/machines/donedata/__init__.py b/tests/machines/donedata/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/donedata/destroy_the_ring.py b/tests/machines/donedata/destroy_the_ring.py new file mode 100644 index 00000000..2c65e7de --- /dev/null +++ b/tests/machines/donedata/destroy_the_ring.py @@ -0,0 +1,20 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class DestroyTheRing(StateChart): + class quest(State.Compound): + traveling = State(initial=True) + completed = State(final=True, donedata="get_quest_result") + + finish = traveling.to(completed) + + def get_quest_result(self): + return {"ring_destroyed": True, "hero": "frodo"} + + epilogue = State(final=True) + done_state_quest = Event(quest.to(epilogue, on="capture_result")) # type: ignore[arg-type] + + def capture_result(self, ring_destroyed=None, hero=None, **kwargs): + self.received = {"ring_destroyed": ring_destroyed, "hero": hero} diff --git a/tests/machines/donedata/destroy_the_ring_simple.py b/tests/machines/donedata/destroy_the_ring_simple.py new file mode 100644 index 00000000..02197f9e --- /dev/null +++ b/tests/machines/donedata/destroy_the_ring_simple.py @@ -0,0 +1,17 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class DestroyTheRingSimple(StateChart): + class quest(State.Compound): + traveling = State(initial=True) + completed = State(final=True, donedata="get_result") + + finish = traveling.to(completed) + + def get_result(self): + return {"outcome": "victory"} + + celebration = State(final=True) + done_state_quest = Event(quest.to(celebration)) # type: ignore[arg-type] diff --git a/tests/machines/donedata/nested_quest_donedata.py b/tests/machines/donedata/nested_quest_donedata.py new file mode 100644 index 00000000..c1787a32 --- /dev/null +++ b/tests/machines/donedata/nested_quest_donedata.py @@ -0,0 +1,22 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class NestedQuestDoneData(StateChart): + class outer(State.Compound): + class inner(State.Compound): + start = State(initial=True) + end = State(final=True, donedata="inner_result") + + go = start.to(end) + + def inner_result(self): + return {"level": "inner"} + + assert isinstance(inner, State) + after_inner = State(final=True) + done_state_inner = Event(inner.to(after_inner)) # type: ignore[arg-type] + + final = State(final=True) + done_state_outer = Event(outer.to(final)) # type: ignore[arg-type] diff --git a/tests/machines/donedata/quest_for_erebor_done_convention.py b/tests/machines/donedata/quest_for_erebor_done_convention.py new file mode 100644 index 00000000..553297c9 --- /dev/null +++ b/tests/machines/donedata/quest_for_erebor_done_convention.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class QuestForEreborDoneConvention(StateChart): + class quest(State.Compound): + traveling = State(initial=True) + arrived = State(final=True) + + finish = traveling.to(arrived) + + celebration = State(final=True) + done_state_quest = quest.to(celebration) diff --git a/tests/machines/donedata/quest_for_erebor_explicit_id.py b/tests/machines/donedata/quest_for_erebor_explicit_id.py new file mode 100644 index 00000000..075f6eca --- /dev/null +++ b/tests/machines/donedata/quest_for_erebor_explicit_id.py @@ -0,0 +1,14 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class QuestForEreborExplicitId(StateChart): + class quest(State.Compound): + traveling = State(initial=True) + arrived = State(final=True) + + finish = traveling.to(arrived) + + celebration = State(final=True) + done_state_quest = Event(quest.to(celebration), id="done.state.quest") # type: ignore[arg-type] diff --git a/tests/machines/donedata/quest_for_erebor_multi_word.py b/tests/machines/donedata/quest_for_erebor_multi_word.py new file mode 100644 index 00000000..7fb842da --- /dev/null +++ b/tests/machines/donedata/quest_for_erebor_multi_word.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class QuestForEreborMultiWord(StateChart): + class lonely_mountain(State.Compound): + approach = State(initial=True) + inside = State(final=True) + + enter_mountain = approach.to(inside) + + victory = State(final=True) + done_state_lonely_mountain = lonely_mountain.to(victory) diff --git a/tests/machines/donedata/quest_for_erebor_with_event.py b/tests/machines/donedata/quest_for_erebor_with_event.py new file mode 100644 index 00000000..e2767fab --- /dev/null +++ b/tests/machines/donedata/quest_for_erebor_with_event.py @@ -0,0 +1,14 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class QuestForEreborWithEvent(StateChart): + class quest(State.Compound): + traveling = State(initial=True) + arrived = State(final=True) + + finish = traveling.to(arrived) + + celebration = State(final=True) + done_state_quest = Event(quest.to(celebration)) # type: ignore[arg-type] diff --git a/tests/machines/error/__init__.py b/tests/machines/error/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/error/error_convention_event.py b/tests/machines/error/error_convention_event.py new file mode 100644 index 00000000..72c99ce5 --- /dev/null +++ b/tests/machines/error/error_convention_event.py @@ -0,0 +1,16 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorConventionEventSC(StateChart): + """Using Event without explicit id with error_ prefix auto-registers dot notation.""" + + s1 = State("s1", initial=True) + error_state = State("error_state", final=True) + + go = s1.to(s1, on="bad_action") + error_execution = Event(s1.to(error_state)) + + def bad_action(self): + raise RuntimeError("action failed") diff --git a/tests/machines/error/error_convention_transition_list.py b/tests/machines/error/error_convention_transition_list.py new file mode 100644 index 00000000..f1445387 --- /dev/null +++ b/tests/machines/error/error_convention_transition_list.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class ErrorConventionTransitionListSC(StateChart): + """Using bare TransitionList with error_ prefix auto-registers dot notation.""" + + s1 = State("s1", initial=True) + error_state = State("error_state", final=True) + + go = s1.to(s1, on="bad_action") + error_execution = s1.to(error_state) + + def bad_action(self): + raise RuntimeError("action failed") diff --git a/tests/machines/error/error_in_action_sc.py b/tests/machines/error/error_in_action_sc.py new file mode 100644 index 00000000..b016bba9 --- /dev/null +++ b/tests/machines/error/error_in_action_sc.py @@ -0,0 +1,15 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInActionSC(StateChart): + s1 = State("s1", initial=True) + s2 = State("s2") + error_state = State("error_state", final=True) + + go = s1.to(s2, on="bad_action") + error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") + + def bad_action(self): + raise RuntimeError("action failed") diff --git a/tests/machines/error/error_in_action_sm_with_flag.py b/tests/machines/error/error_in_action_sm_with_flag.py new file mode 100644 index 00000000..95e03435 --- /dev/null +++ b/tests/machines/error/error_in_action_sm_with_flag.py @@ -0,0 +1,17 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInActionSMWithFlag(StateChart): + """StateChart subclass (catch_errors_as_events = True by default).""" + + s1 = State("s1", initial=True) + s2 = State("s2") + error_state = State("error_state", final=True) + + go = s1.to(s2, on="bad_action") + error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") + + def bad_action(self): + raise RuntimeError("action failed") diff --git a/tests/machines/error/error_in_after_sc.py b/tests/machines/error/error_in_after_sc.py new file mode 100644 index 00000000..f507d82c --- /dev/null +++ b/tests/machines/error/error_in_after_sc.py @@ -0,0 +1,15 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInAfterSC(StateChart): + s1 = State("s1", initial=True) + s2 = State("s2") + error_state = State("error_state", final=True) + + go = s1.to(s2, after="bad_after") + error_execution = Event(s2.to(error_state), id="error.execution") + + def bad_after(self): + raise RuntimeError("after failed") diff --git a/tests/machines/error/error_in_error_handler_sc.py b/tests/machines/error/error_in_error_handler_sc.py new file mode 100644 index 00000000..49d5de4e --- /dev/null +++ b/tests/machines/error/error_in_error_handler_sc.py @@ -0,0 +1,24 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInErrorHandlerSC(StateChart): + """Error in error.execution handler should not cause infinite loop.""" + + s1 = State("s1", initial=True) + s2 = State("s2") + s3 = State("s3", final=True) + + go = s1.to(s2, on="bad_action") + finish = s2.to(s3) + error_execution = Event( + s1.to(s1, on="bad_error_handler") | s2.to(s2, on="bad_error_handler"), + id="error.execution", + ) + + def bad_action(self): + raise RuntimeError("action failed") + + def bad_error_handler(self): + raise RuntimeError("error handler also failed") diff --git a/tests/machines/error/error_in_guard_sc.py b/tests/machines/error/error_in_guard_sc.py new file mode 100644 index 00000000..24be427a --- /dev/null +++ b/tests/machines/error/error_in_guard_sc.py @@ -0,0 +1,14 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInGuardSC(StateChart): + initial = State("initial", initial=True) + error_state = State("error_state", final=True) + + go = initial.to(initial, cond="bad_guard") | initial.to(initial) + error_execution = Event(initial.to(error_state), id="error.execution") + + def bad_guard(self): + raise RuntimeError("guard failed") diff --git a/tests/machines/error/error_in_guard_sm.py b/tests/machines/error/error_in_guard_sm.py new file mode 100644 index 00000000..1b622453 --- /dev/null +++ b/tests/machines/error/error_in_guard_sm.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class ErrorInGuardSM(StateChart): + """StateChart subclass with catch_errors_as_events=False: exceptions should propagate.""" + + catch_errors_as_events = False + + initial = State("initial", initial=True) + + go = initial.to(initial, cond="bad_guard") | initial.to(initial) + + def bad_guard(self): + raise RuntimeError("guard failed") diff --git a/tests/machines/error/error_in_on_enter_sc.py b/tests/machines/error/error_in_on_enter_sc.py new file mode 100644 index 00000000..966e9c9b --- /dev/null +++ b/tests/machines/error/error_in_on_enter_sc.py @@ -0,0 +1,15 @@ +from statemachine import Event +from statemachine import State +from statemachine import StateChart + + +class ErrorInOnEnterSC(StateChart): + s1 = State("s1", initial=True) + s2 = State("s2") + error_state = State("error_state", final=True) + + go = s1.to(s2) + error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") + + def on_enter_s2(self): + raise RuntimeError("on_enter failed") diff --git a/tests/machines/eventless/__init__.py b/tests/machines/eventless/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/eventless/auto_advance.py b/tests/machines/eventless/auto_advance.py new file mode 100644 index 00000000..00c5dc56 --- /dev/null +++ b/tests/machines/eventless/auto_advance.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class AutoAdvance(StateChart): + class journey(State.Compound): + step1 = State(initial=True) + step2 = State() + step3 = State(final=True) + + step1.to(step2) + step2.to(step3) + + done = State(final=True) + done_state_journey = journey.to(done) diff --git a/tests/machines/eventless/beacon_chain.py b/tests/machines/eventless/beacon_chain.py new file mode 100644 index 00000000..9747e00a --- /dev/null +++ b/tests/machines/eventless/beacon_chain.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class BeaconChain(StateChart): + class beacons(State.Compound): + first = State(initial=True) + last = State(final=True) + + first.to(last) + + signal_received = State(final=True) + done_state_beacons = beacons.to(signal_received) diff --git a/tests/machines/eventless/beacon_chain_lighting.py b/tests/machines/eventless/beacon_chain_lighting.py new file mode 100644 index 00000000..90771e63 --- /dev/null +++ b/tests/machines/eventless/beacon_chain_lighting.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class BeaconChainLighting(StateChart): + class chain(State.Compound): + amon_din = State(initial=True) + eilenach = State() + nardol = State() + halifirien = State(final=True) + + # Eventless chain: each fires immediately + amon_din.to(eilenach) + eilenach.to(nardol) + nardol.to(halifirien) + + all_lit = State(final=True) + done_state_chain = chain.to(all_lit) diff --git a/tests/machines/eventless/coordinated_advance.py b/tests/machines/eventless/coordinated_advance.py new file mode 100644 index 00000000..c80b97a9 --- /dev/null +++ b/tests/machines/eventless/coordinated_advance.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class CoordinatedAdvance(StateChart): + class forces(State.Parallel): + class vanguard(State.Compound): + waiting = State(initial=True) + advanced = State(final=True) + + move_forward = waiting.to(advanced) + + class rearguard(State.Compound): + holding = State(initial=True) + moved_up = State(final=True) + + # Eventless: advance only when vanguard has advanced + holding.to(moved_up, cond="In('advanced')") diff --git a/tests/machines/eventless/ring_corruption.py b/tests/machines/eventless/ring_corruption.py new file mode 100644 index 00000000..67893553 --- /dev/null +++ b/tests/machines/eventless/ring_corruption.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class RingCorruption(StateChart): + resisting = State(initial=True) + corrupted = State(final=True) + + # eventless: no event name + resisting.to(corrupted, cond="is_corrupted") + + ring_power = 0 + + def is_corrupted(self): + return self.ring_power > 5 + + def increase_power(self): + self.ring_power += 3 diff --git a/tests/machines/eventless/ring_corruption_with_bear_ring.py b/tests/machines/eventless/ring_corruption_with_bear_ring.py new file mode 100644 index 00000000..f211b1fb --- /dev/null +++ b/tests/machines/eventless/ring_corruption_with_bear_ring.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class RingCorruptionWithBearRing(StateChart): + resisting = State(initial=True) + corrupted = State(final=True) + + resisting.to(corrupted, cond="is_corrupted") + bear_ring = resisting.to.itself(internal=True, on="increase_power") + + ring_power = 0 + + def is_corrupted(self): + return self.ring_power > 5 + + def increase_power(self): + self.ring_power += 2 diff --git a/tests/machines/eventless/ring_corruption_with_tick.py b/tests/machines/eventless/ring_corruption_with_tick.py new file mode 100644 index 00000000..4a6f1c0d --- /dev/null +++ b/tests/machines/eventless/ring_corruption_with_tick.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class RingCorruptionWithTick(StateChart): + resisting = State(initial=True) + corrupted = State(final=True) + + resisting.to(corrupted, cond="is_corrupted") + tick = resisting.to.itself(internal=True) + + ring_power = 0 + + def is_corrupted(self): + return self.ring_power > 5 diff --git a/tests/machines/history/__init__.py b/tests/machines/history/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/history/deep_memory_of_moria.py b/tests/machines/history/deep_memory_of_moria.py new file mode 100644 index 00000000..44e15b03 --- /dev/null +++ b/tests/machines/history/deep_memory_of_moria.py @@ -0,0 +1,21 @@ +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart + + +class DeepMemoryOfMoria(StateChart): + class moria(State.Compound): + class halls(State.Compound): + entrance = State(initial=True) + chamber = State() + + explore = entrance.to(chamber) + + assert isinstance(halls, State) + h = HistoryState(type="deep") + bridge = State(final=True) + flee = halls.to(bridge) + + outside = State() + escape = moria.to(outside) + return_deep = outside.to(moria.h) # type: ignore[has-type] diff --git a/tests/machines/history/gollum_personality.py b/tests/machines/history/gollum_personality.py new file mode 100644 index 00000000..8f7a1abb --- /dev/null +++ b/tests/machines/history/gollum_personality.py @@ -0,0 +1,17 @@ +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart + + +class GollumPersonality(StateChart): + class personality(State.Compound): + smeagol = State(initial=True) + gollum = State() + h = HistoryState() + + dark_side = smeagol.to(gollum) + light_side = gollum.to(smeagol) + + outside = State() + leave = personality.to(outside) + return_via_history = outside.to(personality.h) diff --git a/tests/machines/history/gollum_personality_default_gollum.py b/tests/machines/history/gollum_personality_default_gollum.py new file mode 100644 index 00000000..861dcaa3 --- /dev/null +++ b/tests/machines/history/gollum_personality_default_gollum.py @@ -0,0 +1,17 @@ +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart + + +class GollumPersonalityDefaultGollum(StateChart): + class personality(State.Compound): + smeagol = State(initial=True) + gollum = State() + h = HistoryState() + + dark_side = smeagol.to(gollum) + _ = h.to(gollum) # default: gollum (not the initial smeagol) + + outside = State(initial=True) + enter_via_history = outside.to(personality.h) + leave = personality.to(outside) diff --git a/tests/machines/history/gollum_personality_with_default.py b/tests/machines/history/gollum_personality_with_default.py new file mode 100644 index 00000000..41b5d9be --- /dev/null +++ b/tests/machines/history/gollum_personality_with_default.py @@ -0,0 +1,17 @@ +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart + + +class GollumPersonalityWithDefault(StateChart): + class personality(State.Compound): + smeagol = State(initial=True) + gollum = State() + h = HistoryState() + + dark_side = smeagol.to(gollum) + _ = h.to(smeagol) # default: smeagol + + outside = State(initial=True) + enter_via_history = outside.to(personality.h) + leave = personality.to(outside) diff --git a/tests/machines/history/shallow_moria.py b/tests/machines/history/shallow_moria.py new file mode 100644 index 00000000..2b85e43b --- /dev/null +++ b/tests/machines/history/shallow_moria.py @@ -0,0 +1,21 @@ +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart + + +class ShallowMoria(StateChart): + class moria(State.Compound): + class halls(State.Compound): + entrance = State(initial=True) + chamber = State() + + explore = entrance.to(chamber) + + assert isinstance(halls, State) + h = HistoryState() + bridge = State(final=True) + flee = halls.to(bridge) + + outside = State() + escape = moria.to(outside) + return_shallow = outside.to(moria.h) # type: ignore[has-type] diff --git a/tests/machines/in_condition/__init__.py b/tests/machines/in_condition/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/in_condition/combined_guard.py b/tests/machines/in_condition/combined_guard.py new file mode 100644 index 00000000..29f379c1 --- /dev/null +++ b/tests/machines/in_condition/combined_guard.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class CombinedGuard(StateChart): + class positions(State.Parallel): + class scout(State.Compound): + out = State(initial=True) + back = State(final=True) + + return_scout = out.to(back) + + class warrior(State.Compound): + idle = State(initial=True) + attacking = State(final=True) + + # Only attacks when scout is back + charge = idle.to(attacking, cond="In('back')") diff --git a/tests/machines/in_condition/descendant_check.py b/tests/machines/in_condition/descendant_check.py new file mode 100644 index 00000000..354847d8 --- /dev/null +++ b/tests/machines/in_condition/descendant_check.py @@ -0,0 +1,15 @@ +from statemachine import State +from statemachine import StateChart + + +class DescendantCheck(StateChart): + class realm(State.Compound): + village = State(initial=True) + castle = State() + + ascend = village.to(castle) + + conquered = State(final=True) + # Guarded by being inside the castle + conquer = realm.to(conquered, cond="In('castle')") + explore = realm.to.itself(internal=True) # type: ignore[attr-defined] diff --git a/tests/machines/in_condition/eventless_in.py b/tests/machines/in_condition/eventless_in.py new file mode 100644 index 00000000..7ffbdf7c --- /dev/null +++ b/tests/machines/in_condition/eventless_in.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class EventlessIn(StateChart): + class coordination(State.Parallel): + class leader(State.Compound): + planning = State(initial=True) + ready = State(final=True) + + get_ready = planning.to(ready) + + class follower(State.Compound): + waiting = State(initial=True) + moving = State(final=True) + + # Eventless: move when leader is ready + waiting.to(moving, cond="In('ready')") diff --git a/tests/machines/in_condition/fellowship.py b/tests/machines/in_condition/fellowship.py new file mode 100644 index 00000000..9b9e2674 --- /dev/null +++ b/tests/machines/in_condition/fellowship.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class Fellowship(StateChart): + class positions(State.Parallel): + class frodo(State.Compound): + shire_f = State(initial=True) + mordor_f = State(final=True) + + journey = shire_f.to(mordor_f) + + class sam(State.Compound): + shire_s = State(initial=True) + mordor_s = State(final=True) + + # Sam follows Frodo: eventless, guarded by In('mordor_f') + shire_s.to(mordor_s, cond="In('mordor_f')") diff --git a/tests/machines/in_condition/fellowship_coordination.py b/tests/machines/in_condition/fellowship_coordination.py new file mode 100644 index 00000000..87aa81bd --- /dev/null +++ b/tests/machines/in_condition/fellowship_coordination.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class FellowshipCoordination(StateChart): + class mission(State.Parallel): + class scouts(State.Compound): + scouting = State(initial=True) + reported = State(final=True) + + report = scouting.to(reported) + + class army(State.Compound): + waiting = State(initial=True) + marching = State(final=True) + + # Army marches only after scouts report + waiting.to(marching, cond="In('reported')") diff --git a/tests/machines/in_condition/gate_of_moria.py b/tests/machines/in_condition/gate_of_moria.py new file mode 100644 index 00000000..eb01c423 --- /dev/null +++ b/tests/machines/in_condition/gate_of_moria.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class GateOfMoria(StateChart): + outside = State(initial=True) + at_gate = State() + inside = State(final=True) + + approach = outside.to(at_gate) + # Can only enter if we are at the gate + enter_gate = outside.to(inside, cond="In('at_gate')") + speak_friend = at_gate.to(inside) diff --git a/tests/machines/parallel/__init__.py b/tests/machines/parallel/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/parallel/session.py b/tests/machines/parallel/session.py new file mode 100644 index 00000000..78ec6d74 --- /dev/null +++ b/tests/machines/parallel/session.py @@ -0,0 +1,17 @@ +from statemachine import State +from statemachine import StateChart + + +class Session(StateChart): + class session(State.Parallel): + class ui(State.Compound): + active = State(initial=True) + closed = State(final=True) + + close_ui = active.to(closed) + + class backend(State.Compound): + running = State(initial=True) + stopped = State(final=True) + + stop_backend = running.to(stopped) diff --git a/tests/machines/parallel/session_with_done_state.py b/tests/machines/parallel/session_with_done_state.py new file mode 100644 index 00000000..0804e751 --- /dev/null +++ b/tests/machines/parallel/session_with_done_state.py @@ -0,0 +1,20 @@ +from statemachine import State +from statemachine import StateChart + + +class SessionWithDoneState(StateChart): + class session(State.Parallel): + class ui(State.Compound): + active = State(initial=True) + closed = State(final=True) + + close_ui = active.to(closed) + + class backend(State.Compound): + running = State(initial=True) + stopped = State(final=True) + + stop_backend = running.to(stopped) + + finished = State(final=True) + done_state_session = session.to(finished) diff --git a/tests/machines/parallel/two_towers.py b/tests/machines/parallel/two_towers.py new file mode 100644 index 00000000..3abcda59 --- /dev/null +++ b/tests/machines/parallel/two_towers.py @@ -0,0 +1,20 @@ +from statemachine import State +from statemachine import StateChart + + +class TwoTowers(StateChart): + class battle(State.Parallel): + class helms_deep(State.Compound): + fighting = State(initial=True) + victory = State(final=True) + + win = fighting.to(victory) + + class isengard(State.Compound): + besieging = State(initial=True) + flooded = State(final=True) + + flood = besieging.to(flooded) + + aftermath = State(final=True) + done_state_battle = battle.to(aftermath) diff --git a/tests/machines/parallel/war_of_the_ring.py b/tests/machines/parallel/war_of_the_ring.py new file mode 100644 index 00000000..04fb6d2b --- /dev/null +++ b/tests/machines/parallel/war_of_the_ring.py @@ -0,0 +1,25 @@ +from statemachine import State +from statemachine import StateChart + + +class WarOfTheRing(StateChart): + class war(State.Parallel): + class frodos_quest(State.Compound): + shire = State(initial=True) + mordor = State() + mount_doom = State(final=True) + + journey = shire.to(mordor) + destroy_ring = mordor.to(mount_doom) + + class aragorns_path(State.Compound): + ranger = State(initial=True) + king = State(final=True) + + coronation = ranger.to(king) + + class gandalfs_defense(State.Compound): + rohan = State(initial=True) + gondor = State(final=True) + + ride_to_gondor = rohan.to(gondor) diff --git a/tests/machines/parallel/war_with_exit.py b/tests/machines/parallel/war_with_exit.py new file mode 100644 index 00000000..f89b5986 --- /dev/null +++ b/tests/machines/parallel/war_with_exit.py @@ -0,0 +1,20 @@ +from statemachine import State +from statemachine import StateChart + + +class WarWithExit(StateChart): + class war(State.Parallel): + class front_a(State.Compound): + fighting = State(initial=True) + won = State(final=True) + + win_a = fighting.to(won) + + class front_b(State.Compound): + holding = State(initial=True) + held = State(final=True) + + hold_b = holding.to(held) + + peace = State(final=True) + truce = war.to(peace) diff --git a/tests/machines/validators/__init__.py b/tests/machines/validators/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/validators/multi_validator.py b/tests/machines/validators/multi_validator.py new file mode 100644 index 00000000..445c693e --- /dev/null +++ b/tests/machines/validators/multi_validator.py @@ -0,0 +1,19 @@ +from statemachine import State +from statemachine import StateChart + + +class MultiValidator(StateChart): + """Machine with multiple validators — first failure stops the chain.""" + + idle = State(initial=True) + active = State(final=True) + + start = idle.to(active, validators=["check_a", "check_b"]) + + def check_a(self, **kwargs): + if not kwargs.get("a_ok"): + raise ValueError("A failed") + + def check_b(self, **kwargs): + if not kwargs.get("b_ok"): + raise ValueError("B failed") diff --git a/tests/machines/validators/order_validation.py b/tests/machines/validators/order_validation.py new file mode 100644 index 00000000..3b8f749e --- /dev/null +++ b/tests/machines/validators/order_validation.py @@ -0,0 +1,17 @@ +from statemachine import State +from statemachine import StateChart + + +class OrderValidation(StateChart): + """StateChart with catch_errors_as_events=True (the default).""" + + pending = State(initial=True) + confirmed = State() + cancelled = State(final=True) + + confirm = pending.to(confirmed, validators="check_stock") + cancel = confirmed.to(cancelled) + + def check_stock(self, quantity=0, **kwargs): + if quantity <= 0: + raise ValueError("Quantity must be positive") diff --git a/tests/machines/validators/order_validation_no_error_events.py b/tests/machines/validators/order_validation_no_error_events.py new file mode 100644 index 00000000..3e55565b --- /dev/null +++ b/tests/machines/validators/order_validation_no_error_events.py @@ -0,0 +1,19 @@ +from statemachine import State +from statemachine import StateChart + + +class OrderValidationNoErrorEvents(StateChart): + """Same machine but with catch_errors_as_events=False.""" + + catch_errors_as_events = False + + pending = State(initial=True) + confirmed = State() + cancelled = State(final=True) + + confirm = pending.to(confirmed, validators="check_stock") + cancel = confirmed.to(cancelled) + + def check_stock(self, quantity=0, **kwargs): + if quantity <= 0: + raise ValueError("Quantity must be positive") diff --git a/tests/machines/validators/validator_fallthrough.py b/tests/machines/validators/validator_fallthrough.py new file mode 100644 index 00000000..8b0a7ce8 --- /dev/null +++ b/tests/machines/validators/validator_fallthrough.py @@ -0,0 +1,20 @@ +from statemachine import State +from statemachine import StateChart + + +class ValidatorFallthrough(StateChart): + """Machine with multiple transitions for the same event. + + When the first transition's validator rejects, the exception propagates + immediately — the engine does NOT fall through to the next transition. + """ + + idle = State(initial=True) + path_a = State(final=True) + path_b = State(final=True) + + go = idle.to(path_a, validators="must_be_premium") | idle.to(path_b) + + def must_be_premium(self, **kwargs): + if not kwargs.get("premium"): + raise PermissionError("Premium required") diff --git a/tests/machines/validators/validator_with_cond.py b/tests/machines/validators/validator_with_cond.py new file mode 100644 index 00000000..d3351054 --- /dev/null +++ b/tests/machines/validators/validator_with_cond.py @@ -0,0 +1,17 @@ +from statemachine import State +from statemachine import StateChart + + +class ValidatorWithCond(StateChart): + """Machine that combines validators and conditions on the same transition.""" + + idle = State(initial=True) + active = State(final=True) + + start = idle.to(active, validators="check_auth", cond="has_permission") + + has_permission = False + + def check_auth(self, token=None, **kwargs): + if token != "valid": + raise PermissionError("Invalid token") diff --git a/tests/machines/validators/validator_with_error_transition.py b/tests/machines/validators/validator_with_error_transition.py new file mode 100644 index 00000000..d069e2f9 --- /dev/null +++ b/tests/machines/validators/validator_with_error_transition.py @@ -0,0 +1,25 @@ +from statemachine import State +from statemachine import StateChart + + +class ValidatorWithErrorTransition(StateChart): + """Machine with both a validator and an error.execution transition. + + The error.execution transition should NOT be triggered by validator + rejection — only by actual execution errors in actions. + """ + + idle = State(initial=True) + active = State() + error_state = State(final=True) + + start = idle.to(active, validators="check_input") + do_work = active.to.itself(on="risky_action") + error_execution = active.to(error_state) + + def check_input(self, value=None, **kwargs): + if value is None: + raise ValueError("Input required") + + def risky_action(self, **kwargs): + raise RuntimeError("Boom") diff --git a/tests/machines/workflow/__init__.py b/tests/machines/workflow/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/machines/workflow/campaign_machine.py b/tests/machines/workflow/campaign_machine.py new file mode 100644 index 00000000..fddbcc1f --- /dev/null +++ b/tests/machines/workflow/campaign_machine.py @@ -0,0 +1,14 @@ +from statemachine import State +from statemachine import StateChart + + +class CampaignMachine(StateChart): + "A workflow machine" + + draft = State(initial=True) + producing = State("Being produced") + closed = State(final=True) + + add_job = draft.to(draft) | producing.to(producing) + produce = draft.to(producing) + deliver = producing.to(closed) diff --git a/tests/machines/workflow/campaign_machine_with_validator.py b/tests/machines/workflow/campaign_machine_with_validator.py new file mode 100644 index 00000000..29f69049 --- /dev/null +++ b/tests/machines/workflow/campaign_machine_with_validator.py @@ -0,0 +1,18 @@ +from statemachine import State +from statemachine import StateChart + + +class CampaignMachineWithValidator(StateChart): + "A workflow machine" + + draft = State(initial=True) + producing = State("Being produced") + closed = State(final=True) + + add_job = draft.to(draft) | producing.to(producing) + produce = draft.to(producing, validators="can_produce") + deliver = producing.to(closed) + + def can_produce(*args, **kwargs): + if "goods" not in kwargs: + raise LookupError("Goods not found.") diff --git a/tests/machines/workflow/campaign_machine_with_values.py b/tests/machines/workflow/campaign_machine_with_values.py new file mode 100644 index 00000000..6dc6e75a --- /dev/null +++ b/tests/machines/workflow/campaign_machine_with_values.py @@ -0,0 +1,14 @@ +from statemachine import State +from statemachine import StateChart + + +class CampaignMachineWithValues(StateChart): + "A workflow machine" + + draft = State(initial=True, value=1) + producing = State("Being produced", value=2) + closed = State(value=3, final=True) + + add_job = draft.to(draft) | producing.to(producing) + produce = draft.to(producing) + deliver = producing.to(closed) diff --git a/tests/machines/workflow/reverse_traffic_light.py b/tests/machines/workflow/reverse_traffic_light.py new file mode 100644 index 00000000..c41c8681 --- /dev/null +++ b/tests/machines/workflow/reverse_traffic_light.py @@ -0,0 +1,13 @@ +from statemachine import State +from statemachine import StateChart + + +class ReverseTrafficLightMachine(StateChart): + "A traffic light machine" + + green = State(initial=True) + yellow = State() + red = State() + + stop = red.from_(yellow, green, red) + cycle = green.from_(red) | yellow.from_(green) | red.from_(yellow) | red.from_.itself() diff --git a/tests/test_error_execution.py b/tests/test_error_execution.py index 3fb9f8f1..ccda4d89 100644 --- a/tests/test_error_execution.py +++ b/tests/test_error_execution.py @@ -4,101 +4,15 @@ from statemachine import Event from statemachine import State from statemachine import StateChart - - -class ErrorInGuardSC(StateChart): - initial = State("initial", initial=True) - error_state = State("error_state", final=True) - - go = initial.to(initial, cond="bad_guard") | initial.to(initial) - error_execution = Event(initial.to(error_state), id="error.execution") - - def bad_guard(self): - raise RuntimeError("guard failed") - - -class ErrorInOnEnterSC(StateChart): - s1 = State("s1", initial=True) - s2 = State("s2") - error_state = State("error_state", final=True) - - go = s1.to(s2) - error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") - - def on_enter_s2(self): - raise RuntimeError("on_enter failed") - - -class ErrorInActionSC(StateChart): - s1 = State("s1", initial=True) - s2 = State("s2") - error_state = State("error_state", final=True) - - go = s1.to(s2, on="bad_action") - error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") - - def bad_action(self): - raise RuntimeError("action failed") - - -class ErrorInAfterSC(StateChart): - s1 = State("s1", initial=True) - s2 = State("s2") - error_state = State("error_state", final=True) - - go = s1.to(s2, after="bad_after") - error_execution = Event(s2.to(error_state), id="error.execution") - - def bad_after(self): - raise RuntimeError("after failed") - - -class ErrorInGuardSM(StateChart): - """StateChart subclass with catch_errors_as_events=False: exceptions should propagate.""" - - catch_errors_as_events = False - - initial = State("initial", initial=True) - - go = initial.to(initial, cond="bad_guard") | initial.to(initial) - - def bad_guard(self): - raise RuntimeError("guard failed") - - -class ErrorInActionSMWithFlag(StateChart): - """StateChart subclass (catch_errors_as_events = True by default).""" - - s1 = State("s1", initial=True) - s2 = State("s2") - error_state = State("error_state", final=True) - - go = s1.to(s2, on="bad_action") - error_execution = Event(s1.to(error_state) | s2.to(error_state), id="error.execution") - - def bad_action(self): - raise RuntimeError("action failed") - - -class ErrorInErrorHandlerSC(StateChart): - """Error in error.execution handler should not cause infinite loop.""" - - s1 = State("s1", initial=True) - s2 = State("s2") - s3 = State("s3", final=True) - - go = s1.to(s2, on="bad_action") - finish = s2.to(s3) - error_execution = Event( - s1.to(s1, on="bad_error_handler") | s2.to(s2, on="bad_error_handler"), - id="error.execution", - ) - - def bad_action(self): - raise RuntimeError("action failed") - - def bad_error_handler(self): - raise RuntimeError("error handler also failed") +from tests.machines.error.error_convention_event import ErrorConventionEventSC +from tests.machines.error.error_convention_transition_list import ErrorConventionTransitionListSC +from tests.machines.error.error_in_action_sc import ErrorInActionSC +from tests.machines.error.error_in_action_sm_with_flag import ErrorInActionSMWithFlag +from tests.machines.error.error_in_after_sc import ErrorInAfterSC +from tests.machines.error.error_in_error_handler_sc import ErrorInErrorHandlerSC +from tests.machines.error.error_in_guard_sc import ErrorInGuardSC +from tests.machines.error.error_in_guard_sm import ErrorInGuardSM +from tests.machines.error.error_in_on_enter_sc import ErrorInOnEnterSC def test_exception_in_guard_sends_error_execution(): @@ -229,32 +143,6 @@ def handle_error(self, error=None, **kwargs): # --- Tests for error_ naming convention --- -class ErrorConventionTransitionListSC(StateChart): - """Using bare TransitionList with error_ prefix auto-registers dot notation.""" - - s1 = State("s1", initial=True) - error_state = State("error_state", final=True) - - go = s1.to(s1, on="bad_action") - error_execution = s1.to(error_state) - - def bad_action(self): - raise RuntimeError("action failed") - - -class ErrorConventionEventSC(StateChart): - """Using Event without explicit id with error_ prefix auto-registers dot notation.""" - - s1 = State("s1", initial=True) - error_state = State("error_state", final=True) - - go = s1.to(s1, on="bad_action") - error_execution = Event(s1.to(error_state)) - - def bad_action(self): - raise RuntimeError("action failed") - - def test_error_convention_with_transition_list(): """Bare TransitionList with error_ prefix matches error.execution event.""" sm = ErrorConventionTransitionListSC() diff --git a/tests/test_mixins.py b/tests/test_mixins.py index 938c710e..f1ae83b9 100644 --- a/tests/test_mixins.py +++ b/tests/test_mixins.py @@ -5,7 +5,7 @@ class MyMixedModel(MyModel, MachineMixin): - state_machine_name = "tests.conftest.CampaignMachine" + state_machine_name = "tests.machines.workflow.campaign_machine.CampaignMachine" def test_mixin_should_instantiate_a_machine(campaign_machine): diff --git a/tests/test_statechart_compound.py b/tests/test_statechart_compound.py index 49757ae9..334b4f28 100644 --- a/tests/test_statechart_compound.py +++ b/tests/test_statechart_compound.py @@ -11,39 +11,26 @@ from statemachine import State from statemachine import StateChart +from tests.machines.compound.middle_earth_journey import MiddleEarthJourney +from tests.machines.compound.middle_earth_journey_two_compounds import ( + MiddleEarthJourneyTwoCompounds, +) +from tests.machines.compound.middle_earth_journey_with_finals import MiddleEarthJourneyWithFinals +from tests.machines.compound.moria_expedition import MoriaExpedition +from tests.machines.compound.moria_expedition_with_escape import MoriaExpeditionWithEscape +from tests.machines.compound.quest_for_erebor import QuestForErebor +from tests.machines.compound.shire_to_rivendell import ShireToRivendell @pytest.mark.timeout(5) class TestCompoundStates: async def test_enter_compound_activates_initial_child(self, sm_runner): """Entering a compound activates both parent and the initial child.""" - - class ShireToRivendell(StateChart): - class shire(State.Compound): - bag_end = State(initial=True) - green_dragon = State() - - visit_pub = bag_end.to(green_dragon) - - road = State(final=True) - depart = shire.to(road) - sm = await sm_runner.start(ShireToRivendell) assert {"shire", "bag_end"} == set(sm.configuration_values) async def test_transition_within_compound(self, sm_runner): """Inner state changes while parent stays active.""" - - class ShireToRivendell(StateChart): - class shire(State.Compound): - bag_end = State(initial=True) - green_dragon = State() - - visit_pub = bag_end.to(green_dragon) - - road = State(final=True) - depart = shire.to(road) - sm = await sm_runner.start(ShireToRivendell) await sm_runner.send(sm, "visit_pub") assert "shire" in sm.configuration_values @@ -52,86 +39,23 @@ class shire(State.Compound): async def test_exit_compound_removes_all_descendants(self, sm_runner): """Leaving a compound removes the parent and all children.""" - - class ShireToRivendell(StateChart): - class shire(State.Compound): - bag_end = State(initial=True) - green_dragon = State() - - visit_pub = bag_end.to(green_dragon) - - road = State(final=True) - depart = shire.to(road) - sm = await sm_runner.start(ShireToRivendell) await sm_runner.send(sm, "depart") assert {"road"} == set(sm.configuration_values) async def test_nested_compound_two_levels(self, sm_runner): """Three-level nesting: outer > middle > leaf.""" - - class MoriaExpedition(StateChart): - class moria(State.Compound): - class upper_halls(State.Compound): - entrance = State(initial=True) - bridge = State(final=True) - - cross = entrance.to(bridge) - - assert isinstance(upper_halls, State) - depths = State(final=True) - descend = upper_halls.to(depths) - sm = await sm_runner.start(MoriaExpedition) assert {"moria", "upper_halls", "entrance"} == set(sm.configuration_values) async def test_transition_from_inner_to_outer(self, sm_runner): """A deep child can transition to an outer state.""" - - class MoriaExpedition(StateChart): - class moria(State.Compound): - class upper_halls(State.Compound): - entrance = State(initial=True) - bridge = State() - - cross = entrance.to(bridge) - - assert isinstance(upper_halls, State) - depths = State(final=True) - descend = upper_halls.to(depths) - - daylight = State(final=True) - escape = moria.to(daylight) - - sm = await sm_runner.start(MoriaExpedition) + sm = await sm_runner.start(MoriaExpeditionWithEscape) await sm_runner.send(sm, "escape") assert {"daylight"} == set(sm.configuration_values) async def test_cross_compound_transition(self, sm_runner): """Transition from one compound to another removes old children.""" - - class MiddleEarthJourney(StateChart): - class rivendell(State.Compound): - council = State(initial=True) - preparing = State() - - get_ready = council.to(preparing) - - class moria(State.Compound): - gates = State(initial=True) - bridge = State(final=True) - - cross = gates.to(bridge) - - class lothlorien(State.Compound): - mirror = State(initial=True) - departure = State(final=True) - - leave = mirror.to(departure) - - march_to_moria = rivendell.to(moria) - march_to_lorien = moria.to(lothlorien) - sm = await sm_runner.start(MiddleEarthJourney) assert "rivendell" in sm.configuration_values assert "council" in sm.configuration_values @@ -144,40 +68,13 @@ class lothlorien(State.Compound): async def test_enter_compound_lands_on_initial(self, sm_runner): """Entering a compound from outside lands on the initial child.""" - - class MiddleEarthJourney(StateChart): - class rivendell(State.Compound): - council = State(initial=True) - preparing = State() - - get_ready = council.to(preparing) - - class moria(State.Compound): - gates = State(initial=True) - bridge = State(final=True) - - cross = gates.to(bridge) - - march_to_moria = rivendell.to(moria) - - sm = await sm_runner.start(MiddleEarthJourney) + sm = await sm_runner.start(MiddleEarthJourneyTwoCompounds) await sm_runner.send(sm, "march_to_moria") assert "gates" in sm.configuration_values assert "moria" in sm.configuration_values async def test_final_child_fires_done_state(self, sm_runner): """Reaching a final child triggers done.state.{parent_id}.""" - - class QuestForErebor(StateChart): - class lonely_mountain(State.Compound): - approach = State(initial=True) - inside = State(final=True) - - enter_mountain = approach.to(inside) - - victory = State(final=True) - done_state_lonely_mountain = lonely_mountain.to(victory) - sm = await sm_runner.start(QuestForErebor) assert "approach" in sm.configuration_values @@ -186,30 +83,7 @@ class lonely_mountain(State.Compound): async def test_multiple_compound_sequential_traversal(self, sm_runner): """Traverse all three compounds sequentially.""" - - class MiddleEarthJourney(StateChart): - class rivendell(State.Compound): - council = State(initial=True) - preparing = State(final=True) - - get_ready = council.to(preparing) - - class moria(State.Compound): - gates = State(initial=True) - bridge = State(final=True) - - cross = gates.to(bridge) - - class lothlorien(State.Compound): - mirror = State(initial=True) - departure = State(final=True) - - leave = mirror.to(departure) - - march_to_moria = rivendell.to(moria) - march_to_lorien = moria.to(lothlorien) - - sm = await sm_runner.start(MiddleEarthJourney) + sm = await sm_runner.start(MiddleEarthJourneyWithFinals) await sm_runner.send(sm, "march_to_moria") assert "moria" in sm.configuration_values diff --git a/tests/test_statechart_donedata.py b/tests/test_statechart_donedata.py index ca191361..60acd02a 100644 --- a/tests/test_statechart_donedata.py +++ b/tests/test_statechart_donedata.py @@ -13,77 +13,32 @@ from statemachine import Event from statemachine import State from statemachine import StateChart +from tests.machines.donedata.destroy_the_ring import DestroyTheRing +from tests.machines.donedata.destroy_the_ring_simple import DestroyTheRingSimple +from tests.machines.donedata.nested_quest_donedata import NestedQuestDoneData +from tests.machines.donedata.quest_for_erebor_done_convention import QuestForEreborDoneConvention +from tests.machines.donedata.quest_for_erebor_explicit_id import QuestForEreborExplicitId +from tests.machines.donedata.quest_for_erebor_multi_word import QuestForEreborMultiWord +from tests.machines.donedata.quest_for_erebor_with_event import QuestForEreborWithEvent @pytest.mark.timeout(5) class TestDoneData: async def test_donedata_callable_returns_dict(self, sm_runner): """Handler receives donedata as kwargs.""" - received = {} - - class DestroyTheRing(StateChart): - class quest(State.Compound): - traveling = State(initial=True) - completed = State(final=True, donedata="get_quest_result") - - finish = traveling.to(completed) - - def get_quest_result(self): - return {"ring_destroyed": True, "hero": "frodo"} - - epilogue = State(final=True) - done_state_quest = Event(quest.to(epilogue, on="capture_result")) - - def capture_result(self, ring_destroyed=None, hero=None, **kwargs): - received["ring_destroyed"] = ring_destroyed - received["hero"] = hero - sm = await sm_runner.start(DestroyTheRing) await sm_runner.send(sm, "finish") - assert received["ring_destroyed"] is True - assert received["hero"] == "frodo" + assert sm.received["ring_destroyed"] is True + assert sm.received["hero"] == "frodo" async def test_donedata_fires_done_state_with_data(self, sm_runner): """done.state event fires and triggers a transition.""" - - class DestroyTheRing(StateChart): - class quest(State.Compound): - traveling = State(initial=True) - completed = State(final=True, donedata="get_result") - - finish = traveling.to(completed) - - def get_result(self): - return {"outcome": "victory"} - - celebration = State(final=True) - done_state_quest = Event(quest.to(celebration)) - - sm = await sm_runner.start(DestroyTheRing) + sm = await sm_runner.start(DestroyTheRingSimple) await sm_runner.send(sm, "finish") assert {"celebration"} == set(sm.configuration_values) async def test_donedata_in_nested_compound(self, sm_runner): """Inner done.state propagates up through nesting.""" - - class NestedQuestDoneData(StateChart): - class outer(State.Compound): - class inner(State.Compound): - start = State(initial=True) - end = State(final=True, donedata="inner_result") - - go = start.to(end) - - def inner_result(self): - return {"level": "inner"} - - assert isinstance(inner, State) - after_inner = State(final=True) - done_state_inner = Event(inner.to(after_inner)) - - final = State(final=True) - done_state_outer = Event(outer.to(final)) - sm = await sm_runner.start(NestedQuestDoneData) await sm_runner.send(sm, "go") # inner finishes -> done.state.inner -> after_inner (final) @@ -108,7 +63,7 @@ class QuestListener: def on_enter_celebration(self, ring_destroyed=None, **kwargs): captured["ring_destroyed"] = ring_destroyed - class DestroyTheRing(StateChart): + class DestroyTheRingWithListener(StateChart): class quest(State.Compound): traveling = State(initial=True) completed = State(final=True, donedata="get_result") @@ -122,7 +77,7 @@ def get_result(self): done_state_quest = Event(quest.to(celebration)) listener = QuestListener() - sm = await sm_runner.start(DestroyTheRing, listeners=[listener]) + sm = await sm_runner.start(DestroyTheRingWithListener, listeners=[listener]) await sm_runner.send(sm, "finish") assert {"celebration"} == set(sm.configuration_values) @@ -131,68 +86,24 @@ def get_result(self): class TestDoneStateConvention: async def test_done_state_convention_with_transition_list(self, sm_runner): """Bare TransitionList with done_state_ name auto-registers done.state.X.""" - - class QuestForErebor(StateChart): - class quest(State.Compound): - traveling = State(initial=True) - arrived = State(final=True) - - finish = traveling.to(arrived) - - celebration = State(final=True) - done_state_quest = quest.to(celebration) - - sm = await sm_runner.start(QuestForErebor) + sm = await sm_runner.start(QuestForEreborDoneConvention) await sm_runner.send(sm, "finish") assert {"celebration"} == set(sm.configuration_values) async def test_done_state_convention_with_event_no_explicit_id(self, sm_runner): """Event() wrapper without explicit id= applies the convention.""" - - class QuestForErebor(StateChart): - class quest(State.Compound): - traveling = State(initial=True) - arrived = State(final=True) - - finish = traveling.to(arrived) - - celebration = State(final=True) - done_state_quest = Event(quest.to(celebration)) - - sm = await sm_runner.start(QuestForErebor) + sm = await sm_runner.start(QuestForEreborWithEvent) await sm_runner.send(sm, "finish") assert {"celebration"} == set(sm.configuration_values) async def test_done_state_convention_preserves_explicit_id(self, sm_runner): """Explicit id= takes precedence over the convention.""" - - class QuestForErebor(StateChart): - class quest(State.Compound): - traveling = State(initial=True) - arrived = State(final=True) - - finish = traveling.to(arrived) - - celebration = State(final=True) - done_state_quest = Event(quest.to(celebration), id="done.state.quest") - - sm = await sm_runner.start(QuestForErebor) + sm = await sm_runner.start(QuestForEreborExplicitId) await sm_runner.send(sm, "finish") assert {"celebration"} == set(sm.configuration_values) async def test_done_state_convention_with_multi_word_state(self, sm_runner): """done_state_lonely_mountain maps to done.state.lonely_mountain.""" - - class QuestForErebor(StateChart): - class lonely_mountain(State.Compound): - approach = State(initial=True) - inside = State(final=True) - - enter_mountain = approach.to(inside) - - victory = State(final=True) - done_state_lonely_mountain = lonely_mountain.to(victory) - - sm = await sm_runner.start(QuestForErebor) + sm = await sm_runner.start(QuestForEreborMultiWord) await sm_runner.send(sm, "enter_mountain") assert {"victory"} == set(sm.configuration_values) diff --git a/tests/test_statechart_eventless.py b/tests/test_statechart_eventless.py index 4e69eb68..ab75f37e 100644 --- a/tests/test_statechart_eventless.py +++ b/tests/test_statechart_eventless.py @@ -9,30 +9,19 @@ import pytest -from statemachine import State -from statemachine import StateChart +from tests.machines.eventless.auto_advance import AutoAdvance +from tests.machines.eventless.beacon_chain import BeaconChain +from tests.machines.eventless.beacon_chain_lighting import BeaconChainLighting +from tests.machines.eventless.coordinated_advance import CoordinatedAdvance +from tests.machines.eventless.ring_corruption import RingCorruption +from tests.machines.eventless.ring_corruption_with_bear_ring import RingCorruptionWithBearRing +from tests.machines.eventless.ring_corruption_with_tick import RingCorruptionWithTick @pytest.mark.timeout(5) class TestEventlessTransitions: async def test_eventless_fires_when_condition_met(self, sm_runner): """Eventless transition fires when guard is True.""" - - class RingCorruption(StateChart): - resisting = State(initial=True) - corrupted = State(final=True) - - # eventless: no event name - resisting.to(corrupted, cond="is_corrupted") - - ring_power = 0 - - def is_corrupted(self): - return self.ring_power > 5 - - def increase_power(self): - self.ring_power += 3 - sm = await sm_runner.start(RingCorruption) assert "resisting" in sm.configuration_values @@ -43,65 +32,20 @@ def increase_power(self): async def test_eventless_does_not_fire_when_condition_false(self, sm_runner): """Eventless transition stays when guard is False.""" - - class RingCorruption(StateChart): - resisting = State(initial=True) - corrupted = State(final=True) - - resisting.to(corrupted, cond="is_corrupted") - tick = resisting.to.itself(internal=True) - - ring_power = 0 - - def is_corrupted(self): - return self.ring_power > 5 - - sm = await sm_runner.start(RingCorruption) + sm = await sm_runner.start(RingCorruptionWithTick) sm.ring_power = 2 await sm_runner.send(sm, "tick") assert "resisting" in sm.configuration_values async def test_eventless_chain_cascades(self, sm_runner): """All beacons light in a single macrostep via unconditional eventless chain.""" - - class BeaconChainLighting(StateChart): - class chain(State.Compound): - amon_din = State(initial=True) - eilenach = State() - nardol = State() - halifirien = State(final=True) - - # Eventless chain: each fires immediately - amon_din.to(eilenach) - eilenach.to(nardol) - nardol.to(halifirien) - - all_lit = State(final=True) - done_state_chain = chain.to(all_lit) - sm = await sm_runner.start(BeaconChainLighting) # The chain should cascade through all states in a single macrostep assert {"all_lit"} == set(sm.configuration_values) async def test_eventless_gradual_condition(self, sm_runner): """Multiple events needed before the condition threshold is met.""" - - class RingCorruption(StateChart): - resisting = State(initial=True) - corrupted = State(final=True) - - resisting.to(corrupted, cond="is_corrupted") - bear_ring = resisting.to.itself(internal=True, on="increase_power") - - ring_power = 0 - - def is_corrupted(self): - return self.ring_power > 5 - - def increase_power(self): - self.ring_power += 2 - - sm = await sm_runner.start(RingCorruption) + sm = await sm_runner.start(RingCorruptionWithBearRing) await sm_runner.send(sm, "bear_ring") # power = 2 assert "resisting" in sm.configuration_values @@ -113,41 +57,12 @@ def increase_power(self): async def test_eventless_in_compound_state(self, sm_runner): """Eventless transition between compound children.""" - - class AutoAdvance(StateChart): - class journey(State.Compound): - step1 = State(initial=True) - step2 = State() - step3 = State(final=True) - - step1.to(step2) - step2.to(step3) - - done = State(final=True) - done_state_journey = journey.to(done) - sm = await sm_runner.start(AutoAdvance) # Eventless chain cascades through all children assert {"done"} == set(sm.configuration_values) async def test_eventless_with_in_condition(self, sm_runner): """Eventless transition guarded by In('state_id').""" - - class CoordinatedAdvance(StateChart): - class forces(State.Parallel): - class vanguard(State.Compound): - waiting = State(initial=True) - advanced = State(final=True) - - move_forward = waiting.to(advanced) - - class rearguard(State.Compound): - holding = State(initial=True) - moved_up = State(final=True) - - # Eventless: advance only when vanguard has advanced - holding.to(moved_up, cond="In('advanced')") - sm = await sm_runner.start(CoordinatedAdvance) assert "waiting" in sm.configuration_values @@ -159,16 +74,5 @@ class rearguard(State.Compound): async def test_eventless_chain_with_final_triggers_done(self, sm_runner): """Eventless chain reaches final state -> done.state fires.""" - - class BeaconChain(StateChart): - class beacons(State.Compound): - first = State(initial=True) - last = State(final=True) - - first.to(last) - - signal_received = State(final=True) - done_state_beacons = beacons.to(signal_received) - sm = await sm_runner.start(BeaconChain) assert {"signal_received"} == set(sm.configuration_values) diff --git a/tests/test_statechart_history.py b/tests/test_statechart_history.py index 3ed35fe0..2e1d1923 100644 --- a/tests/test_statechart_history.py +++ b/tests/test_statechart_history.py @@ -9,29 +9,17 @@ import pytest -from statemachine import HistoryState -from statemachine import State -from statemachine import StateChart +from tests.machines.history.deep_memory_of_moria import DeepMemoryOfMoria +from tests.machines.history.gollum_personality import GollumPersonality +from tests.machines.history.gollum_personality_default_gollum import GollumPersonalityDefaultGollum +from tests.machines.history.gollum_personality_with_default import GollumPersonalityWithDefault +from tests.machines.history.shallow_moria import ShallowMoria @pytest.mark.timeout(5) class TestHistoryStates: async def test_shallow_history_remembers_last_child(self, sm_runner): """Exit compound, re-enter via history -> restores last active child.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - light_side = gollum.to(smeagol) - - outside = State() - leave = personality.to(outside) - return_via_history = outside.to(personality.h) - sm = await sm_runner.start(GollumPersonality) await sm_runner.send(sm, "dark_side") assert "gollum" in sm.configuration_values @@ -45,21 +33,7 @@ class personality(State.Compound): async def test_shallow_history_default_on_first_visit(self, sm_runner): """No prior visit -> history uses default transition target.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - _ = h.to(smeagol) # default: smeagol - - outside = State(initial=True) - enter_via_history = outside.to(personality.h) - leave = personality.to(outside) - - sm = await sm_runner.start(GollumPersonality) + sm = await sm_runner.start(GollumPersonalityWithDefault) assert {"outside"} == set(sm.configuration_values) await sm_runner.send(sm, "enter_via_history") @@ -67,24 +41,6 @@ class personality(State.Compound): async def test_deep_history_remembers_full_descendant(self, sm_runner): """Deep history restores the exact leaf in a nested compound.""" - - class DeepMemoryOfMoria(StateChart): - class moria(State.Compound): - class halls(State.Compound): - entrance = State(initial=True) - chamber = State() - - explore = entrance.to(chamber) - - assert isinstance(halls, State) - h = HistoryState(type="deep") - bridge = State(final=True) - flee = halls.to(bridge) - - outside = State() - escape = moria.to(outside) - return_deep = outside.to(moria.h) - sm = await sm_runner.start(DeepMemoryOfMoria) await sm_runner.send(sm, "explore") assert "chamber" in sm.configuration_values @@ -99,20 +55,6 @@ class halls(State.Compound): async def test_multiple_exits_and_reentries(self, sm_runner): """History updates each time we exit the compound.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - light_side = gollum.to(smeagol) - - outside = State() - leave = personality.to(outside) - return_via_history = outside.to(personality.h) - sm = await sm_runner.start(GollumPersonality) await sm_runner.send(sm, "leave") await sm_runner.send(sm, "return_via_history") @@ -130,19 +72,6 @@ class personality(State.Compound): async def test_history_after_state_change(self, sm_runner): """Change state within compound, exit, re-enter -> new state restored.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - - outside = State() - leave = personality.to(outside) - return_via_history = outside.to(personality.h) - sm = await sm_runner.start(GollumPersonality) await sm_runner.send(sm, "dark_side") await sm_runner.send(sm, "leave") @@ -151,24 +80,6 @@ class personality(State.Compound): async def test_shallow_only_remembers_immediate_child(self, sm_runner): """Shallow history in nested compound restores direct child, not grandchild.""" - - class ShallowMoria(StateChart): - class moria(State.Compound): - class halls(State.Compound): - entrance = State(initial=True) - chamber = State() - - explore = entrance.to(chamber) - - assert isinstance(halls, State) - h = HistoryState() - bridge = State(final=True) - flee = halls.to(bridge) - - outside = State() - escape = moria.to(outside) - return_shallow = outside.to(moria.h) - sm = await sm_runner.start(ShallowMoria) await sm_runner.send(sm, "explore") assert "chamber" in sm.configuration_values @@ -182,19 +93,6 @@ class halls(State.Compound): async def test_history_values_dict_populated(self, sm_runner): """sm.history_values[history_id] has saved states after exit.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - - outside = State() - leave = personality.to(outside) - return_via_history = outside.to(personality.h) - sm = await sm_runner.start(GollumPersonality) await sm_runner.send(sm, "dark_side") await sm_runner.send(sm, "leave") @@ -205,20 +103,6 @@ class personality(State.Compound): async def test_history_with_default_transition(self, sm_runner): """HistoryState with explicit default .to() transition.""" - - class GollumPersonality(StateChart): - class personality(State.Compound): - smeagol = State(initial=True) - gollum = State() - h = HistoryState() - - dark_side = smeagol.to(gollum) - _ = h.to(gollum) # default: gollum (not the initial smeagol) - - outside = State(initial=True) - enter_via_history = outside.to(personality.h) - leave = personality.to(outside) - - sm = await sm_runner.start(GollumPersonality) + sm = await sm_runner.start(GollumPersonalityDefaultGollum) await sm_runner.send(sm, "enter_via_history") assert "gollum" in sm.configuration_values diff --git a/tests/test_statechart_in_condition.py b/tests/test_statechart_in_condition.py index 593ea6c4..f9dba574 100644 --- a/tests/test_statechart_in_condition.py +++ b/tests/test_statechart_in_condition.py @@ -9,30 +9,18 @@ import pytest -from statemachine import State -from statemachine import StateChart +from tests.machines.in_condition.combined_guard import CombinedGuard +from tests.machines.in_condition.descendant_check import DescendantCheck +from tests.machines.in_condition.eventless_in import EventlessIn +from tests.machines.in_condition.fellowship import Fellowship +from tests.machines.in_condition.fellowship_coordination import FellowshipCoordination +from tests.machines.in_condition.gate_of_moria import GateOfMoria @pytest.mark.timeout(5) class TestInCondition: async def test_in_condition_true_enables_transition(self, sm_runner): """In('state_id') when state is active -> transition fires.""" - - class Fellowship(StateChart): - class positions(State.Parallel): - class frodo(State.Compound): - shire_f = State(initial=True) - mordor_f = State(final=True) - - journey = shire_f.to(mordor_f) - - class sam(State.Compound): - shire_s = State(initial=True) - mordor_s = State(final=True) - - # Sam follows Frodo: eventless, guarded by In('mordor_f') - shire_s.to(mordor_s, cond="In('mordor_f')") - sm = await sm_runner.start(Fellowship) await sm_runner.send(sm, "journey") vals = set(sm.configuration_values) @@ -41,39 +29,12 @@ class sam(State.Compound): async def test_in_condition_false_blocks_transition(self, sm_runner): """In('state_id') when state is not active -> transition blocked.""" - - class GateOfMoria(StateChart): - outside = State(initial=True) - at_gate = State() - inside = State(final=True) - - approach = outside.to(at_gate) - # Can only enter if we are at the gate - enter_gate = outside.to(inside, cond="In('at_gate')") - speak_friend = at_gate.to(inside) - sm = await sm_runner.start(GateOfMoria) await sm_runner.send(sm, "enter_gate") assert "outside" in sm.configuration_values async def test_in_with_parallel_regions(self, sm_runner): """Cross-region In() evaluation in parallel states.""" - - class FellowshipCoordination(StateChart): - class mission(State.Parallel): - class scouts(State.Compound): - scouting = State(initial=True) - reported = State(final=True) - - report = scouting.to(reported) - - class army(State.Compound): - waiting = State(initial=True) - marching = State(final=True) - - # Army marches only after scouts report - waiting.to(marching, cond="In('reported')") - sm = await sm_runner.start(FellowshipCoordination) vals = set(sm.configuration_values) assert "waiting" in vals @@ -86,19 +47,6 @@ class army(State.Compound): async def test_in_with_compound_descendant(self, sm_runner): """In('child') when child is an active descendant.""" - - class DescendantCheck(StateChart): - class realm(State.Compound): - village = State(initial=True) - castle = State() - - ascend = village.to(castle) - - conquered = State(final=True) - # Guarded by being inside the castle - conquer = realm.to(conquered, cond="In('castle')") - explore = realm.to.itself(internal=True) - sm = await sm_runner.start(DescendantCheck) await sm_runner.send(sm, "conquer") assert "realm" in sm.configuration_values @@ -111,22 +59,6 @@ class realm(State.Compound): async def test_in_combined_with_event(self, sm_runner): """Event + In() guard together.""" - - class CombinedGuard(StateChart): - class positions(State.Parallel): - class scout(State.Compound): - out = State(initial=True) - back = State(final=True) - - return_scout = out.to(back) - - class warrior(State.Compound): - idle = State(initial=True) - attacking = State(final=True) - - # Only attacks when scout is back - charge = idle.to(attacking, cond="In('back')") - sm = await sm_runner.start(CombinedGuard) await sm_runner.send(sm, "charge") assert "idle" in sm.configuration_values @@ -137,22 +69,6 @@ class warrior(State.Compound): async def test_in_with_eventless_transition(self, sm_runner): """Eventless + In() guard.""" - - class EventlessIn(StateChart): - class coordination(State.Parallel): - class leader(State.Compound): - planning = State(initial=True) - ready = State(final=True) - - get_ready = planning.to(ready) - - class follower(State.Compound): - waiting = State(initial=True) - moving = State(final=True) - - # Eventless: move when leader is ready - waiting.to(moving, cond="In('ready')") - sm = await sm_runner.start(EventlessIn) assert "waiting" in sm.configuration_values diff --git a/tests/test_statechart_parallel.py b/tests/test_statechart_parallel.py index 6e87d42e..619d1b36 100644 --- a/tests/test_statechart_parallel.py +++ b/tests/test_statechart_parallel.py @@ -9,41 +9,18 @@ import pytest -from statemachine import State -from statemachine import StateChart +from tests.machines.parallel.session import Session +from tests.machines.parallel.session_with_done_state import SessionWithDoneState +from tests.machines.parallel.two_towers import TwoTowers +from tests.machines.parallel.war_of_the_ring import WarOfTheRing +from tests.machines.parallel.war_with_exit import WarWithExit @pytest.mark.timeout(5) class TestParallelStates: - @pytest.fixture() - def war_of_the_ring_cls(self): - class WarOfTheRing(StateChart): - class war(State.Parallel): - class frodos_quest(State.Compound): - shire = State(initial=True) - mordor = State() - mount_doom = State(final=True) - - journey = shire.to(mordor) - destroy_ring = mordor.to(mount_doom) - - class aragorns_path(State.Compound): - ranger = State(initial=True) - king = State(final=True) - - coronation = ranger.to(king) - - class gandalfs_defense(State.Compound): - rohan = State(initial=True) - gondor = State(final=True) - - ride_to_gondor = rohan.to(gondor) - - return WarOfTheRing - - async def test_parallel_activates_all_regions(self, sm_runner, war_of_the_ring_cls): + async def test_parallel_activates_all_regions(self, sm_runner): """Entering a parallel state activates the initial child of every region.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) vals = set(sm.configuration_values) assert "war" in vals assert "frodos_quest" in vals @@ -53,18 +30,18 @@ async def test_parallel_activates_all_regions(self, sm_runner, war_of_the_ring_c assert "gandalfs_defense" in vals assert "rohan" in vals - async def test_independent_transitions_in_regions(self, sm_runner, war_of_the_ring_cls): + async def test_independent_transitions_in_regions(self, sm_runner): """An event in one region does not affect others.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) await sm_runner.send(sm, "journey") vals = set(sm.configuration_values) assert "mordor" in vals assert "ranger" in vals # unchanged assert "rohan" in vals # unchanged - async def test_configuration_includes_all_active_states(self, sm_runner, war_of_the_ring_cls): + async def test_configuration_includes_all_active_states(self, sm_runner): """Configuration set includes all active states across regions.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) config_ids = {s.id for s in sm.configuration} assert config_ids == { "war", @@ -78,48 +55,30 @@ async def test_configuration_includes_all_active_states(self, sm_runner, war_of_ async def test_exit_parallel_exits_all_regions(self, sm_runner): """Transition out of a parallel clears everything.""" - - class WarWithExit(StateChart): - class war(State.Parallel): - class front_a(State.Compound): - fighting = State(initial=True) - won = State(final=True) - - win_a = fighting.to(won) - - class front_b(State.Compound): - holding = State(initial=True) - held = State(final=True) - - hold_b = holding.to(held) - - peace = State(final=True) - truce = war.to(peace) - sm = await sm_runner.start(WarWithExit) assert "war" in sm.configuration_values await sm_runner.send(sm, "truce") assert {"peace"} == set(sm.configuration_values) - async def test_event_in_one_region_no_effect_on_others(self, sm_runner, war_of_the_ring_cls): + async def test_event_in_one_region_no_effect_on_others(self, sm_runner): """Region isolation: events affect only the targeted region.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) await sm_runner.send(sm, "coronation") vals = set(sm.configuration_values) assert "king" in vals assert "shire" in vals # Frodo's region unchanged assert "rohan" in vals # Gandalf's region unchanged - async def test_parallel_with_compound_children(self, sm_runner, war_of_the_ring_cls): + async def test_parallel_with_compound_children(self, sm_runner): """Mixed hierarchy: parallel with compound regions verified.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) assert "shire" in sm.configuration_values assert "ranger" in sm.configuration_values assert "rohan" in sm.configuration_values - async def test_current_state_value_set_comparison(self, sm_runner, war_of_the_ring_cls): + async def test_current_state_value_set_comparison(self, sm_runner): """configuration_values supports set comparison for parallel states.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) vals = set(sm.configuration_values) expected = { "war", @@ -134,24 +93,6 @@ async def test_current_state_value_set_comparison(self, sm_runner, war_of_the_ri async def test_parallel_done_when_all_regions_final(self, sm_runner): """done.state fires when ALL regions reach a final state.""" - - class TwoTowers(StateChart): - class battle(State.Parallel): - class helms_deep(State.Compound): - fighting = State(initial=True) - victory = State(final=True) - - win = fighting.to(victory) - - class isengard(State.Compound): - besieging = State(initial=True) - flooded = State(final=True) - - flood = besieging.to(flooded) - - aftermath = State(final=True) - done_state_battle = battle.to(aftermath) - sm = await sm_runner.start(TwoTowers) await sm_runner.send(sm, "win") # Only one region is final, battle continues @@ -163,35 +104,15 @@ class isengard(State.Compound): async def test_parallel_not_done_when_one_region_final(self, sm_runner): """Parallel not done when only one region reaches final.""" - - class TwoTowers(StateChart): - class battle(State.Parallel): - class helms_deep(State.Compound): - fighting = State(initial=True) - victory = State(final=True) - - win = fighting.to(victory) - - class isengard(State.Compound): - besieging = State(initial=True) - flooded = State(final=True) - - flood = besieging.to(flooded) - - aftermath = State(final=True) - done_state_battle = battle.to(aftermath) - sm = await sm_runner.start(TwoTowers) await sm_runner.send(sm, "win") assert "battle" in sm.configuration_values assert "victory" in sm.configuration_values assert "besieging" in sm.configuration_values - async def test_transition_within_compound_inside_parallel( - self, sm_runner, war_of_the_ring_cls - ): + async def test_transition_within_compound_inside_parallel(self, sm_runner): """Deep transition within a compound region of a parallel state.""" - sm = await sm_runner.start(war_of_the_ring_cls) + sm = await sm_runner.start(WarOfTheRing) await sm_runner.send(sm, "journey") await sm_runner.send(sm, "destroy_ring") vals = set(sm.configuration_values) @@ -200,21 +121,6 @@ async def test_transition_within_compound_inside_parallel( async def test_top_level_parallel_terminates_when_all_children_final(self, sm_runner): """A root parallel terminates when all regions reach final states.""" - - class Session(StateChart): - class session(State.Parallel): - class ui(State.Compound): - active = State(initial=True) - closed = State(final=True) - - close_ui = active.to(closed) - - class backend(State.Compound): - running = State(initial=True) - stopped = State(final=True) - - stop_backend = running.to(stopped) - sm = await sm_runner.start(Session) assert sm.is_terminated is False @@ -226,25 +132,7 @@ class backend(State.Compound): async def test_top_level_parallel_done_state_fires_before_termination(self, sm_runner): """done.state fires and transitions before root-final check terminates.""" - - class Session(StateChart): - class session(State.Parallel): - class ui(State.Compound): - active = State(initial=True) - closed = State(final=True) - - close_ui = active.to(closed) - - class backend(State.Compound): - running = State(initial=True) - stopped = State(final=True) - - stop_backend = running.to(stopped) - - finished = State(final=True) - done_state_session = session.to(finished) - - sm = await sm_runner.start(Session) + sm = await sm_runner.start(SessionWithDoneState) await sm_runner.send(sm, "close_ui") await sm_runner.send(sm, "stop_backend") # done.state.session fires, transitions to finished, then terminates @@ -253,21 +141,6 @@ class backend(State.Compound): async def test_top_level_parallel_not_terminated_when_one_region_pending(self, sm_runner): """Machine keeps running when only one region reaches final.""" - - class Session(StateChart): - class session(State.Parallel): - class ui(State.Compound): - active = State(initial=True) - closed = State(final=True) - - close_ui = active.to(closed) - - class backend(State.Compound): - running = State(initial=True) - stopped = State(final=True) - - stop_backend = running.to(stopped) - sm = await sm_runner.start(Session) await sm_runner.send(sm, "close_ui") assert sm.is_terminated is False diff --git a/tests/test_validators.py b/tests/test_validators.py index 2910538b..ca07afcd 100644 --- a/tests/test_validators.py +++ b/tests/test_validators.py @@ -12,122 +12,12 @@ import pytest -from statemachine import State -from statemachine import StateChart - -# --------------------------------------------------------------------------- -# State machine definitions used across tests -# --------------------------------------------------------------------------- - - -class OrderValidation(StateChart): - """StateChart with catch_errors_as_events=True (the default).""" - - pending = State(initial=True) - confirmed = State() - cancelled = State(final=True) - - confirm = pending.to(confirmed, validators="check_stock") - cancel = confirmed.to(cancelled) - - def check_stock(self, quantity=0, **kwargs): - if quantity <= 0: - raise ValueError("Quantity must be positive") - - -class OrderValidationNoErrorEvents(StateChart): - """Same machine but with catch_errors_as_events=False.""" - - catch_errors_as_events = False - - pending = State(initial=True) - confirmed = State() - cancelled = State(final=True) - - confirm = pending.to(confirmed, validators="check_stock") - cancel = confirmed.to(cancelled) - - def check_stock(self, quantity=0, **kwargs): - if quantity <= 0: - raise ValueError("Quantity must be positive") - - -class MultiValidator(StateChart): - """Machine with multiple validators — first failure stops the chain.""" - - idle = State(initial=True) - active = State(final=True) - - start = idle.to(active, validators=["check_a", "check_b"]) - - def check_a(self, **kwargs): - if not kwargs.get("a_ok"): - raise ValueError("A failed") - - def check_b(self, **kwargs): - if not kwargs.get("b_ok"): - raise ValueError("B failed") - - -class ValidatorWithCond(StateChart): - """Machine that combines validators and conditions on the same transition.""" - - idle = State(initial=True) - active = State(final=True) - - start = idle.to(active, validators="check_auth", cond="has_permission") - - has_permission = False - - def check_auth(self, token=None, **kwargs): - if token != "valid": - raise PermissionError("Invalid token") - - -class ValidatorWithErrorTransition(StateChart): - """Machine with both a validator and an error.execution transition. - - The error.execution transition should NOT be triggered by validator - rejection — only by actual execution errors in actions. - """ - - idle = State(initial=True) - active = State() - error_state = State(final=True) - - start = idle.to(active, validators="check_input") - do_work = active.to.itself(on="risky_action") - error_execution = active.to(error_state) - - def check_input(self, value=None, **kwargs): - if value is None: - raise ValueError("Input required") - - def risky_action(self, **kwargs): - raise RuntimeError("Boom") - - -class ValidatorFallthrough(StateChart): - """Machine with multiple transitions for the same event. - - When the first transition's validator rejects, the exception propagates - immediately — the engine does NOT fall through to the next transition. - """ - - idle = State(initial=True) - path_a = State(final=True) - path_b = State(final=True) - - go = idle.to(path_a, validators="must_be_premium") | idle.to(path_b) - - def must_be_premium(self, **kwargs): - if not kwargs.get("premium"): - raise PermissionError("Premium required") - - -# --------------------------------------------------------------------------- -# Tests -# --------------------------------------------------------------------------- +from tests.machines.validators.multi_validator import MultiValidator +from tests.machines.validators.order_validation import OrderValidation +from tests.machines.validators.order_validation_no_error_events import OrderValidationNoErrorEvents +from tests.machines.validators.validator_fallthrough import ValidatorFallthrough +from tests.machines.validators.validator_with_cond import ValidatorWithCond +from tests.machines.validators.validator_with_error_transition import ValidatorWithErrorTransition class TestValidatorPropagation: