openstack/automaton
Code Issues Proposed changes

Add ruff

Browse Source 
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: Ib34f2ff5f7da87a0ae4b1f7371afefddd001a62e
This commit is contained in:
Stephen Finucane
2025-12-23 19:57:56 +00:00
parent a49b103475
commit 46a3e9e65f
11 changed files with 294 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
.pre-commit-config.yaml
View File

@@ -12,13 +12,14 @@ repos:
- id: debug-statements
- id: check-yaml
files: .*\.(yaml|yml)$
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.14.10
hooks:
- id: ruff-check
args: ['--fix', '--unsafe-fixes']
- id: ruff-format
- repo: https://opendev.org/openstack/hacking
rev: 7.0.0
rev: 8.0.0
hooks:
- id: hacking
additional_dependencies: []
- repo: https://github.com/asottile/pyupgrade
rev: v3.20.0
hooks:
- id: pyupgrade
args: [--py310-plus]

5
automaton/_utils.py
View File

@@ -12,10 +12,9 @@
# License for the specific language governing permissions and limitations
# under the License.
def get_callback_name(cb):
"""Tries to get a callbacks fully-qualified name.
"""
def get_callback_name(cb):
"""Tries to get a callbacks fully-qualified name."""
segments = [cb.__qualname__]
# When running under sphinx it appears this can be none?

55
automaton/converters/pydot.py
View File

