You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
59 lines
2.2 KiB
59 lines
2.2 KiB
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:
|