Browse files

get tests to work on py2.7

only install 'noiseprotocol' (which is necessary for dilation to work) if the
"dilate" feature is requested (e.g. `pip install magic-wormhole[dilate]`)
  • Loading branch information...
warner committed Dec 24, 2018
1 parent a458fe9 commit 29c269ac8d03e23898a8a3d057bda60bfcb11d57
@@ -48,13 +48,13 @@
"txtorcon >= 18.0.2", # 18.0.2 fixes py3.4 support
':sys_platform=="win32"': ["pywin32"],
"dev": ["mock", "tox", "pyflakes",
"dilate": ["noiseprotocol"],
@@ -0,0 +1,11 @@
from noise.exceptions import NoiseInvalidMessage
except ImportError:
class NoiseInvalidMessage(Exception):

from noise.connection import NoiseConnection
except ImportError:
# allow imports to work on py2.7, even if dilation doesn't
NoiseConnection = None
@@ -12,6 +12,7 @@
from import OneShotObserver
from .encode import to_be4, from_be4
from .roles import FOLLOWER
from ._noise import NoiseInvalidMessage

# InboundFraming is given data and returns Frames (Noise wire-side
# bytestrings). It handles the relay handshake and the prologue. The Frames it
@@ -346,7 +347,6 @@ def send_handshake(self):

def process_handshake(self, frame):
from noise.exceptions import NoiseInvalidMessage
payload = self._noise.read_message(frame)
# Noise can include unencrypted data in the handshake, but we don't
@@ -359,7 +359,6 @@ def process_handshake(self, frame):

def decrypt_message(self, frame):
from noise.exceptions import NoiseInvalidMessage
message = self._noise.decrypt(frame)
except NoiseInvalidMessage as e:
@@ -1,7 +1,6 @@
from __future__ import print_function, unicode_literals
from collections import defaultdict
from binascii import hexlify
import six
from attr import attrs, attrib
from attr.validators import instance_of, provides, optional
from automat import MethodicalMachine
@@ -15,13 +14,14 @@
from .._interfaces import IDilationConnector, IDilationManager
from ..timing import DebugTiming
from import EmptyableSet
from ..util import HKDF
from ..util import HKDF, to_unicode
from .connection import DilatedConnectionProtocol, KCM
from .roles import LEADER

from .._hints import (DirectTCPV1Hint, TorTCPV1Hint, RelayV1Hint,
parse_hint_argv, describe_hint_obj, endpoint_from_hint_obj,
from ._noise import NoiseConnection

def build_sided_relay_handshake(key, side):
@@ -37,14 +37,13 @@ def build_sided_relay_handshake(key, side):
NOISEPROTO = b"Noise_NNpsk0_25519_ChaChaPoly_BLAKE2s"

def build_noise():
from noise.connection import NoiseConnection
return NoiseConnection.from_name(NOISEPROTO)

class Connector(object):
_dilation_key = attrib(validator=instance_of(type(b"")))
_transit_relay_location = attrib(validator=optional(instance_of(str)))
_transit_relay_location = attrib(validator=optional(instance_of(type(u""))))
_manager = attrib(validator=provides(IDilationManager))
_reactor = attrib()
_eventual_queue = attrib()
@@ -265,7 +264,7 @@ def _listening(lp):
# lp is an IListeningPort
self._listeners.add(lp) # for shutdown and tests
portnum = lp.getHost().port
direct_hints = [DirectTCPV1Hint(six.u(addr), portnum, 0.0)
direct_hints = [DirectTCPV1Hint(to_unicode(addr), portnum, 0.0)
for addr in addresses]
@@ -7,8 +7,9 @@
from twisted.internet.defer import Deferred
from ...eventual import EventualQueue
from ..._interfaces import IDilationManager, IDilationConnector
from ..._dilation import roles
from ..._hints import DirectTCPV1Hint, RelayV1Hint, TorTCPV1Hint
from ..._dilation import roles
from ..._dilation._noise import NoiseConnection
from ..._dilation.connection import KCM
from ..._dilation.connector import (Connector,
@@ -101,6 +102,8 @@ def test_connection_abilities(self):

def test_build_noise(self):
if not NoiseConnection:
raise unittest.SkipTest("noiseprotocol unavailable")

def test_build_protocol_leader(self):
@@ -2,7 +2,7 @@
import mock
from zope.interface import alsoProvides
from twisted.trial import unittest
from noise.exceptions import NoiseInvalidMessage
from ..._dilation._noise import NoiseInvalidMessage
from ..._dilation.connection import (IFramer, Frame, Prologue,
_Record, Handshake,
Disconnect, Ping)
@@ -12,6 +12,10 @@ def HKDF(skm, outlen, salt=None, CTXinfo=b""):
def to_bytes(u):
return unicodedata.normalize("NFC", u).encode("utf-8")

def to_unicode(any):
if isinstance(any, type(u"")):
return any
return any.decode("ascii")

def bytes_to_hexstr(b):
assert isinstance(b, type(b""))

0 comments on commit 29c269a

Please sign in to comment.