TestConnector.py
2.21 KB
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
import unittest
import mock
from os.path import dirname, realpath
from sys import argv, path
path.append(dirname(dirname(realpath(__file__))) + '/lib')
from Communication.Connector import Connector
from Communication.Connection import Connection
from Transport import Transport
class TestConnector(unittest.TestCase):
def setUp(self):
self._connector = Connector()
self._connector.issueEvent = mock.Mock()
self._event = mock.Mock()
self._endpoint = mock.Mock()
self._protocol = mock.Mock()
self._dispatcher = mock.Mock()
self._new_transp = mock.Mock()
self._event.subject = self._endpoint
self._endpoint.getProtocol.return_value = self._protocol
def testTransportFail(self):
self._endpoint.accept.side_effect = Transport.Error(1)
self.assertTrue(self._connector._accept(self._event))
self._endpoint.getProtocol.assert_called_once_with()
self._endpoint.accept.called_once_with()
self._connector.issueEvent.assert_called_once_with(
self._endpoint, 'close')
def testNoNewTransports(self):
self._endpoint.accept.return_value = False
self.assertTrue(self._connector._accept(self._event))
self._endpoint.getProtocol.assert_called_once_with()
self._endpoint.accept.called_once_with()
self.assertFalse(self._connector.issueEvent.called)
def testNewTransports(self):
self._endpoint.accept.return_value = True
self._endpoint.pop.side_effect = iter([self._new_transp, False])
self.assertTrue(self._connector._accept(self._event))
self._endpoint.getProtocol.assert_called_once_with()
self._endpoint.accept.called_once_with()
issueEvent_args = self._connector.issueEvent.call_args
self.assertNotEqual(issueEvent_args, None)
if issueEvent_args:
self.assertIsInstance(issueEvent_args[0][0], Connection)
self.assertEqual(issueEvent_args[0][1], 'new_con')
def suite():
return unittest.TestLoader().loadTestsFromTestCase(TestConnector)
if __name__ == '__main__':
unittest.TextTestRunner(verbosity=2).run(suite())
# vim: set ft=python et ts=8 sw=4 sts=4: