Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_simple_machine(self):
idle = State("I")
cmd = State("C")
idle.add_transitions([Transition(cmd,
self.match,
None)])
fsm = FSM([idle, cmd])
self.commands = []
self.assertEqual(fsm.state, idle)
fsm.process("250 OK\n")
self.assertEqual(fsm.state, cmd)
def test_no_init_ctor(self):
fsm = FSM([])
idle = State("I")
str(idle)
fsm.add_state(idle)
self.assertWarns(RuntimeWarning, "No next state",
txtorcon.spaghetti.__file__, fsm.process, "")
def test_handler_state(self):
idle = State("I")
cmd = State("C")
idle.add_transitions([Transition(cmd,
self.match,
lambda x: idle)])
fsm = FSM([idle, cmd])
self.commands = []
self.assertEqual(fsm.state, idle)
fsm.process("250 OK\n")
self.assertEqual(fsm.state, idle)
def test_no_matcher(self):
idle = State("I")
other = State("O")
fsm = FSM([idle, other])
idle.add_transition(Transition(other, None, None))
fsm.process("")
def set_name(self, name):
self.create_key()
self.name = name.split()[1]
def set_cookie(self, cookie):
self.cookie = cookie.split()[1]
if self.cookie.endswith('=='):
self.cookie = self.cookie[:-2]
def add_key_line(self, line):
self.key.append(line)
from txtorcon.spaghetti import FSM, State, Transition
init = State('init')
got_name = State('got_name')
got_cookie = State('got_cookie')
reading_key = State('got_key')
parser_state = ParserState()
# initial state; we want "client-name" or it's an error
init.add_transitions([
Transition(got_name, lambda line: line.startswith('client-name '), parser_state.set_name),
Transition(init, lambda line: not line.startswith('client-name '), parse_error),
])
# next up is "descriptor-cookie" or it's an error
got_name.add_transitions([
Transition(got_cookie, lambda line: line.startswith('descriptor-cookie '), parser_state.set_cookie),
Transition(init, lambda line: not line.startswith('descriptor-cookie '), parse_error),
])
self.create_key()
self.name = name.split()[1]
def set_cookie(self, cookie):
self.cookie = cookie.split()[1]
if self.cookie.endswith('=='):
self.cookie = self.cookie[:-2]
def add_key_line(self, line):
self.key.append(line)
from txtorcon.spaghetti import FSM, State, Transition
init = State('init')
got_name = State('got_name')
got_cookie = State('got_cookie')
reading_key = State('got_key')
parser_state = ParserState()
# initial state; we want "client-name" or it's an error
init.add_transitions([
Transition(got_name, lambda line: line.startswith('client-name '), parser_state.set_name),
Transition(init, lambda line: not line.startswith('client-name '), parse_error),
])
# next up is "descriptor-cookie" or it's an error
got_name.add_transitions([
Transition(got_cookie, lambda line: line.startswith('descriptor-cookie '), parser_state.set_cookie),
Transition(init, lambda line: not line.startswith('descriptor-cookie '), parse_error),
])
# the "interesting bit": there's either a client-name if we're a
self.reset()
def set_name(self, name):
self.create_key()
self.name = name.split()[1]
def set_cookie(self, cookie):
self.cookie = cookie.split()[1]
if self.cookie.endswith('=='):
self.cookie = self.cookie[:-2]
def add_key_line(self, line):
self.key.append(line)
from txtorcon.spaghetti import FSM, State, Transition
init = State('init')
got_name = State('got_name')
got_cookie = State('got_cookie')
reading_key = State('got_key')
parser_state = ParserState()
# initial state; we want "client-name" or it's an error
init.add_transitions([
Transition(got_name, lambda line: line.startswith('client-name '), parser_state.set_name),
Transition(init, lambda line: not line.startswith('client-name '), parse_error),
])
# next up is "descriptor-cookie" or it's an error
got_name.add_transitions([
Transition(got_cookie, lambda line: line.startswith('descriptor-cookie '), parser_state.set_cookie),
Transition(init, lambda line: not line.startswith('descriptor-cookie '), parse_error),
def set_name(self, name):
self.create_key()
self.name = name.split()[1]
def set_cookie(self, cookie):
self.cookie = cookie.split()[1]
if self.cookie.endswith('=='):
self.cookie = self.cookie[:-2]
def add_key_line(self, line):
self.key.append(line)
from txtorcon.spaghetti import FSM, State, Transition
init = State('init')
got_name = State('got_name')
got_cookie = State('got_cookie')
reading_key = State('got_key')
parser_state = ParserState()
# initial state; we want "client-name" or it's an error
init.add_transitions([
Transition(got_name, lambda line: line.startswith('client-name '), parser_state.set_name),
Transition(init, lambda line: not line.startswith('client-name '), parse_error),
])
# next up is "descriptor-cookie" or it's an error
got_name.add_transitions([
Transition(got_cookie, lambda line: line.startswith('descriptor-cookie '), parser_state.set_cookie),
Transition(init, lambda line: not line.startswith('descriptor-cookie '), parse_error),
])
def __init__(self, create_relay):
self._create_relay = create_relay
self._relay_attrs = None
class die(object):
__name__ = 'die' # FIXME? just to ease spagetti.py:82's pain
def __init__(self, msg):
self.msg = msg
def __call__(self, *args):
raise RuntimeError(self.msg % tuple(args))
waiting_r = State("waiting_r")
waiting_w = State("waiting_w")
waiting_p = State("waiting_p")
waiting_s = State("waiting_s")
def ignorable_line(x):
x = x.strip()
return x in ['.', 'OK', ''] or x.startswith('ns/')
waiting_r.add_transition(Transition(waiting_r, ignorable_line, None))
waiting_r.add_transition(Transition(waiting_s, lambda x: x.startswith('r '), self._router_begin))
# FIXME use better method/func than die!!
waiting_r.add_transition(Transition(waiting_r, lambda x: not x.startswith('r '), die('Expected "r " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(waiting_w, lambda x: x.startswith('s '), self._router_flags))
waiting_s.add_transition(Transition(waiting_s, lambda x: x.startswith('a '), self._router_address))
waiting_s.add_transition(Transition(waiting_r, ignorable_line, None))
waiting_s.add_transition(Transition(waiting_r, lambda x: not x.startswith('s ') and not x.startswith('a '), die('Expected "s " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', None))
self._create_relay = create_relay
self._relay_attrs = None
class die(object):
__name__ = 'die' # FIXME? just to ease spagetti.py:82's pain
def __init__(self, msg):
self.msg = msg
def __call__(self, *args):
raise RuntimeError(self.msg % tuple(args))
waiting_r = State("waiting_r")
waiting_w = State("waiting_w")
waiting_p = State("waiting_p")
waiting_s = State("waiting_s")
def ignorable_line(x):
x = x.strip()
return x in ['.', 'OK', ''] or x.startswith('ns/')
waiting_r.add_transition(Transition(waiting_r, ignorable_line, None))
waiting_r.add_transition(Transition(waiting_s, lambda x: x.startswith('r '), self._router_begin))
# FIXME use better method/func than die!!
waiting_r.add_transition(Transition(waiting_r, lambda x: not x.startswith('r '), die('Expected "r " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(waiting_w, lambda x: x.startswith('s '), self._router_flags))
waiting_s.add_transition(Transition(waiting_s, lambda x: x.startswith('a '), self._router_address))
waiting_s.add_transition(Transition(waiting_r, ignorable_line, None))
waiting_s.add_transition(Transition(waiting_r, lambda x: not x.startswith('s ') and not x.startswith('a '), die('Expected "s " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', None))