Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions docs/states.md
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,20 @@ True
Use `HistoryState(type="deep")` for deep history that remembers the exact leaf state
in nested compounds.

On the first visit there is nothing recorded yet. A transition declared from the history
state (`h.to(b)`) is its default: its target is entered and its actions run. Without one,
the compound state is entered as usual, through its initial state. Either way, the actions
of the transition that led to the history state run exactly once.

```py
>>> sm = WithHistory()
>>> sm.send("leave")
>>> sm.send("resume")
>>> sorted(sm.configuration_values)
['a', 'mode']

```


```{seealso}
See {ref}`querying-configuration` for how to inspect which states are currently
Expand Down
13 changes: 6 additions & 7 deletions statemachine/engines/async_.py
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,7 @@ async def _enter_states(
if self.sm.atomic_configuration_update:
self.sm.configuration = new_configuration

initial_transitions = {i.state.id: i.transition for i in states_for_default_entry}
for info in ordered_states:
target = info.state
transition = info.transition
Expand All @@ -273,13 +274,11 @@ async def _enter_states(
target.enter.key, *args, on_error=on_error, **kwargs
)

# Handle default initial states
if target.id in {t.state.id for t in states_for_default_entry if t.state}:
initial_transitions = [t for t in target.transitions if t.initial]
if len(initial_transitions) == 1:
result += await self.sm._callbacks.async_call(
initial_transitions[0].on.key, *args, **kwargs
)
initial_transition = initial_transitions.get(target.id)
if initial_transition is not None:
result += await self.sm._callbacks.async_call(
initial_transition.on.key, *args, **kwargs
)

# Handle default history states
default_history_transitions = [
Expand Down
95 changes: 52 additions & 43 deletions statemachine/engines/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -387,12 +387,14 @@ def get_effective_target_states(self, transition: Transition) -> OrderedSet[Stat
if state.is_history:
if state.id in self.sm.history_values:
targets.update(self.sm.history_values[state.id])
else:
elif state.transitions:
targets.update(
state
for t in state.transitions
for state in self.get_effective_target_states(t)
)
else:
targets.add(cast(State, state.parent))
else:
targets.add(state)

Expand Down Expand Up @@ -628,6 +630,7 @@ def _enter_states(
if self.sm.atomic_configuration_update:
self.sm.configuration = new_configuration

initial_transitions = {i.state.id: i.transition for i in states_for_default_entry}
for info in ordered_states:
target = info.state
transition = info.transition
Expand All @@ -646,13 +649,9 @@ def _enter_states(
target.enter.key, *args, on_error=on_error, **kwargs
)

# Handle default initial states
if target.id in {t.state.id for t in states_for_default_entry if t.state}:
initial_transitions = [t for t in target.transitions if t.initial]
if len(initial_transitions) == 1:
result += self.sm._callbacks.call(
initial_transitions[0].on.key, *args, **kwargs
)
initial_transition = initial_transitions.get(target.id)
if initial_transition is not None:
result += self.sm._callbacks.call(initial_transition.on.key, *args, **kwargs)

# Handle default history states
default_history_transitions = [
Expand Down Expand Up @@ -778,20 +777,26 @@ def _add_history_states_to_enter(
states_for_default_entry,
default_history_content,
):
"""Add the states a history state stands for: those it recorded, or its default."""
parent_id = state.parent and state.parent.id
default_history_content[parent_id] = [info]
"""Add the states a history state stands for.

Those it recorded; else its default transition's targets, whose content runs on entry
of the parent; else the parent's own default entry.
"""
parent = cast(State, state.parent)
if state.id in self.sm.history_values:
self._debug(
"%s History state '%s.%s' %s restoring: '%s'",
self._log_id,
state.parent,
parent,
state,
state.type.value,
[s.id for s in self.sm.history_values[state.id]],
)
for history_state in self.sm.history_values[state.id]:
info_to_add = StateTransition(transition=info.transition, state=history_state)
infos = [
StateTransition(transition=info.transition, state=history_state)
for history_state in self.sm.history_values[state.id]
]
for info_to_add in infos:
if state.type.is_deep:
states_to_enter.add(info_to_add)
else:
Expand All @@ -801,46 +806,50 @@ def _add_history_states_to_enter(
states_for_default_entry,
default_history_content,
)
for history_state in self.sm.history_values[state.id]:
info_to_add = StateTransition(transition=info.transition, state=history_state)
self.add_ancestor_states_to_enter(
info_to_add,
state.parent,
states_to_enter,
states_for_default_entry,
default_history_content,
)
else:
# Handle default history content
elif state.transitions:
self._debug(
"%s History state '%s.%s' default content: %s",
self._log_id,
state.parent,
parent,
state,
[t.target.id for t in state.transitions if t.target],
)

for transition in state.transitions:
target = cast(State, transition.target)
info_history = StateTransition(transition=transition, state=target)
default_history_content[parent_id].append(info_history)
infos = [
StateTransition(transition=transition, state=cast(State, transition.target))
for transition in state.transitions
]
default_history_content[parent.id] = infos
for info_to_add in infos:
self.add_descendant_states_to_enter(
info_history,
info_to_add,
states_to_enter,
states_for_default_entry,
default_history_content,
)
for transition in state.transitions:
target = cast(State, transition.target)
info_history = StateTransition(transition=transition, state=target)
else:
self._debug(
"%s History state '%s.%s' has no content, default entry of '%s'",
self._log_id,
parent,
state,
parent,
)
self.add_descendant_states_to_enter(
StateTransition(transition=info.transition, state=parent),
states_to_enter,
states_for_default_entry,
default_history_content,
)
return

self.add_ancestor_states_to_enter(
info_history,
state.parent,
states_to_enter,
states_for_default_entry,
default_history_content,
)
for info_to_add in infos:
self.add_ancestor_states_to_enter(
info_to_add,
parent,
states_to_enter,
states_for_default_entry,
default_history_content,
)

def _add_regions_to_enter(
self,
Expand Down Expand Up @@ -869,8 +878,8 @@ def _add_initial_states_to_enter(
):
"""Add the states a compound state's initial transition targets."""
state = info.state
states_for_default_entry.add(info)
transition = next(t for t in state.transitions if t.initial)
states_for_default_entry.add(StateTransition(transition=transition, state=state))
# Process all targets (supports multi-target initial transitions for parallel regions)
for initial_target in transition.targets:
info_initial = StateTransition(transition=transition, state=initial_target)
Expand Down
156 changes: 156 additions & 0 deletions tests/test_statechart_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@

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
Expand Down Expand Up @@ -106,3 +109,156 @@ async def test_history_with_default_transition(self, sm_runner):
sm = await sm_runner.start(GollumPersonalityDefaultGollum)
await sm_runner.send(sm, "enter_via_history")
assert "gollum" in sm.configuration_values


def _build_traced_machine(kind, with_default=False):
class Traced(StateChart):
outside = State(initial=True)

class region(State.Compound):
class nested(State.Compound):
n1 = State(initial=True)
n2 = State()
deeper = n1.to(n2)

other = State()
h = HistoryState(type=kind)
sideways = nested.to(other)
if with_default:
h.to(other, on="default_effect")

enter = outside.to(region, on="enter_effect")
leave = region.to(outside)
back = outside.to(region.h, on="back_effect")

def __init__(self, *args, **kwargs):
self.trace = []
super().__init__(*args, **kwargs)

def enter_effect(self):
self.trace.append("enter_effect")

def back_effect(self):
self.trace.append("back_effect")

def default_effect(self):
self.trace.append("default_effect")

def on_enter_state(self, target):
self.trace.append(f"enter:{target.id}")

def on_exit_state(self, source):
self.trace.append(f"exit:{source.id}")

return Traced


@pytest.mark.timeout(5)
@pytest.mark.parametrize("kind", ["shallow", "deep"])
class TestHistoryDefaultEntry:
async def test_no_stored_value_and_no_default_enters_parent_by_default(self, sm_runner, kind):
sm = await sm_runner.start(_build_traced_machine(kind))
sm.trace.clear()

await sm_runner.send(sm, "back")

assert {"region", "nested", "n1"} == set(sm.configuration_values)
assert sm.trace == [
"exit:outside",
"back_effect",
"enter:region",
"enter:nested",
"enter:n1",
]

async def test_machine_keeps_running_after_default_entry_of_parent(self, sm_runner, kind):
sm = await sm_runner.start(_build_traced_machine(kind))
await sm_runner.send(sm, "back")
await sm_runner.send(sm, "deeper")
await sm_runner.send(sm, "leave")
await sm_runner.send(sm, "back")

expected_leaf = "n2" if kind == "deep" else "n1"
assert {"region", "nested", expected_leaf} == set(sm.configuration_values)

async def test_stored_value_runs_incoming_effect_once(self, sm_runner, kind):
sm = await sm_runner.start(_build_traced_machine(kind))
await sm_runner.send(sm, "enter")
await sm_runner.send(sm, "sideways")
await sm_runner.send(sm, "leave")
sm.trace.clear()

await sm_runner.send(sm, "back")

assert {"region", "other"} == set(sm.configuration_values)
assert sm.trace == ["exit:outside", "back_effect", "enter:region", "enter:other"]

async def test_default_runs_incoming_effect_once_then_default_effect(self, sm_runner, kind):
sm = await sm_runner.start(_build_traced_machine(kind, with_default=True))
sm.trace.clear()

await sm_runner.send(sm, "back")

assert {"region", "other"} == set(sm.configuration_values)
assert sm.trace == [
"exit:outside",
"back_effect",
"enter:region",
"default_effect",
"enter:other",
]

async def test_stored_value_skips_default_effect(self, sm_runner, kind):
sm = await sm_runner.start(_build_traced_machine(kind, with_default=True))
await sm_runner.send(sm, "enter")
await sm_runner.send(sm, "deeper")
await sm_runner.send(sm, "leave")
sm.trace.clear()

await sm_runner.send(sm, "back")

expected_leaf = "n2" if kind == "deep" else "n1"
assert {"region", "nested", expected_leaf} == set(sm.configuration_values)
assert sm.trace == [
"exit:outside",
"back_effect",
"enter:region",
"enter:nested",
f"enter:{expected_leaf}",
]

async def test_default_entry_of_parent_enters_its_ancestors(self, sm_runner, kind):
class Nested(StateChart):
outside = State(initial=True)

class top(State.Compound):
class region(State.Compound):
r1 = State(initial=True)
h = HistoryState(type=kind)

t1 = State(initial=True)

back = outside.to(top.region.h)

sm = await sm_runner.start(Nested)

await sm_runner.send(sm, "back")

assert {"top", "region", "r1"} == set(sm.configuration_values)

async def test_default_entry_of_parent_from_a_sibling(self, sm_runner, kind):
class Siblings(StateChart):
class top(State.Compound):
t1 = State(initial=True)

class region(State.Compound):
r1 = State(initial=True)
h = HistoryState(type=kind)

back = t1.to(region.h)

sm = await sm_runner.start(Siblings)

await sm_runner.send(sm, "back")

assert {"top", "region", "r1"} == set(sm.configuration_values)
8 changes: 1 addition & 7 deletions tests/uml/pssm/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,7 @@
"Exit_003": "requires:exit_point: exit point effect runs before the state exit",
"Exiting_004": "requires:exit_point: exit point effect runs before the state exit",
"Fork_002": "requires:fork: fork branch effects interleave with entries",
"History_001-A": "bug:B2: effect into a restored history runs twice",
"History_002-A": "bug:B2: effect into a restored history runs twice",
"History_002-B": (
"bug:B2, unclassified: effect into a restored history runs twice; completions of "
"two regions share one microstep"
),
"History_002-C": "bug:B1: history without stored value nor default empties the config",
"History_002-B": "unclassified: completions of two regions share one microstep",
"History_002-D": "conflict:K10: history restores a final state, completion loops",
"Join_001": "requires:join: join effects run after all exits",
"Join_002": "requires:join: join effects run after all exits",
Expand Down
Loading