1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
# Copyright (c) str4d <str4d@mail.i2p>
# See COPYING for details.
from builtins import object
from twisted.internet import defer
from twisted.internet.error import ConnectionLost, ConnectionRefusedError
from twisted.python import failure
from twisted.test import proto_helpers
from twisted.trial import unittest
from txi2p.bob.factory import (BOBI2PClientFactory,
BOBI2PServerFactory,
BOBClientFactoryWrapper,
BOBServerFactoryWrapper)
from txi2p.test.util import FakeFactory
connectionLostFailure = failure.Failure(ConnectionLost())
connectionRefusedFailure = failure.Failure(ConnectionRefusedError())
class BOBFactoryTestMixin(object):
def setUp(self):
self.aborted = []
def makeProto(self, *a, **kw):
fac = self.factory(*a, **kw)
proto = fac.buildProtocol(None)
transport = proto_helpers.StringTransport()
transport.abortConnection = lambda: self.aborted.append(True)
proto.makeConnection(transport)
return fac, proto
def test_cancellation(self):
fac, proto = self.makeProto(None, None, None, '')
fac.deferred.cancel()
self.assert_(self.aborted)
return self.assertFailure(fac.deferred, defer.CancelledError)
def test_cancellationBeforeFailure(self):
fac, proto = self.makeProto(None, None, None, '')
fac.deferred.cancel()
proto.connectionLost(connectionLostFailure)
self.assert_(self.aborted)
return self.assertFailure(fac.deferred, defer.CancelledError)
def test_cancellationAfterFailure(self):
fac, proto = self.makeProto(None, None, None, '')
proto.connectionLost(connectionLostFailure)
fac.deferred.cancel()
self.assertFalse(self.aborted)
return self.assertFailure(fac.deferred, ConnectionLost)
def test_clientConnectionFailed(self):
fac, proto = self.makeProto(None, None, None, '')
fac.clientConnectionFailed(None, connectionRefusedFailure)
return self.assertFailure(fac.deferred, ConnectionRefusedError)
def test_defaultFactoryListsTunnels(self):
fac, proto = self.makeProto(None, None, None, '')
proto.dataReceived('BOB 00.00.10\nOK\n')
self.assertEqual(proto.transport.value(), b'list\n')
class TestBOBI2PClientFactory(BOBFactoryTestMixin, unittest.TestCase):
factory = BOBI2PClientFactory
def TODO_test_noProtocolFromWrappedFactory(self):
wrappedFac = FakeFactory(returnNoProtocol=True)
mreactor = proto_helpers.MemoryReactor()
fac, proto = self.makeProto(mreactor, wrappedFac, None, '')
fac.tunnelNick = 'spam'
# Shortcut to end of BOB protocol
proto.receiver.currentRule = 'State_start'
proto._parser._setupInterp()
proto.dataReceived('OK HTTP 418\n')
self.assert_(self.aborted) # TODO: Check the Deferred chain
return self.assertFailure(fac.deferred, defer.CancelledError)
class TestBOBI2PServerFactory(BOBFactoryTestMixin, unittest.TestCase):
factory = BOBI2PServerFactory
def TODO_test_noProtocolFromWrappedFactory(self):
wrappedFac = FakeFactory(returnNoProtocol=True)
mreactor = proto_helpers.MemoryReactor()
fac, proto = self.makeProto(mreactor, wrappedFac, None, '')
fac.tunnelNick = 'spam'
fac.localDest = 'spam.i2p'
# Shortcut to end of BOB protocol
proto.receiver.currentRule = 'State_start'
proto._parser._setupInterp()
proto.dataReceived('OK HTTP 418\n')
self.assert_(self.aborted) # TODO: Check the Deferred chain
return self.assertFailure(fac.deferred, defer.CancelledError)
class TestBOBClientFactoryWrapper(unittest.TestCase):
def test_buildProtocol(self):
wrappedFac = FakeFactory()
fac = BOBClientFactoryWrapper(wrappedFac, None, None, '', True)
fac.setDest('spam.i2p')
proto = fac.buildProtocol(None)
self.assertEqual(proto.wrappedProto.factory, wrappedFac)
class TestBOBServerFactoryWrapper(unittest.TestCase):
def test_buildProtocol(self):
wrappedFac = FakeFactory()
fac = BOBServerFactoryWrapper(wrappedFac, None, None, '', True)
proto = fac.buildProtocol(None)
self.assertEqual(proto.wrappedProto.factory, wrappedFac)
|
