make Input tests pass, clarify error cases, cleanups

This commit is contained in:
Brian Warner 2017-03-19 17:35:05 +01:00
parent 175fef2ab4
commit 3873f55d64
9 changed files with 461 additions and 113 deletions

View File

@ -181,42 +181,49 @@ The code-entry Helper object has the following API:
"4" in "4-purple-sausages"). Note that they are unicode strings (so "4", "4" in "4-purple-sausages"). Note that they are unicode strings (so "4",
not 4). The Helper will get the response in the background, and calls to not 4). The Helper will get the response in the background, and calls to
`get_nameplate_completions()` after the response will use the new list. `get_nameplate_completions()` after the response will use the new list.
Calling this after `h.choose_nameplate` will raise
`AlreadyChoseNameplateError`.
* `completions = h.get_nameplate_completions(prefix)`: returns * `completions = h.get_nameplate_completions(prefix)`: returns
(synchronously) a list of suffixes for the given nameplate prefix. For (synchronously) a set of suffixes for the given nameplate prefix. For
example, if the server reports nameplates 1, 12, 13, 24, and 170 are in example, if the server reports nameplates 1, 12, 13, 24, and 170 are in
use, `get_nameplate_completions("1")` will return `["", "2", "3", "70"]`. use, `get_nameplate_completions("1")` will return `{"", "2", "3", "70"}`.
Raises `AlreadyClaimedNameplateError` if called after `h.choose_nameplate`. You may want to sort these before displaying them to the user. Raises
* `h.choose_nameplate(nameplate)`: accepts a string with the chosen nameplate. `AlreadyChoseNameplateError` if called after `h.choose_nameplate`.
May only be called once, after which `OnlyOneNameplateError` is raised. (in * `h.choose_nameplate(nameplate)`: accepts a string with the chosen
this future, this might return a Deferred that fires (with None) when the nameplate. May only be called once, after which
nameplate's wordlist is known (which happens after the nameplate is `AlreadyChoseNameplateError` is raised. (in this future, this might
claimed, requiring a roundtrip to the server)). return a Deferred that fires (with None) when the nameplate's wordlist is
known (which happens after the nameplate is claimed, requiring a roundtrip
to the server)).
* `completions = h.get_word_completions(prefix)`: return (synchronously) a * `completions = h.get_word_completions(prefix)`: return (synchronously) a
list of suffixes for the given words prefix. The possible completions set of suffixes for the given words prefix. The possible completions depend
depend upon the wordlist in use for the previously-claimed nameplate, so upon the wordlist in use for the previously-claimed nameplate, so calling
calling this before `choose_nameplate` will raise this before `choose_nameplate` will raise `MustChooseNameplateFirstError`.
`MustClaimNameplateFirstError`. Given a prefix like "su", this returns a Calling this after `h.choose_words()` will raise `AlreadyChoseWordsError`.
list of strings which are appropriate to append to the prefix (e.g. Given a prefix like "su", this returns a set of strings which are
`["pportive", "rrender", "spicious"]`, for expansion into "supportive", appropriate to append to the prefix (e.g. `{"pportive", "rrender",
"surrender", and "suspicious". The prefix should not include the nameplate, "spicious"}`, for expansion into "supportive", "surrender", and
but *should* include whatever words and hyphens have been typed so far (the "suspicious". The prefix should not include the nameplate, but *should*
default wordlist uses alternate lists, where even numbered words have three include whatever words and hyphens have been typed so far (the default
wordlist uses alternate lists, where even numbered words have three
syllables, and odd numbered words have two, so the completions depend upon syllables, and odd numbered words have two, so the completions depend upon
how many words are present, not just the partial last word). E.g. how many words are present, not just the partial last word). E.g.
`get_word_completions("pr")` will return `["ocessor", "ovincial", `get_word_completions("pr")` will return `{"ocessor", "ovincial",
"oximate"]`, while `get_word_completions("opulent-pr")` will return "oximate"}`, while `get_word_completions("opulent-pr")` will return
`["eclude", "efer", "eshrunk", "inter", "owler"]`. If the wordlist is not `{"eclude", "efer", "eshrunk", "inter", "owler"}`. If the wordlist is not
yet known, this returns an empty list. It will also return an empty list if yet known, this returns an empty set. It will also return an empty set if
the prefix is complete (the last word matches something in the completion the prefix is complete (the last word matches something in the completion
list, and there are no longer extension words), although the code may not list, and there are no longer extension words), although the code may not
yet be complete if there are additional words. The completions will never yet be complete if there are additional words. The completions will never
include a hyphen: the UI frontend must supply these if desired. include a hyphen: the UI frontend must supply these if desired. The
frontend is also responsible for sorting the results before display.
* `h.choose_words(words)`: call this when the user is finished typing in the * `h.choose_words(words)`: call this when the user is finished typing in the
code. It does not return anything, but will cause the Wormhole's code. It does not return anything, but will cause the Wormhole's
`w.when_code()` (or corresponding delegate) to fire, and triggers the `w.when_code()` (or corresponding delegate) to fire, and triggers the
wormhole connection process. This accepts a string like "purple-sausages", wormhole connection process. This accepts a string like "purple-sausages",
without the nameplate. It must be called after `h.choose_nameplate()` or without the nameplate. It must be called after `h.choose_nameplate()` or
`MustClaimNameplateFirstError` will be raised. `MustChooseNameplateFirstError` will be raised. May only be called once,
after which `AlreadyChoseWordsError` is raised.
The `rlcompleter` wrapper is a function that knows how to use the code-entry The `rlcompleter` wrapper is a function that knows how to use the code-entry
helper to do tab completion of wormhole codes: helper to do tab completion of wormhole codes:

View File

@ -36,7 +36,7 @@ digraph {
S4 [label="S4: done" color="green"] S4 [label="S4: done" color="green"]
S4 -> S4 [label="got_nameplates\ngot_wordlist"] S4 -> S4 [label="got_nameplates\ngot_wordlist"]
other [shape="box" style="dotted" other [shape="box" style="dotted" color="orange" fontcolor="orange"
label="h.refresh_nameplates()\nh.get_nameplate_completions(prefix)\nh.choose_nameplate(nameplate)\nh.get_word_completions(prefix)\nh.choose_words(words)" label="h.refresh_nameplates()\nh.get_nameplate_completions(prefix)\nh.choose_nameplate(nameplate)\nh.get_word_completions(prefix)\nh.choose_words(words)"
] ]
{rank=same; S4 other} {rank=same; S4 other}

View File

@ -35,7 +35,7 @@ class Code(object):
@m.input() @m.input()
def allocate_code(self, length, wordlist): pass def allocate_code(self, length, wordlist): pass
@m.input() @m.input()
def input_code(self, input_helper): pass def input_code(self): pass
@m.input() @m.input()
def set_code(self, code): pass def set_code(self, code): pass
@ -57,8 +57,8 @@ class Code(object):
self._B.got_code(code) self._B.got_code(code)
@m.output() @m.output()
def do_start_input(self, input_helper): def do_start_input(self):
self._I.start(input_helper) return self._I.start()
@m.output() @m.output()
def do_middle_input(self, nameplate): def do_middle_input(self, nameplate):
self._N.set_nameplate(nameplate) self._N.set_nameplate(nameplate)
@ -72,6 +72,7 @@ class Code(object):
self._A.allocate(length, wordlist) self._A.allocate(length, wordlist)
@m.output() @m.output()
def do_finish_allocate(self, nameplate, code): def do_finish_allocate(self, nameplate, code):
assert code.startswith(nameplate+"-"), (nameplate, code)
self._N.set_nameplate(nameplate) self._N.set_nameplate(nameplate)
self._K.got_code(code) self._K.got_code(code)
self._B.got_code(code) self._B.got_code(code)

View File

@ -3,7 +3,10 @@ from zope.interface import implementer
from attr import attrs, attrib from attr import attrs, attrib
from attr.validators import provides from attr.validators import provides
from automat import MethodicalMachine from automat import MethodicalMachine
from . import _interfaces from . import _interfaces, errors
def first(outputs):
return list(outputs)[0]
@attrs @attrs
@implementer(_interfaces.IInput) @implementer(_interfaces.IInput)
@ -14,6 +17,7 @@ class Input(object):
def set_trace(): pass # pragma: no cover def set_trace(): pass # pragma: no cover
def __attrs_post_init__(self): def __attrs_post_init__(self):
self._all_nameplates = set()
self._nameplate = None self._nameplate = None
self._wordlist = None self._wordlist = None
@ -34,11 +38,11 @@ class Input(object):
# from Code # from Code
@m.input() @m.input()
def start(self, input_helper): pass def start(self): pass
# from Lister # from Lister
@m.input() @m.input()
def got_nameplates(self, nameplates): pass def got_nameplates(self, all_nameplates): pass
# from Nameplate # from Nameplate
@m.input() @m.input()
@ -48,62 +52,163 @@ class Input(object):
@m.input() @m.input()
def refresh_nameplates(self): pass def refresh_nameplates(self): pass
@m.input() @m.input()
def get_nameplate_completions(self, prefix): pass
@m.input()
def choose_nameplate(self, nameplate): pass def choose_nameplate(self, nameplate): pass
@m.input() @m.input()
def get_word_completions(self, prefix): pass
@m.input()
def choose_words(self, words): pass def choose_words(self, words): pass
@m.output() @m.output()
def do_start(self, input_helper): def do_start(self):
self._input_helper = input_helper self._L.refresh()
self._L.refresh_nameplates() return Helper(self)
@m.output() @m.output()
def do_refresh(self): def do_refresh(self):
self._L.refresh_nameplates() self._L.refresh()
@m.output() @m.output()
def do_nameplate(self, nameplate): def record_nameplates(self, all_nameplates):
# we get a set of nameplate id strings
self._all_nameplates = all_nameplates
@m.output()
def _get_nameplate_completions(self, prefix):
lp = len(prefix)
completions = set()
for nameplate in self._all_nameplates:
if nameplate.startswith(prefix):
completions.add(nameplate[lp:])
return completions
@m.output()
def record_all_nameplates(self, nameplate):
self._nameplate = nameplate self._nameplate = nameplate
self._C.got_nameplate(nameplate) self._C.got_nameplate(nameplate)
@m.output() @m.output()
def do_wordlist(self, wordlist): def record_wordlist(self, wordlist):
self._wordlist = wordlist self._wordlist = wordlist
@m.output()
def no_word_completions(self, prefix):
return set()
@m.output()
def _get_word_completions(self, prefix):
assert self._wordlist
return self._wordlist.get_completions(prefix)
@m.output()
def raise_must_choose_nameplate1(self, prefix):
raise errors.MustChooseNameplateFirstError()
@m.output()
def raise_must_choose_nameplate2(self, words):
raise errors.MustChooseNameplateFirstError()
@m.output()
def raise_already_chose_nameplate1(self):
raise errors.AlreadyChoseNameplateError()
@m.output()
def raise_already_chose_nameplate2(self, prefix):
raise errors.AlreadyChoseNameplateError()
@m.output()
def raise_already_chose_nameplate3(self, nameplate):
raise errors.AlreadyChoseNameplateError()
@m.output()
def raise_already_chose_words1(self, prefix):
raise errors.AlreadyChoseWordsError()
@m.output()
def raise_already_chose_words2(self, words):
raise errors.AlreadyChoseWordsError()
@m.output() @m.output()
def do_words(self, words): def do_words(self, words):
code = self._nameplate + "-" + words code = self._nameplate + "-" + words
self._C.finished_input(code) self._C.finished_input(code)
S0_idle.upon(start, enter=S1_typing_nameplate, outputs=[do_start]) S0_idle.upon(start, enter=S1_typing_nameplate,
outputs=[do_start], collector=first)
S1_typing_nameplate.upon(got_nameplates, enter=S1_typing_nameplate,
outputs=[record_nameplates])
# too early for got_wordlist, should never happen
S1_typing_nameplate.upon(refresh_nameplates, enter=S1_typing_nameplate, S1_typing_nameplate.upon(refresh_nameplates, enter=S1_typing_nameplate,
outputs=[do_refresh]) outputs=[do_refresh])
S1_typing_nameplate.upon(get_nameplate_completions,
enter=S1_typing_nameplate,
outputs=[_get_nameplate_completions],
collector=first)
S1_typing_nameplate.upon(choose_nameplate, enter=S2_typing_code_no_wordlist, S1_typing_nameplate.upon(choose_nameplate, enter=S2_typing_code_no_wordlist,
outputs=[do_nameplate]) outputs=[record_all_nameplates])
S2_typing_code_no_wordlist.upon(got_wordlist, S1_typing_nameplate.upon(get_word_completions,
enter=S3_typing_code_yes_wordlist, enter=S1_typing_nameplate,
outputs=[do_wordlist]) outputs=[raise_must_choose_nameplate1])
S2_typing_code_no_wordlist.upon(choose_words, enter=S4_done, S1_typing_nameplate.upon(choose_words, enter=S1_typing_nameplate,
outputs=[do_words]) outputs=[raise_must_choose_nameplate2])
S2_typing_code_no_wordlist.upon(got_nameplates, S2_typing_code_no_wordlist.upon(got_nameplates,
enter=S2_typing_code_no_wordlist, outputs=[]) enter=S2_typing_code_no_wordlist, outputs=[])
S3_typing_code_yes_wordlist.upon(choose_words, enter=S4_done, S2_typing_code_no_wordlist.upon(got_wordlist,
outputs=[do_words]) enter=S3_typing_code_yes_wordlist,
outputs=[record_wordlist])
S2_typing_code_no_wordlist.upon(refresh_nameplates,
enter=S2_typing_code_no_wordlist,
outputs=[raise_already_chose_nameplate1])
S2_typing_code_no_wordlist.upon(get_nameplate_completions,
enter=S2_typing_code_no_wordlist,
outputs=[raise_already_chose_nameplate2])
S2_typing_code_no_wordlist.upon(choose_nameplate,
enter=S2_typing_code_no_wordlist,
outputs=[raise_already_chose_nameplate3])
S2_typing_code_no_wordlist.upon(get_word_completions,
enter=S2_typing_code_no_wordlist,
outputs=[no_word_completions],
collector=first)
S2_typing_code_no_wordlist.upon(choose_words, enter=S4_done,
outputs=[do_words])
S3_typing_code_yes_wordlist.upon(got_nameplates, S3_typing_code_yes_wordlist.upon(got_nameplates,
enter=S3_typing_code_yes_wordlist, enter=S3_typing_code_yes_wordlist,
outputs=[]) outputs=[])
# got_wordlist: should never happen
S3_typing_code_yes_wordlist.upon(refresh_nameplates,
enter=S3_typing_code_yes_wordlist,
outputs=[raise_already_chose_nameplate1])
S3_typing_code_yes_wordlist.upon(get_nameplate_completions,
enter=S3_typing_code_yes_wordlist,
outputs=[raise_already_chose_nameplate2])
S3_typing_code_yes_wordlist.upon(choose_nameplate,
enter=S3_typing_code_yes_wordlist,
outputs=[raise_already_chose_nameplate3])
S3_typing_code_yes_wordlist.upon(get_word_completions,
enter=S3_typing_code_yes_wordlist,
outputs=[_get_word_completions],
collector=first)
S3_typing_code_yes_wordlist.upon(choose_words, enter=S4_done,
outputs=[do_words])
S4_done.upon(got_nameplates, enter=S4_done, outputs=[]) S4_done.upon(got_nameplates, enter=S4_done, outputs=[])
S4_done.upon(got_wordlist, enter=S4_done, outputs=[]) S4_done.upon(got_wordlist, enter=S4_done, outputs=[])
S4_done.upon(refresh_nameplates,
enter=S4_done,
outputs=[raise_already_chose_nameplate1])
S4_done.upon(get_nameplate_completions,
enter=S4_done,
outputs=[raise_already_chose_nameplate2])
S4_done.upon(choose_nameplate, enter=S4_done,
outputs=[raise_already_chose_nameplate3])
S4_done.upon(get_word_completions, enter=S4_done,
outputs=[raise_already_chose_words1])
S4_done.upon(choose_words, enter=S4_done,
outputs=[raise_already_chose_words2])
# methods for the CodeInputHelper to use # we only expose the Helper to application code, not _Input
#refresh_nameplates/_choose_nameplate/choose_words: @m.input methods @attrs
class Helper(object):
_input = attrib()
def refresh_nameplates(self):
self._input.refresh_nameplates()
def get_nameplate_completions(self, prefix): def get_nameplate_completions(self, prefix):
lp = len(prefix) return self._input.get_nameplate_completions(prefix)
completions = [] def choose_nameplate(self, nameplate):
for nameplate in self._nameplates: self._input.choose_nameplate(nameplate)
if nameplate.startswith(prefix):
completions.append(nameplate[lp:])
return completions
def get_word_completions(self, prefix): def get_word_completions(self, prefix):
if self._wordlist: return self._input.get_word_completions(prefix)
return self._wordlist.get_completions(prefix) def choose_words(self, words):
return [] self._input.choose_words(words)

View File

@ -1,10 +1,11 @@
from __future__ import print_function, absolute_import, unicode_literals from __future__ import print_function, absolute_import, unicode_literals
from zope.interface import implementer from zope.interface import implementer
from attr import attrib from attr import attrs, attrib
from attr.validators import provides from attr.validators import provides
from automat import MethodicalMachine from automat import MethodicalMachine
from . import _interfaces from . import _interfaces
@attrs
@implementer(_interfaces.ILister) @implementer(_interfaces.ILister)
class Lister(object): class Lister(object):
_timing = attrib(validator=provides(_interfaces.ITiming)) _timing = attrib(validator=provides(_interfaces.ITiming))
@ -39,32 +40,35 @@ class Lister(object):
@m.input() @m.input()
def lost(self): pass def lost(self): pass
@m.input() @m.input()
def refresh_nameplates(self): pass def refresh(self): pass
@m.input() @m.input()
def rx_nameplates(self, message): pass def rx_nameplates(self, all_nameplates): pass
@m.output() @m.output()
def RC_tx_list(self): def RC_tx_list(self):
self._RC.tx_list() self._RC.tx_list()
@m.output() @m.output()
def I_got_nameplates(self, message): def I_got_nameplates(self, all_nameplates):
self._I.got_nameplates(message["nameplates"]) # We get a set of nameplate ids. There may be more attributes in the
# future: change RendezvousConnector._response_handle_nameplates to
# get them
self._I.got_nameplates(all_nameplates)
S0A_idle_disconnected.upon(connected, enter=S0B_idle_connected, outputs=[]) S0A_idle_disconnected.upon(connected, enter=S0B_idle_connected, outputs=[])
S0B_idle_connected.upon(lost, enter=S0A_idle_disconnected, outputs=[]) S0B_idle_connected.upon(lost, enter=S0A_idle_disconnected, outputs=[])
S0A_idle_disconnected.upon(refresh_nameplates, S0A_idle_disconnected.upon(refresh,
enter=S1A_wanting_disconnected, outputs=[]) enter=S1A_wanting_disconnected, outputs=[])
S1A_wanting_disconnected.upon(refresh_nameplates, S1A_wanting_disconnected.upon(refresh,
enter=S1A_wanting_disconnected, outputs=[]) enter=S1A_wanting_disconnected, outputs=[])
S1A_wanting_disconnected.upon(connected, enter=S1B_wanting_connected, S1A_wanting_disconnected.upon(connected, enter=S1B_wanting_connected,
outputs=[RC_tx_list]) outputs=[RC_tx_list])
S0B_idle_connected.upon(refresh_nameplates, enter=S1B_wanting_connected, S0B_idle_connected.upon(refresh, enter=S1B_wanting_connected,
outputs=[RC_tx_list]) outputs=[RC_tx_list])
S0B_idle_connected.upon(rx_nameplates, enter=S0B_idle_connected, S0B_idle_connected.upon(rx_nameplates, enter=S0B_idle_connected,
outputs=[I_got_nameplates]) outputs=[I_got_nameplates])
S1B_wanting_connected.upon(lost, enter=S1A_wanting_disconnected, outputs=[]) S1B_wanting_connected.upon(lost, enter=S1A_wanting_disconnected, outputs=[])
S1B_wanting_connected.upon(refresh_nameplates, enter=S1B_wanting_connected, S1B_wanting_connected.upon(refresh, enter=S1B_wanting_connected,
outputs=[RC_tx_list]) outputs=[RC_tx_list])
S1B_wanting_connected.upon(rx_nameplates, enter=S0B_idle_connected, S1B_wanting_connected.upon(rx_nameplates, enter=S0B_idle_connected,
outputs=[I_got_nameplates]) outputs=[I_got_nameplates])

View File

@ -147,6 +147,7 @@ class RendezvousConnector(object):
self._N.connected() self._N.connected()
self._M.connected() self._M.connected()
self._L.connected() self._L.connected()
self._A.connected()
except Exception as e: except Exception as e:
self._B.error(e) self._B.error(e)
raise raise
@ -186,6 +187,7 @@ class RendezvousConnector(object):
self._N.lost() self._N.lost()
self._M.lost() self._M.lost()
self._L.lost() self._L.lost()
self._A.lost()
# internal # internal
def _stopped(self, res): def _stopped(self, res):
@ -207,17 +209,19 @@ class RendezvousConnector(object):
def _response_handle_allocated(self, msg): def _response_handle_allocated(self, msg):
nameplate = msg["nameplate"] nameplate = msg["nameplate"]
assert isinstance(nameplate, type("")), type(nameplate) assert isinstance(nameplate, type("")), type(nameplate)
self._C.rx_allocated(nameplate) self._A.rx_allocated(nameplate)
def _response_handle_nameplates(self, msg): def _response_handle_nameplates(self, msg):
# we get list of {id: ID}, with maybe more attributes in the future
nameplates = msg["nameplates"] nameplates = msg["nameplates"]
assert isinstance(nameplates, list), type(nameplates) assert isinstance(nameplates, list), type(nameplates)
nids = [] nids = set()
for n in nameplates: for n in nameplates:
assert isinstance(n, dict), type(n) assert isinstance(n, dict), type(n)
nameplate_id = n["id"] nameplate_id = n["id"]
assert isinstance(nameplate_id, type("")), type(nameplate_id) assert isinstance(nameplate_id, type("")), type(nameplate_id)
nids.append(nameplate_id) nids.add(nameplate_id)
# deliver a set of nameplate ids
self._L.rx_nameplates(nids) self._L.rx_nameplates(nids)
def _response_handle_ack(self, msg): def _response_handle_ack(self, msg):

View File

@ -165,10 +165,10 @@ class PGPWordList(object):
words = even_words_lowercase words = even_words_lowercase
last_partial_word = prefix.split("-")[-1] last_partial_word = prefix.split("-")[-1]
lp = len(last_partial_word) lp = len(last_partial_word)
completions = [] completions = set()
for word in words: for word in words:
if word.startswith(prefix): if word.startswith(prefix):
completions.append(word[lp:]) completions.add(word[lp:])
return completions return completions
def choose_words(self, length): def choose_words(self, length):

View File

@ -57,6 +57,16 @@ class NoKeyError(WormholeError):
class OnlyOneCodeError(WormholeError): class OnlyOneCodeError(WormholeError):
"""Only one w.generate_code/w.set_code/w.input_code may be called""" """Only one w.generate_code/w.set_code/w.input_code may be called"""
class MustChooseNameplateFirstError(WormholeError):
"""The InputHelper was asked to do get_word_completions() or
choose_words() before the nameplate was chosen."""
class AlreadyChoseNameplateError(WormholeError):
"""The InputHelper was asked to do get_nameplate_completions() after
choose_nameplate() was called, or choose_nameplate() was called a second
time."""
class AlreadyChoseWordsError(WormholeError):
"""The InputHelper was asked to do get_word_completions() after
choose_words() was called, or choose_words() was called a second time."""
class WormholeClosed(Exception): class WormholeClosed(Exception):
"""Deferred-returning API calls errback with WormholeClosed if the """Deferred-returning API calls errback with WormholeClosed if the
wormhole was already closed, or if it closes before a real result can be wormhole was already closed, or if it closes before a real result can be

View File

@ -1,14 +1,23 @@
from __future__ import print_function, unicode_literals from __future__ import print_function, unicode_literals
import json import json
from zope.interface import directlyProvides from zope.interface import directlyProvides, implementer
from twisted.trial import unittest from twisted.trial import unittest
from .. import timing, _order, _receive, _key, _code from .. import errors, timing, _order, _receive, _key, _code, _lister, _input, _allocator
from .._interfaces import (IKey, IReceive, IBoss, ISend, IMailbox, from .._interfaces import (IKey, IReceive, IBoss, ISend, IMailbox,
IRendezvousConnector, ILister) IRendezvousConnector, ILister, IInput, IAllocator,
INameplate, ICode, IWordlist)
from .._key import derive_key, derive_phase_key, encrypt_data from .._key import derive_key, derive_phase_key, encrypt_data
from ..util import dict_to_bytes, hexstr_to_bytes, bytes_to_hexstr, to_bytes from ..util import dict_to_bytes, hexstr_to_bytes, bytes_to_hexstr, to_bytes
from spake2 import SPAKE2_Symmetric from spake2 import SPAKE2_Symmetric
@implementer(IWordlist)
class FakeWordList(object):
def choose_words(self, length):
return "-".join(["word"] * length)
def get_completions(self, prefix):
self._get_completions_prefix = prefix
return self._completions
class Dummy: class Dummy:
def __init__(self, name, events, iface, *meths): def __init__(self, name, events, iface, *meths):
self.name = name self.name = name
@ -194,52 +203,260 @@ class Code(unittest.TestCase):
events = [] events = []
c = _code.Code(timing.DebugTiming()) c = _code.Code(timing.DebugTiming())
b = Dummy("b", events, IBoss, "got_code") b = Dummy("b", events, IBoss, "got_code")
a = Dummy("a", events, IAllocator, "allocate")
n = Dummy("n", events, INameplate, "set_nameplate")
k = Dummy("k", events, IKey, "got_code")
i = Dummy("i", events, IInput, "start")
c.wire(b, a, n, k, i)
return c, b, a, n, k, i, events
def test_set_code(self):
c, b, a, n, k, i, events = self.build()
c.set_code(u"1-code")
self.assertEqual(events, [("n.set_nameplate", u"1"),
("k.got_code", u"1-code"),
("b.got_code", u"1-code"),
])
def test_allocate_code(self):
c, b, a, n, k, i, events = self.build()
wl = FakeWordList()
c.allocate_code(2, wl)
self.assertEqual(events, [("a.allocate", 2, wl)])
events[:] = []
c.allocated("1", "1-code")
self.assertEqual(events, [("n.set_nameplate", u"1"),
("k.got_code", u"1-code"),
("b.got_code", u"1-code"),
])
def test_input_code(self):
c, b, a, n, k, i, events = self.build()
c.input_code()
self.assertEqual(events, [("i.start",)])
events[:] = []
c.got_nameplate("1")
self.assertEqual(events, [("n.set_nameplate", u"1"),
])
events[:] = []
c.finished_input("1-code")
self.assertEqual(events, [("k.got_code", u"1-code"),
("b.got_code", u"1-code"),
])
class Input(unittest.TestCase):
def build(self):
events = []
i = _input.Input(timing.DebugTiming())
c = Dummy("c", events, ICode, "got_nameplate", "finished_input")
l = Dummy("l", events, ILister, "refresh")
i.wire(c, l)
return i, c, l, events
def test_ignore_completion(self):
i, c, l, events = self.build()
helper = i.start()
self.assertIsInstance(helper, _input.Helper)
self.assertEqual(events, [("l.refresh",)])
events[:] = []
with self.assertRaises(errors.MustChooseNameplateFirstError):
helper.choose_words("word-word")
helper.choose_nameplate("1")
self.assertEqual(events, [("c.got_nameplate", "1")])
events[:] = []
with self.assertRaises(errors.AlreadyChoseNameplateError):
helper.choose_nameplate("2")
helper.choose_words("word-word")
with self.assertRaises(errors.AlreadyChoseWordsError):
helper.choose_words("word-word")
self.assertEqual(events, [("c.finished_input", "1-word-word")])
def test_with_completion(self):
i, c, l, events = self.build()
helper = i.start()
self.assertIsInstance(helper, _input.Helper)
self.assertEqual(events, [("l.refresh",)])
events[:] = []
helper.refresh_nameplates()
self.assertEqual(events, [("l.refresh",)])
events[:] = []
with self.assertRaises(errors.MustChooseNameplateFirstError):
helper.get_word_completions("prefix")
i.got_nameplates({"1", "12", "34", "35", "367"})
self.assertEqual(helper.get_nameplate_completions(""),
{"1", "12", "34", "35", "367"})
self.assertEqual(helper.get_nameplate_completions("1"),
{"", "2"})
self.assertEqual(helper.get_nameplate_completions("2"), set())
self.assertEqual(helper.get_nameplate_completions("3"),
{"4", "5", "67"})
helper.choose_nameplate("34")
with self.assertRaises(errors.AlreadyChoseNameplateError):
helper.refresh_nameplates()
with self.assertRaises(errors.AlreadyChoseNameplateError):
helper.get_nameplate_completions("1")
self.assertEqual(events, [("c.got_nameplate", "34")])
events[:] = []
# no wordlist yet
self.assertEqual(helper.get_word_completions(""), set())
wl = FakeWordList()
i.got_wordlist(wl)
wl._completions = {"bc", "bcd", "e"}
self.assertEqual(helper.get_word_completions("a"), wl._completions)
self.assertEqual(wl._get_completions_prefix, "a")
with self.assertRaises(errors.AlreadyChoseNameplateError):
helper.refresh_nameplates()
with self.assertRaises(errors.AlreadyChoseNameplateError):
helper.get_nameplate_completions("1")
helper.choose_words("word-word")
with self.assertRaises(errors.AlreadyChoseWordsError):
helper.get_word_completions("prefix")
with self.assertRaises(errors.AlreadyChoseWordsError):
helper.choose_words("word-word")
self.assertEqual(events, [("c.finished_input", "34-word-word")])
class Lister(unittest.TestCase):
def build(self):
events = []
l = _lister.Lister(timing.DebugTiming())
rc = Dummy("rc", events, IRendezvousConnector, "tx_list")
i = Dummy("i", events, IInput, "got_nameplates")
l.wire(rc, i)
return l, rc, i, events
def test_connect_first(self):
l, rc, i, events = self.build()
l.connected()
l.lost()
l.connected()
self.assertEqual(events, [])
l.refresh()
self.assertEqual(events, [("rc.tx_list",),
])
events[:] = []
l.rx_nameplates({"1", "2", "3"})
self.assertEqual(events, [("i.got_nameplates", {"1", "2", "3"}),
])
events[:] = []
# now we're satisfied: disconnecting and reconnecting won't ask again
l.lost()
l.connected()
self.assertEqual(events, [])
# but if we're told to refresh, we'll do so
l.refresh()
self.assertEqual(events, [("rc.tx_list",),
])
def test_connect_first_ask_twice(self):
l, rc, i, events = self.build()
l.connected()
self.assertEqual(events, [])
l.refresh()
l.refresh()
self.assertEqual(events, [("rc.tx_list",),
("rc.tx_list",),
])
l.rx_nameplates({"1", "2", "3"})
self.assertEqual(events, [("rc.tx_list",),
("rc.tx_list",),
("i.got_nameplates", {"1", "2", "3"}),
])
l.rx_nameplates({"1" ,"2", "3", "4"})
self.assertEqual(events, [("rc.tx_list",),
("rc.tx_list",),
("i.got_nameplates", {"1", "2", "3"}),
("i.got_nameplates", {"1", "2", "3", "4"}),
])
def test_reconnect(self):
l, rc, i, events = self.build()
l.refresh()
l.connected()
self.assertEqual(events, [("rc.tx_list",),
])
events[:] = []
l.lost()
l.connected()
self.assertEqual(events, [("rc.tx_list",),
])
def test_refresh_first(self):
l, rc, i, events = self.build()
l.refresh()
self.assertEqual(events, [])
l.connected()
self.assertEqual(events, [("rc.tx_list",),
])
l.rx_nameplates({"1", "2", "3"})
self.assertEqual(events, [("rc.tx_list",),
("i.got_nameplates", {"1", "2", "3"}),
])
def test_unrefreshed(self):
l, rc, i, events = self.build()
self.assertEqual(events, [])
# we receive a spontaneous rx_nameplates, without asking
l.connected()
self.assertEqual(events, [])
l.rx_nameplates({"1", "2", "3"})
self.assertEqual(events, [("i.got_nameplates", {"1", "2", "3"}),
])
class Allocator(unittest.TestCase):
def build(self):
events = []
a = _allocator.Allocator(timing.DebugTiming())
rc = Dummy("rc", events, IRendezvousConnector, "tx_allocate") rc = Dummy("rc", events, IRendezvousConnector, "tx_allocate")
l = Dummy("l", events, ILister, "refresh_nameplates") c = Dummy("c", events, ICode, "allocated")
c.wire(b, rc, l) a.wire(rc, c)
return c, b, rc, l, events return a, rc, c, events
def test_set_disconnected(self): def test_no_allocation(self):
c, b, rc, l, events = self.build() a, rc, c, events = self.build()
c.set_code(u"code") a.connected()
self.assertEqual(events, [("b.got_code", u"code")])
def test_set_connected(self):
c, b, rc, l, events = self.build()
c.connected()
c.set_code(u"code")
self.assertEqual(events, [("b.got_code", u"code")])
def test_allocate_disconnected(self):
c, b, rc, l, events = self.build()
c.allocate_code(2)
self.assertEqual(events, []) self.assertEqual(events, [])
c.connected()
self.assertEqual(events, [("rc.tx_allocate",)]) def test_allocate_first(self):
events[:] = [] a, rc, c, events = self.build()
c.lost() a.allocate(2, FakeWordList())
self.assertEqual(events, []) self.assertEqual(events, [])
c.connected() a.connected()
self.assertEqual(events, [("rc.tx_allocate",)]) self.assertEqual(events, [("rc.tx_allocate",)])
events[:] = [] events[:] = []
c.rx_allocated("4") a.lost()
self.assertEqual(len(events), 1, events) a.connected()
self.assertEqual(events[0][0], "b.got_code") self.assertEqual(events, [("rc.tx_allocate",),
code = events[0][1] ])
self.assert_(code.startswith("4-"), code) events[:] = []
a.rx_allocated("1")
self.assertEqual(events, [("c.allocated", "1", "1-word-word"),
])
def test_allocate_connected(self): def test_connect_first(self):
c, b, rc, l, events = self.build() a, rc, c, events = self.build()
c.connected() a.connected()
c.allocate_code(2) self.assertEqual(events, [])
a.allocate(2, FakeWordList())
self.assertEqual(events, [("rc.tx_allocate",)]) self.assertEqual(events, [("rc.tx_allocate",)])
events[:] = [] events[:] = []
c.rx_allocated("4") a.lost()
self.assertEqual(len(events), 1, events) a.connected()
self.assertEqual(events[0][0], "b.got_code") self.assertEqual(events, [("rc.tx_allocate",),
code = events[0][1] ])
self.assert_(code.startswith("4-"), code) events[:] = []
a.rx_allocated("1")
# TODO: input_code self.assertEqual(events, [("c.allocated", "1", "1-word-word"),
])
# TODO
# Send
# Mailbox
# Nameplate
# Terminator
# Boss
# RendezvousConnector (not a state machine)
# #Input: exercise helper methods
# wordlist