open.source/magic-wormhole-transit-relay
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

post-rebase: update newer tests

Browse Source
This commit is contained in:
Brian Warner 2020-05-23 17:50:26 -07:00
parent 45824ca5d6
commit ca309d5283
1 changed files with 12 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
src/wormhole_transit_relay/test/test_transit_server.py
View File

@ -1,15 +1,9 @@
from __future__ import print_function, unicode_literals from __future__ import print_function, unicode_literals
from binascii import hexlify from binascii import hexlify
from twisted.trial import unittest from twisted.trial import unittest
from twisted.internet import reactor, defer
from .common import ServerBase from .common import ServerBase
from .. import transit_server from .. import transit_server
def wait():
d = defer.Deferred()
reactor.callLater(0.001, d.callback, None)
return d
def handshake(token, side=None): def handshake(token, side=None):
hs = b"please relay " + hexlify(token) hs = b"please relay " + hexlify(token)
if side is not None: if side is not None:
@ -296,22 +290,16 @@ class _Transit:
p1.transport.loseConnection() p1.transport.loseConnection()
@defer.inlineCallbacks
def test_short_handshake(self): def test_short_handshake(self):
ep = clientFromString(reactor, self.transit) p1 = self.new_protocol()
a1 = yield connectProtocol(ep, Accumulator())
# hang up before sending a complete handshake # hang up before sending a complete handshake
a1.transport.write(b"short") p1.dataReceived(b"short")
a1.transport.loseConnection() p1.transport.loseConnection()
@defer.inlineCallbacks
def test_empty_handshake(self): def test_empty_handshake(self):
ep = clientFromString(reactor, self.transit) p1 = self.new_protocol()
a1 = yield connectProtocol(ep, Accumulator())
# hang up before sending anything # hang up before sending anything
a1.transport.loseConnection() p1.transport.loseConnection()
class TransitWithLogs(_Transit, ServerBase, unittest.TestCase): class TransitWithLogs(_Transit, ServerBase, unittest.TestCase):
log_requests = True log_requests = True
@ -328,46 +316,27 @@ class Usage(ServerBase, unittest.TestCase):
total_time, waiting_time)) total_time, waiting_time))
self._transit_server.recordUsage = record self._transit_server.recordUsage = record
@defer.inlineCallbacks
def test_empty(self): def test_empty(self):
ep = clientFromString(reactor, self.transit) p1 = self.new_protocol()
a1 = yield connectProtocol(ep, Accumulator())
# hang up before sending anything # hang up before sending anything
a1.transport.loseConnection() p1.transport.loseConnection()
yield a1._disconnect
# give the server a chance to react. in most of the other tests, the
# server hangs up on us, so this test needs extra synchronization
while len(self._usage) == 0:
d = defer.Deferred()
reactor.callLater(0.01, d.callback, None)
yield d
# that will log the "empty" usage event # that will log the "empty" usage event
self.assertEqual(len(self._usage), 1, self._usage) self.assertEqual(len(self._usage), 1, self._usage)
(started, result, total_bytes, total_time, waiting_time) = self._usage[0] (started, result, total_bytes, total_time, waiting_time) = self._usage[0]
self.assertEqual(result, "empty", self._usage) self.assertEqual(result, "empty", self._usage)
@defer.inlineCallbacks
def test_short(self): def test_short(self):
ep = clientFromString(reactor, self.transit) p1 = self.new_protocol()
a1 = yield connectProtocol(ep, Accumulator())
# hang up before sending a complete handshake # hang up before sending a complete handshake
a1.transport.write(b"short") p1.transport.write(b"short")
a1.transport.loseConnection() p1.transport.loseConnection()
yield a1._disconnect
# give the server a chance to react. in most of the other tests, the
# server hangs up on us, so this test needs extra synchronization
while len(self._usage) == 0:
d = defer.Deferred()
reactor.callLater(0.01, d.callback, None)
yield d
# that will log the "empty" usage event # that will log the "empty" usage event
self.assertEqual(len(self._usage), 1, self._usage) self.assertEqual(len(self._usage), 1, self._usage)
(started, result, total_bytes, total_time, waiting_time) = self._usage[0] (started, result, total_bytes, total_time, waiting_time) = self._usage[0]
self.assertEqual(result, "empty", self._usage) self.assertEqual(result, "empty", self._usage)
@defer.inlineCallbacks
def test_errory(self): def test_errory(self):
p1 = self.new_protocol() p1 = self.new_protocol()