@@ -14,14 +14,21 @@
try:
import pydot
PYDOT_AVAILABLE = True
except ImportError:
PYDOT_AVAILABLE = False
def convert(machine, graph_name,
graph_attrs=None, node_attrs_cb=None, edge_attrs_cb=None,
add_start_state=True, name_translations=None):
def convert(
machine,
graph_name,
graph_attrs=None,
node_attrs_cb=None,
edge_attrs_cb=None,
add_start_state=True,
name_translations=None,
):
"""Translates the state machine into a pydot graph.
:param machine: state machine to convert
@@ -56,9 +63,11 @@ def convert(machine, graph_name,
:type name_translations: dict
"""
if not PYDOT_AVAILABLE:
raise RuntimeError("pydot (or pydot2 or equivalent) is required"
" to convert a state machine into a pydot"
" graph")
raise RuntimeError(
"pydot (or pydot2 or equivalent) is required"
" to convert a state machine into a pydot"
" graph"
)
if not name_translations:
name_translations = {}
graph_kwargs = {
@@ -78,15 +87,17 @@ def convert(machine, graph_name,
'fontsize': '11',
}
nodes = {}
for (start_state, event, end_state) in machine:
for start_state, event, end_state in machine:
if start_state not in nodes:
start_node_attrs = node_attrs.copy()
if node_attrs_cb is not None:
start_node_attrs.update(node_attrs_cb(start_state))
pretty_start_state = name_translations.get(start_state,
start_state)
nodes[start_state] = pydot.Node(pretty_start_state,
**start_node_attrs)
pretty_start_state = name_translations.get(
start_state, start_state
)
nodes[start_state] = pydot.Node(
pretty_start_state, **start_node_attrs
)
g.add_node(nodes[start_state])
if end_state not in nodes:
end_node_attrs = node_attrs.copy()
@@ -98,12 +109,22 @@ def convert(machine, graph_name,
edge_attrs = {}
if edge_attrs_cb is not None:
edge_attrs.update(edge_attrs_cb(start_state, event, end_state))
g.add_edge(pydot.Edge(nodes[start_state], nodes[end_state],
**edge_attrs))
g.add_edge(
pydot.Edge(nodes[start_state], nodes[end_state], **edge_attrs)
)
if add_start_state and machine.default_start_state:
start = pydot.Node("__start__", shape="point", width="0.1",
xlabel='start', fontcolor='green', **node_attrs)
start = pydot.Node(
"__start__",
shape="point",
width="0.1",
xlabel='start',
fontcolor='green',
**node_attrs,
)
g.add_node(start)
g.add_edge(pydot.Edge(start, nodes[machine.default_start_state],
style='dotted'))
g.add_edge(
pydot.Edge(
start, nodes[machine.default_start_state], style='dotted'
)
)
return g

173
automaton/machines.py
View File

@@ -33,9 +33,14 @@ class State:
:ivar on_exit: callback that will be called when the state is exited.
"""
def __init__(self, name,
is_terminal=False, next_states=None,
on_enter=None, on_exit=None):
def __init__(
self,
name,
is_terminal=False,
next_states=None,
on_enter=None,
on_exit=None,
):
self.name = name
self.is_terminal = bool(is_terminal)
self.next_states = next_states
@@ -60,6 +65,7 @@ def _orderedkeys(data, sort=True):
class _Jump:
"""A FSM transition tracks this data while jumping."""
def __init__(self, name, on_enter, on_exit):
self.name = name
self.on_enter = on_enter
@@ -97,8 +103,9 @@ class FiniteMachine:
@classmethod
def _effect_builder(cls, new_state, event):
return cls.Effect(new_state['reactions'].get(event),
new_state["terminal"])
return cls.Effect(
new_state['reactions'].get(event), new_state["terminal"]
)
def __init__(self):
self._transitions = {}
@@ -122,8 +129,10 @@ class FiniteMachine:
if self.frozen:
raise excp.FrozenMachine()
if state not in self._states:
raise excp.NotFound("Can not set the default start state to"
" undefined state '%s'" % (state))
raise excp.NotFound(
"Can not set the default start state to"
f" undefined state '{state}'"
)
self._default_start_state = state
@classmethod
@@ -137,10 +146,12 @@ class FiniteMachine:
state_space = list(_convert_to_states(state_space))
m = cls()
for state in state_space:
m.add_state(state.name,
terminal=state.is_terminal,
on_enter=state.on_enter,
on_exit=state.on_exit)
m.add_state(
state.name,
terminal=state.is_terminal,
on_enter=state.on_enter,
on_exit=state.on_exit,
)
for state in state_space:
if state.next_states:
for event, next_state in state.next_states.items():
@@ -175,7 +186,7 @@ class FiniteMachine:
if self.frozen:
raise excp.FrozenMachine()
if state in self._states:
raise excp.Duplicate("State '%s' already defined" % state)
raise excp.Duplicate(f"State '{state}' already defined")
if on_enter is not None:
if not callable(on_enter):
raise ValueError("On enter callback must be callable")
@@ -222,15 +233,18 @@ class FiniteMachine:
if self.frozen:
raise excp.FrozenMachine()
if state not in self._states:
raise excp.NotFound("Can not add a reaction to event '%s' for an"
" undefined state '%s'" % (event, state))
raise excp.NotFound(
f"Can not add a reaction to event '{event}' for an"
f" undefined state '{state}'"
)
if not callable(reaction):
raise ValueError("Reaction callback must be callable")
if event not in self._states[state]['reactions']:
self._states[state]['reactions'][event] = (reaction, args, kwargs)
else:
raise excp.Duplicate("State '%s' reaction to event '%s'"
" already defined" % (state, event))
raise excp.Duplicate(
f"State '{state}' reaction to event '{event}' already defined"
)
def add_transition(self, start, end, event, replace=False):
"""Adds an allowed transition from start -> end for the given event.
@@ -246,48 +260,56 @@ class FiniteMachine:
if self.frozen:
raise excp.FrozenMachine()
if start not in self._states:
raise excp.NotFound("Can not add a transition on event '%s' that"
" starts in a undefined state '%s'"
% (event, start))
raise excp.NotFound(
f"Can not add a transition on event '{event}' that"
f" starts in a undefined state '{start}'"
)
if end not in self._states:
raise excp.NotFound("Can not add a transition on event '%s' that"
" ends in a undefined state '%s'"
% (event, end))
raise excp.NotFound(
f"Can not add a transition on event '{event}' that"
f" ends in a undefined state '{end}'"
)
if self._states[start]['terminal']:
raise excp.InvalidState("Can not add a transition on event '%s'"
" that starts in the terminal state '%s'"
% (event, start))
raise excp.InvalidState(
f"Can not add a transition on event '{event}'"
f" that starts in the terminal state '{start}'"
)
if event in self._transitions[start] and not replace:
target = self._transitions[start][event]
if target.name != end:
raise excp.Duplicate("Cannot add transition from"
" '%(start_state)s' to '%(end_state)s'"
" on event '%(event)s' because a"
" transition from '%(start_state)s'"
" to '%(existing_end_state)s' on"
" event '%(event)s' already exists."
% {'existing_end_state': target.name,
'end_state': end, 'event': event,
'start_state': start})
raise excp.Duplicate(
"Cannot add transition from"
f" '{start}' to '{end}'"
f" on event '{event}' because a"
f" transition from '{start}'"
f" to '{target.name}' on"
f" event '{event}' already exists."
)
else:
target = _Jump(end, self._states[end]['on_enter'],
self._states[start]['on_exit'])
target = _Jump(
end,
self._states[end]['on_enter'],
self._states[start]['on_exit'],
)
self._transitions[start][event] = target
def _pre_process_event(self, event):
current = self._current
if current is None:
raise excp.NotInitialized("Can not process event '%s'; the state"
" machine hasn't been initialized"
% event)
raise excp.NotInitialized(
f"Can not process event '{event}'; the state"
" machine hasn't been initialized"
)
if self._states[current.name]['terminal']:
raise excp.InvalidState("Can not transition from terminal"
" state '%s' on event '%s'"
% (current.name, event))
raise excp.InvalidState(
"Can not transition from terminal"
f" state '{current.name}' on event '{event}'"
)
if event not in self._transitions[current.name]:
raise excp.NotFound("Can not transition from state '%s' on"
" event '%s' (no defined transition)"
% (current.name, event))
raise excp.NotFound(
f"Can not transition from state '{current.name}' on"
f" event '{event}' (no defined transition)"
)
def _post_process_event(self, event, result):
return result
@@ -326,17 +348,20 @@ class FiniteMachine:
if start_state is None:
start_state = self._default_start_state
if start_state not in self._states:
raise excp.NotFound("Can not start from a undefined"
" state '%s'" % (start_state))
raise excp.NotFound(
f"Can not start from a undefined state '{start_state}'"
)
if self._states[start_state]['terminal']:
raise excp.InvalidState("Can not start from a terminal"
" state '%s'" % (start_state))
raise excp.InvalidState(
f"Can not start from a terminal state '{start_state}'"
)
# No on enter will be called, since we are priming the state machine
# and have not really transitioned from anything to get here, we will
# though allow on_exit to be called on the event that causes this
# to be moved from...
self._current = _Jump(start_state, None,
self._states[start_state]['on_exit'])
self._current = _Jump(
start_state, None, self._states[start_state]['on_exit']
)
def copy(self, shallow=False, unfreeze=False):
"""Copies the current state machine.
@@ -402,8 +427,9 @@ class FiniteMachine:
and transitions by sort order; with it being provided as false the rows
will be iterated in addition order instead.
"""
tbl = prettytable.PrettyTable(["Start", "Event", "End",
"On Enter", "On Exit"])
tbl = prettytable.PrettyTable(
["Start", "Event", "End", "On Enter", "On Exit"]
)
for state in _orderedkeys(self._states, sort=sort):
prefix_markings = []
if self.current_state == state:
@@ -415,10 +441,9 @@ class FiniteMachine:
postfix_markings.append("$")
pretty_state = "{}{}".format("".join(prefix_markings), state)
if postfix_markings:
pretty_state += "[%s]" % "".join(postfix_markings)
pretty_state += "[{}]".format("".join(postfix_markings))
if self._transitions[state]:
for event in _orderedkeys(self._transitions[state],
sort=sort):
for event in _orderedkeys(self._transitions[state], sort=sort):
target = self._transitions[state][event]
row = [pretty_state, event, target.name]
if target.on_enter is not None:
@@ -449,8 +474,7 @@ class HierarchicalFiniteMachine(FiniteMachine):
"""A fsm that understands how to run in a hierarchical mode."""
#: The result of processing an event (cause and effect...)
Effect = collections.namedtuple('Effect',
'reaction,terminal,machine')
Effect = collections.namedtuple('Effect', 'reaction,terminal,machine')
def __init__(self):
super().__init__()
@@ -458,11 +482,15 @@ class HierarchicalFiniteMachine(FiniteMachine):
@classmethod
def _effect_builder(cls, new_state, event):
return cls.Effect(new_state['reactions'].get(event),
new_state["terminal"], new_state.get('machine'))
return cls.Effect(
new_state['reactions'].get(event),
new_state["terminal"],
new_state.get('machine'),
)
def add_state(self, state,
terminal=False, on_enter=None, on_exit=None, machine=None):
def add_state(
self, state, terminal=False, on_enter=None, on_exit=None, machine=None
):
"""Adds a given state to the state machine.
:param machine: the nested state machine that will be transitioned
@@ -474,24 +502,24 @@ class HierarchicalFiniteMachine(FiniteMachine):
"""
if machine is not None and not isinstance(machine, FiniteMachine):
raise ValueError(
"Nested state machines must themselves be state machines")
"Nested state machines must themselves be state machines"
)
super().add_state(
state, terminal=terminal, on_enter=on_enter, on_exit=on_exit)
state, terminal=terminal, on_enter=on_enter, on_exit=on_exit
)
if machine is not None:
self._states[state]['machine'] = machine
self._nested_machines[state] = machine
def copy(self, shallow=False, unfreeze=False):
c = super().copy(shallow=shallow,
unfreeze=unfreeze)
c = super().copy(shallow=shallow, unfreeze=unfreeze)
if shallow:
c._nested_machines = self._nested_machines
else:
c._nested_machines = self._nested_machines.copy()
return c
def initialize(self, start_state=None,
nested_start_state_fetcher=None):
def initialize(self, start_state=None, nested_start_state_fetcher=None):
"""Sets up the state machine (sets current state to start state...).
:param start_state: explicit start state to use to initialize the
@@ -512,19 +540,20 @@ class HierarchicalFiniteMachine(FiniteMachine):
also be used to initialize any state
machines they contain (recursively).
"""
super().initialize(
start_state=start_state)
super().initialize(start_state=start_state)
for data in self._states.values():
if 'machine' in data:
nested_machine = data['machine']
nested_start_state = None
if nested_start_state_fetcher is not None:
nested_start_state = nested_start_state_fetcher(
nested_machine)
nested_machine
)
if isinstance(nested_machine, HierarchicalFiniteMachine):
nested_machine.initialize(
start_state=nested_start_state,
nested_start_state_fetcher=nested_start_state_fetcher)
nested_start_state_fetcher=nested_start_state_fetcher,
)
else:
nested_machine.initialize(start_state=nested_start_state)

