b3ba539cbddde39753a66911b61568beadbb8196
[WebKit-https.git] / WebKitTools / pywebsocket / test / test_msgutil.py
1 #!/usr/bin/env python
2 #
3 # Copyright 2009, Google Inc.
4 # All rights reserved.
5 #
6 # Redistribution and use in source and binary forms, with or without
7 # modification, are permitted provided that the following conditions are
8 # met:
9 #
10 #     * Redistributions of source code must retain the above copyright
11 # notice, this list of conditions and the following disclaimer.
12 #     * Redistributions in binary form must reproduce the above
13 # copyright notice, this list of conditions and the following disclaimer
14 # in the documentation and/or other materials provided with the
15 # distribution.
16 #     * Neither the name of Google Inc. nor the names of its
17 # contributors may be used to endorse or promote products derived from
18 # this software without specific prior written permission.
19 #
20 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31
32
33 """Tests for msgutil module."""
34
35
36 import Queue
37 import unittest
38
39 import config  # This must be imported before mod_pywebsocket.
40 from mod_pywebsocket import msgutil
41
42 import mock
43
44
45 def _create_request(read_data):
46     return mock.MockRequest(connection=mock.MockConn(read_data))
47
48 def _create_blocking_request():
49     return mock.MockRequest(connection=mock.MockBlockingConn())
50
51 class MessageTest(unittest.TestCase):
52     def test_send_message(self):
53         request = _create_request('')
54         msgutil.send_message(request, 'Hello')
55         self.assertEqual('\x00Hello\xff', request.connection.written_data())
56
57     def test_send_message_unicode(self):
58         request = _create_request('')
59         msgutil.send_message(request, u'\u65e5')
60         # U+65e5 is encoded as e6,97,a5 in UTF-8
61         self.assertEqual('\x00\xe6\x97\xa5\xff',
62                          request.connection.written_data())
63
64     def test_receive_message(self):
65         request = _create_request('\x00Hello\xff\x00World!\xff')
66         self.assertEqual('Hello', msgutil.receive_message(request))
67         self.assertEqual('World!', msgutil.receive_message(request))
68
69     def test_receive_message_unicode(self):
70         request = _create_request('\x00\xe6\x9c\xac\xff')
71         # U+672c is encoded as e6,9c,ac in UTF-8
72         self.assertEqual(u'\u672c', msgutil.receive_message(request))
73
74     def test_receive_message_discard(self):
75         request = _create_request('\x80\x06IGNORE\x00Hello\xff'
76                                 '\x01DISREGARD\xff\x00World!\xff')
77         self.assertEqual('Hello', msgutil.receive_message(request))
78         self.assertEqual('World!', msgutil.receive_message(request))
79
80     def test_payload_length(self):
81         for length, bytes in ((0, '\x00'), (0x7f, '\x7f'), (0x80, '\x81\x00'),
82                               (0x1234, '\x80\xa4\x34')):
83             self.assertEqual(length,
84                              msgutil._payload_length(_create_request(bytes)))
85
86     def test_receive_bytes(self):
87         request = _create_request('abcdefg')
88         self.assertEqual('abc', msgutil._receive_bytes(request, 3))
89         self.assertEqual('defg', msgutil._receive_bytes(request, 4))
90
91     def test_read_until(self):
92         request = _create_request('abcXdefgX')
93         self.assertEqual('abc', msgutil._read_until(request, 'X'))
94         self.assertEqual('defg', msgutil._read_until(request, 'X'))
95
96
97 class MessageReceiverTest(unittest.TestCase):
98     def test_queue(self):
99         request = _create_blocking_request()
100         receiver = msgutil.MessageReceiver(request)
101
102         self.assertEqual(None, receiver.receive_nowait())
103
104         request.connection.put_bytes('\x00Hello!\xff')
105         self.assertEqual('Hello!', receiver.receive())
106
107     def test_onmessage(self):
108         onmessage_queue = Queue.Queue()
109         def onmessage_handler(message):
110             onmessage_queue.put(message)
111
112         request = _create_blocking_request()
113         receiver = msgutil.MessageReceiver(request, onmessage_handler)
114
115         request.connection.put_bytes('\x00Hello!\xff')
116         self.assertEqual('Hello!', onmessage_queue.get())
117
118
119 class MessageSenderTest(unittest.TestCase):
120     def test_send(self):
121         request = _create_blocking_request()
122         sender = msgutil.MessageSender(request)
123
124         sender.send('World')
125         self.assertEqual('\x00World\xff', request.connection.written_data())
126
127     def test_send_nowait(self):
128         # Use a queue to check the bytes written by MessageSender.
129         # request.connection.written_data() cannot be used here because
130         # MessageSender runs in a separate thread.
131         send_queue = Queue.Queue()
132         def write(bytes):
133             send_queue.put(bytes)
134         request = _create_blocking_request()
135         request.connection.write = write
136
137         sender = msgutil.MessageSender(request)
138
139         sender.send_nowait('Hello')
140         sender.send_nowait('World')
141         self.assertEqual('\x00Hello\xff', send_queue.get())
142         self.assertEqual('\x00World\xff', send_queue.get())
143
144
145 if __name__ == '__main__':
146     unittest.main()
147
148
149 # vi:sts=4 sw=4 et