2018-07-10 15:32:43 +00:00
|
|
|
# This defines an FSM equvivalent to test_fsm.yml
|
|
|
|
|
|
|
|
from os.path import exists
|
|
|
|
|
2018-04-10 09:19:53 +00:00
|
|
|
from tfw import LinearFSM
|
|
|
|
from tfw.networking import MessageSender
|
|
|
|
|
|
|
|
|
|
|
|
class TestFSM(LinearFSM):
|
2018-04-19 09:22:43 +00:00
|
|
|
# pylint: disable=unused-argument
|
|
|
|
|
2018-07-10 15:32:43 +00:00
|
|
|
def __init__(self):
|
|
|
|
super().__init__(6)
|
2018-04-10 09:19:53 +00:00
|
|
|
self.message_sender = MessageSender()
|
2018-07-10 15:32:43 +00:00
|
|
|
self.subscribe_predicate('step_3', self.step_3_allowed)
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def step_3_allowed():
|
|
|
|
return exists('/home/user/workdir/allow_step_3')
|
2018-04-10 09:19:53 +00:00
|
|
|
|
|
|
|
def on_enter_1(self, event_data):
|
2018-07-10 15:32:43 +00:00
|
|
|
self.message_sender.send('FSM', 'Entered state 1!')
|
2018-04-10 09:19:53 +00:00
|
|
|
|
|
|
|
def on_enter_2(self, event_data):
|
2018-07-10 15:32:43 +00:00
|
|
|
filename = '/home/user/workdir/cat.txt'
|
|
|
|
with open(filename, 'w') as ofile:
|
|
|
|
ofile.write('As you can see it is possible to write arbitrary python code here.')
|
|
|
|
self.message_sender.send('FSM', f'Entered state 2! Written stuff to {filename}')
|
2018-04-10 09:19:53 +00:00
|
|
|
|
|
|
|
def on_enter_3(self, event_data):
|
2018-07-10 15:32:43 +00:00
|
|
|
self.message_sender.send('FSM', 'Entered state 3!')
|
2018-04-10 09:19:53 +00:00
|
|
|
|
|
|
|
def on_enter_4(self, event_data):
|
2018-07-10 15:32:43 +00:00
|
|
|
self.message_sender.send('FSM', 'Entered state 4!')
|
2018-04-10 09:19:53 +00:00
|
|
|
|
|
|
|
def on_enter_5(self, event_data):
|
2018-07-10 15:32:43 +00:00
|
|
|
self.message_sender.send('FSM', 'Entered state 5!')
|