29
automaton/runners.py
View File

@@ -18,10 +18,12 @@ from automaton import exceptions as excp
from automaton import machines
_JUMPER_NOT_FOUND_TPL = ("Unable to progress since no reaction (or"
" sent event) has been made available in"
" new state '%s' (moved to from state '%s'"
" in response to event '%s')")
_JUMPER_NOT_FOUND_TPL = (
"Unable to progress since no reaction (or"
" sent event) has been made available in"
" new state '%s' (moved to from state '%s'"
" in response to event '%s')"
)
class Runner(metaclass=abc.ABCMeta):
@@ -31,6 +33,7 @@ class Runner(metaclass=abc.ABCMeta):
there should not be multiple runners using the same machine instance at
the same time).
"""
def __init__(self, machine):
self._machine = machine
@@ -81,9 +84,9 @@ class FiniteRunner(Runner):
if terminal:
break
if reaction is None and sent_event is None:
raise excp.NotFound(_JUMPER_NOT_FOUND_TPL % (new_state,
old_state,
event))
raise excp.NotFound(
_JUMPER_NOT_FOUND_TPL % (new_state, old_state, event)
)
elif sent_event is not None:
event = sent_event
else:
@@ -102,8 +105,10 @@ class HierarchicalRunner(Runner):
def __init__(self, machine):
"""Create a runner for the given machine."""
if not isinstance(machine, (machines.HierarchicalFiniteMachine,)):
raise TypeError("HierarchicalRunner only works with"
" HierarchicalFiniteMachine(s)")
raise TypeError(
"HierarchicalRunner only works with"
" HierarchicalFiniteMachine(s)"
)
super().__init__(machine)
def run(self, event, initialize=True):
@@ -175,9 +180,9 @@ class HierarchicalRunner(Runner):
# events if they wish to have the root machine terminate...
break
if effect.reaction is None and sent_event is None:
raise excp.NotFound(_JUMPER_NOT_FOUND_TPL % (new_state,
old_state,
event))
raise excp.NotFound(
_JUMPER_NOT_FOUND_TPL % (new_state, old_state, event)
)
elif sent_event is not None:
event = sent_event
else:

153
automaton/tests/test_fsm.py
View File

@@ -24,7 +24,6 @@ from testtools import testcase
class FSMTest(testcase.TestCase):
@staticmethod
def _create_fsm(start_state, add_start=True, add_states=None):
m = machines.FiniteMachine()
@@ -57,10 +56,12 @@ class FSMTest(testcase.TestCase):
def test_build_transitions(self):
space = [
machines.State('down', is_terminal=False,
next_states={'jump': 'up'}),
machines.State('up', is_terminal=False,
next_states={'fall': 'down'}),
machines.State(
'down', is_terminal=False, next_states={'jump': 'up'}
),
machines.State(
'up', is_terminal=False, next_states={'fall': 'down'}
),
]
m = machines.FiniteMachine.build(space)
m.default_start_state = 'down'
@@ -78,12 +79,20 @@ class FSMTest(testcase.TestCase):
exitted[state].append(event)
space = [
machines.State('down', is_terminal=False,
next_states={'jump': 'up'},
on_enter=on_enter, on_exit=on_exit),
machines.State('up', is_terminal=False,
next_states={'fall': 'down'},
on_enter=on_enter, on_exit=on_exit),
machines.State(
'down',
is_terminal=False,
next_states={'jump': 'up'},
on_enter=on_enter,
on_exit=on_exit,
),
machines.State(
'up',
is_terminal=False,
next_states={'fall': 'down'},
on_enter=on_enter,
on_exit=on_exit,
),
]
m = machines.FiniteMachine.build(space)
m.default_start_state = 'down'
@@ -104,11 +113,13 @@ class FSMTest(testcase.TestCase):
def test_build_transitions_dct(self):
space = [
{
'name': 'down', 'is_terminal': False,
'name': 'down',
'is_terminal': False,
'next_states': {'jump': 'up'},
},
{
'name': 'up', 'is_terminal': False,
'name': 'up',
'is_terminal': False,
'next_states': {'fall': 'down'},
},
]
@@ -119,8 +130,9 @@ class FSMTest(testcase.TestCase):
def test_build_terminal(self):
space = [
machines.State('down', is_terminal=False,
next_states={'jump': 'fell_over'}),
machines.State(
'down', is_terminal=False, next_states={'jump': 'fell_over'}
),
machines.State('fell_over', is_terminal=True),
]
m = machines.FiniteMachine.build(space)
@@ -148,8 +160,9 @@ class FSMTest(testcase.TestCase):
def test_no_add_transition_terminal(self):
m = self._create_fsm('up')
m.add_state('down', terminal=True)
self.assertRaises(excp.InvalidState,
m.add_transition, 'down', 'up', 'jump')
self.assertRaises(
excp.InvalidState, m.add_transition, 'down', 'up', 'jump'
)
def test_duplicate_state(self):
m = self._create_fsm('unknown')
@@ -158,8 +171,9 @@ class FSMTest(testcase.TestCase):
def test_duplicate_transition(self):
m = self.jumper
m.add_state('side_ways')
self.assertRaises(excp.Duplicate,
m.add_transition, 'up', 'side_ways', 'fall')
self.assertRaises(
excp.Duplicate, m.add_transition, 'up', 'side_ways', 'fall'
)
def test_duplicate_transition_replace(self):
m = self.jumper
@@ -174,20 +188,31 @@ class FSMTest(testcase.TestCase):
self.assertRaises(
# Currently duplicate reactions are not allowed...
excp.Duplicate,
self.jumper.add_reaction, 'down', 'fall', lambda *args: 'skate')
self.jumper.add_reaction,
'down',
'fall',
lambda *args: 'skate',
)
def test_bad_transition(self):
m = self._create_fsm('unknown')
m.add_state('fire')
self.assertRaises(excp.NotFound, m.add_transition,
'unknown', 'something', 'boom')
self.assertRaises(excp.NotFound, m.add_transition,
'something', 'unknown', 'boom')
self.assertRaises(
excp.NotFound, m.add_transition, 'unknown', 'something', 'boom'
)
self.assertRaises(
excp.NotFound, m.add_transition, 'something', 'unknown', 'boom'
)
def test_bad_reaction(self):
m = self._create_fsm('unknown')
self.assertRaises(excp.NotFound, m.add_reaction, 'something', 'boom',
lambda *args: 'cough')
self.assertRaises(
excp.NotFound,
m.add_reaction,
'something',
'boom',
lambda *args: 'cough',
)
def test_run(self):
m = self._create_fsm('down', add_states=['up', 'down'])
@@ -204,8 +229,7 @@ class FSMTest(testcase.TestCase):
r.run('jump')
self.assertTrue(m.terminated)
self.assertEqual('broken', m.current_state)
self.assertRaises(excp.InvalidState, r.run,
'jump', initialize=False)
self.assertRaises(excp.InvalidState, r.run, 'jump', initialize=False)
def test_on_enter_on_exit(self):
enter_transitions = []
@@ -229,20 +253,25 @@ class FSMTest(testcase.TestCase):
m.process_event('beat')
m.process_event('jump')
m.process_event('fall')
self.assertEqual([('down', 'beat'),
('up', 'jump'), ('down', 'fall')], enter_transitions)
self.assertEqual([('start', 'beat'), ('down', 'jump'), ('up', 'fall')],
exit_transitions)
self.assertEqual(
[('down', 'beat'), ('up', 'jump'), ('down', 'fall')],
enter_transitions,
)
self.assertEqual(
[('start', 'beat'), ('down', 'jump'), ('up', 'fall')],
exit_transitions,
)
def test_run_iter(self):
up_downs = []
runner = runners.FiniteRunner(self.jumper)
for (old_state, new_state) in runner.run_iter('jump'):
for old_state, new_state in runner.run_iter('jump'):
up_downs.append((old_state, new_state))
if len(up_downs) >= 3:
break
self.assertEqual([('down', 'up'), ('up', 'down'), ('down', 'up')],
up_downs)
self.assertEqual(
[('down', 'up'), ('up', 'down'), ('down', 'up')], up_downs
)
self.assertFalse(self.jumper.terminated)
self.assertEqual('up', self.jumper.current_state)
self.jumper.process_event('fall')
@@ -259,8 +288,9 @@ class FSMTest(testcase.TestCase):
break
self.assertEqual('up', self.jumper.current_state)
self.assertFalse(self.jumper.terminated)
self.assertEqual([('down', 'up'), ('up', 'down'), ('down', 'up')],
up_downs)
self.assertEqual(
[('down', 'up'), ('up', 'down'), ('down', 'up')], up_downs
)
self.assertRaises(StopIteration, next, it)
def test_run_send_fail(self):
@@ -273,8 +303,9 @@ class FSMTest(testcase.TestCase):
self.assertEqual([('down', 'up')], up_downs)
def test_not_initialized(self):
self.assertRaises(excp.NotInitialized,
self.jumper.process_event, 'jump')
self.assertRaises(
excp.NotInitialized, self.jumper.process_event, 'jump'
)
def test_copy_states(self):
c = self._create_fsm('down', add_start=False)
@@ -322,11 +353,20 @@ class FSMTest(testcase.TestCase):
def test_freeze(self):
self.jumper.freeze()
self.assertRaises(excp.FrozenMachine, self.jumper.add_state, 'test')
self.assertRaises(excp.FrozenMachine,
self.jumper.add_transition, 'test', 'test', 'test')
self.assertRaises(excp.FrozenMachine,
self.jumper.add_reaction,
'test', 'test', lambda *args: 'test')
self.assertRaises(
excp.FrozenMachine,
self.jumper.add_transition,
'test',
'test',
'test',
)
self.assertRaises(
excp.FrozenMachine,
self.jumper.add_reaction,
'test',
'test',
lambda *args: 'test',
)
def test_freeze_copy_unfreeze(self):
self.jumper.freeze()
@@ -342,10 +382,10 @@ class FSMTest(testcase.TestCase):
class HFSMTest(FSMTest):
@staticmethod
def _create_fsm(start_state,
add_start=True, hierarchical=False, add_states=None):
def _create_fsm(
start_state, add_start=True, hierarchical=False, add_states=None
):
if hierarchical:
m = machines.HierarchicalFiniteMachine()
else:
@@ -360,7 +400,6 @@ class HFSMTest(FSMTest):
return m
def _make_phone_call(self, talk_time=1.0):
def phone_reaction(old_state, new_state, event, chat_iter):
try:
next(chat_iter)
@@ -405,11 +444,13 @@ class HFSMTest(FSMTest):
number_calling = []
digits.add_state(
"accumulate",
on_enter=lambda *args: number_calling.append(digit_maker()))
on_enter=lambda *args: number_calling.append(digit_maker()),
)
digits.add_transition("idle", "accumulate", "press")
digits.add_transition("accumulate", "accumulate", "press")
digits.add_reaction("accumulate", "press",
react_to_press, number_calling)
digits.add_reaction(
"accumulate", "press", react_to_press, number_calling
)
digits.add_state("dial", terminal=True)
digits.add_transition("accumulate", "dial", "call")
digits.add_reaction("dial", "call", lambda *args: 'ringing')
@@ -440,9 +481,13 @@ class HFSMTest(FSMTest):
r = runners.HierarchicalRunner(dialer)
transitions = list(r.run_iter('dial'))
self.assertEqual(('talk', 'hangup'), transitions[-1])
self.assertEqual(len(number_calling),
sum(1 if new_state == 'accumulate' else 0
for (old_state, new_state) in transitions))
self.assertEqual(
len(number_calling),
sum(
1 if new_state == 'accumulate' else 0
for (old_state, new_state) in transitions
),
)
self.assertEqual(10, len(number_calling))
def test_phone_call(self):

11
doc/source/conf.py
View File

@@ -73,8 +73,11 @@ html_theme = 'openstackdocs'
# (source start file, target name, title, author, documentclass
# [howto/manual]).
latex_documents = [
('index',
'%s.tex' % project,
'%s Documentation' % project,
'OpenStack Foundation', 'manual'),
(
'index',
f'{project}.tex',
f'{project} Documentation',
'OpenStack Foundation',
'manual',
),
]

12
pyproject.toml
View File

@@ -35,3 +35,15 @@ packages = [
"automaton"
]
[tool.ruff]
line-length = 79
[tool.ruff.format]
quote-style = "preserve"
docstring-code-format = true
[tool.ruff.lint]
select = ["E4", "E5", "E7", "E9", "F", "G", "LOG", "S", "UP"]
[tool.ruff.lint.per-file-ignores]
"automaton/tests/*" = ["S"]

10
releasenotes/source/conf.py
View File

@@ -62,9 +62,13 @@ html_static_path = ['_static']
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
('index', 'automatonReleaseNotes.tex',
'automaton Release Notes Documentation',
'automaton Developers', 'manual'),
(
'index',
'automatonReleaseNotes.tex',
'automaton Release Notes Documentation',
'automaton Developers',
'manual',
),
]
# -- Options for Internationalization output ------------------------------

4
setup.py
View File

@@ -15,6 +15,4 @@
import setuptools
setuptools.setup(
setup_requires=['pbr>=2.0.0'],
pbr=True)
setuptools.setup(setup_requires=['pbr>=2.0.0'], pbr=True)

4
tox.ini
View File

@@ -60,5 +60,7 @@ commands = bindep test
usedevelop = False
[flake8]
# We only enable the hacking (H) checks
select = H
show-source = True
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build
exclude = .venv,.git,.tox,dist,doc,*lib/python*,*egg,build