TestProtocolHandler.py
4.28 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
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
import unittest
import mock
from os.path import dirname, realpath
from sys import argv, path
path.append(dirname(dirname(realpath(__file__))) + '/lib')
from Communication.ProtocolHandler import ProtocolHandler
class TestProtocolHandler(unittest.TestCase):
def setUp(self):
self._protocol_handler = ProtocolHandler()
self._endpoint = mock.Mock()
self._message = mock.Mock()
self._event = mock.Mock()
self._protocol = mock.Mock()
self._protocol_handler.issueEvent = mock.Mock()
self._event.subject = self._endpoint
self._endpoint.__iter__ = mock.Mock(return_value=iter([self._message]))
self._endpoint.getProtocol.return_value = self._protocol
def test_parseCloseResponse(self):
self._message.isCLoseMessage.return_value = True
self._message.isResponse.return_value = True
self._protocol_handler._parse(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'new_msg', self._message)
self._endpoint.setClose.assert_called_with()
def test_parseCloseRequest(self):
self._message.isCLoseMessage.return_value = True
self._message.isResponse.return_value = False
self._protocol_handler._parse(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'new_msg', self._message)
def test_parseUpgradeResponse(self):
self._message.isCloseMessage.return_value = False
self._message.isUpgradeMessage.return_value = True
self._message.isRequest.return_value = False
self._protocol_handler._parse(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'upgrade', self._message)
def test_parseUpgradeRequest(self):
response = mock.Mock()
self._protocol.createUpgradeResponse.return_value = response
self._message.isCloseMessage.return_value = False
self._message.isUpgradeMessage.return_value = True
self._message.isRequest.return_value = True
self._protocol_handler._parse(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'send_msg', response)
def test_parseNormalMessage(self):
self._message.isCloseMessage.return_value = False
self._message.isUpgradeMessage.return_value = False
self._protocol_handler._parse(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'new_msg', self._message)
def test_composeRequest(self):
self._event.data = self._message
self._endpoint.compose.return_value = True
self._message.isResponse.return_value = False
self._protocol_handler._compose(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'write_ready')
def test_composeUpgradeResponse(self):
self._event.data = self._message
self._endpoint.compose.return_value = True
self._message.isResponse.return_value = True
self._message.isUpgradeMessage.return_value = True
self._message.isCloseMessage.return_value = False
self._protocol_handler._compose(self._event)
self._protocol_handler.issueEvent.assert_any_call(
self._endpoint, 'write_ready')
self._protocol_handler.issueEvent.assert_any_call(
self._endpoint, 'upgrade', self._message)
def test_composeCloseResponse(self):
self._event.data = self._message
self._endpoint.compose.return_value = True
self._message.isResponse.return_value = True
self._message.isUpgradeMessage.return_value = False
self._message.isCloseMessage.return_value = True
self._protocol_handler._compose(self._event)
self._protocol_handler.issueEvent.assert_called_with(
self._endpoint, 'write_ready')
self._endpoint.setClose.assert_called_once_with()
def suite():
return unittest.TestLoader().loadTestsFromTestCase(TestProtocolHandler)
if __name__ == '__main__':
unittest.TextTestRunner(verbosity=2).run(suite())
# vim: set ft=python et ts=8 sw=4 sts=4: