TestDatagramService.py
4.96 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
103
104
105
106
107
108
109
110
111
112
113
import unittest
import mock
from os.path import dirname, realpath
from sys import argv, path
path.append(dirname(dirname(realpath(__file__))) + '/lib')
from Communication.DatagramService import DatagramService
class TestDatagramService(unittest.TestCase):
def setUp(self):
self._transport = mock.Mock()
self._protocol = mock.Mock()
self._message = mock.Mock()
self._parser = mock.Mock()
self._composer = mock.Mock()
self._bufsize = 22655
self._msginfo = ('111222333', 1212)
self._datagram = DatagramService(
self._transport, self._protocol, self._bufsize)
self._protocol.getParser.return_value = self._parser
self._protocol.getComposer.return_value = self._composer
self._message.getRemote.return_value = self._msginfo[1]
def testHasPendingData(self):
self.assertFalse(self._datagram.hasPendingData())
self._datagram._write_buffer = '12345'
self.assertTrue(self._datagram.hasPendingData())
def testIterInit(self):
self.assertEqual(self._datagram.__iter__(), self._datagram)
def testMessageIteratorNoData(self):
self.assertRaises(StopIteration, self._datagram.next)
def testMessageIteratorCreateMessageFails(self):
self._datagram._read_buffer = mock.Mock()
self._datagram._read_buffer.popleft.return_value = self._msginfo
self._protocol.createMessage.return_value = None
self.assertRaises(StopIteration, self._datagram.next)
self._datagram._read_buffer.popleft.assert_called_once_with()
self._protocol.createMessage.assert_called_once_with(self._msginfo[1])
def testMessageIteratorNoDataParsed(self):
self._datagram._read_buffer = mock.Mock()
self._datagram._read_buffer.popleft.return_value = self._msginfo
self._protocol.createMessage.return_value = self._message
self._parser.parse.return_value = 0
self.assertRaises(StopIteration, self._datagram.next)
self._datagram._read_buffer.popleft.assert_called_once_with()
self._protocol.createMessage.assert_called_once_with(self._msginfo[1])
self._parser.parse.assert_called_once_with(
self._message, self._msginfo[0])
def testMessageIteratorGetMessage(self):
self._datagram._read_buffer = mock.Mock()
self._datagram._read_buffer.popleft.return_value = self._msginfo
self._protocol.createMessage.return_value = self._message
self._parser.parse.return_value = 10
self.assertEqual(self._datagram.next(), self._message)
self._datagram._read_buffer.popleft.assert_called_once_with()
self._protocol.createMessage.assert_called_once_with(self._msginfo[1])
self._parser.parse.assert_called_once_with(
self._message, self._msginfo[0])
def testComposeSuccess(self):
self._composer.compose.return_value = '111222333'
self.assertTrue(self._datagram.compose(self._message))
self.assertIn(
('111222333', self._msginfo[1]),
self._datagram._write_buffer)
def testComposeFail(self):
self._composer.compose.side_effect = Exception('Boom!')
self.assertFalse(self._datagram.compose(self._message))
self.assertFalse(self._datagram._write_buffer)
def testAppendReadData(self):
self._datagram.appendReadData(('111', 1212))
self.assertIn(('111', 1212), self._datagram._read_buffer)
self._datagram.appendReadData(('222', 1212))
self.assertIn(('111', 1212), self._datagram._read_buffer)
self.assertIn(('222', 1212), self._datagram._read_buffer)
self._datagram.appendReadData(('333', 1212))
self.assertIn(('111', 1212), self._datagram._read_buffer)
self.assertIn(('222', 1212), self._datagram._read_buffer)
self.assertIn(('333', 1212), self._datagram._read_buffer)
def testNextWriteData(self):
self.assertEqual(self._datagram.nextWriteData(), ('', None))
self._datagram._write_buffer.append(('111222333', 1212))
self.assertEqual(self._datagram.nextWriteData(), ('111222333', 1212))
self.assertNotIn(('111222333', 1212), self._datagram._write_buffer)
def testAppendWriteData(self):
self._datagram.appendWriteData(('111', 1212))
self.assertIn(('111', 1212), self._datagram._write_buffer)
self._datagram.appendWriteData(('222', 1212))
self.assertIn(('111', 1212), self._datagram._write_buffer)
self.assertIn(('222', 1212), self._datagram._write_buffer)
self._datagram.appendWriteData(('333', 1212))
self.assertIn(('111', 1212), self._datagram._write_buffer)
self.assertIn(('222', 1212), self._datagram._write_buffer)
self.assertIn(('333', 1212), self._datagram._write_buffer)
def suite():
return unittest.TestLoader().loadTestsFromTestCase(TestDatagramService)
if __name__ == '__main__':
unittest.TextTestRunner(verbosity=2).run(suite())
# vim: set ft=python et ts=8 sw=4 sts=4: