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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
# -*- coding: utf-8 -*-
import errno
import socket
import unittest
from unittest.mock import Mock, patch
from websocket._socket import recv
from websocket._ssl_compat import SSLWantReadError
from websocket._exceptions import (
WebSocketTimeoutException,
WebSocketConnectionClosedException,
)
"""
test_socket_bugs.py
websocket - WebSocket client library for Python
Copyright 2025 engn33r
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
class SocketBugsTest(unittest.TestCase):
"""Test bugs found in socket handling logic"""
def test_bug_implicit_none_return_from_ssl_want_read_fixed(self):
"""
BUG #5 FIX VERIFICATION: Test SSLWantReadError timeout now raises correct exception
Bug was in _socket.py:100-101 - SSLWantReadError except block returned None implicitly
Fixed: Now properly handles timeout with WebSocketTimeoutException
"""
mock_sock = Mock()
mock_sock.recv.side_effect = SSLWantReadError()
mock_sock.gettimeout.return_value = 1.0
with patch("selectors.DefaultSelector") as mock_selector_class:
mock_selector = Mock()
mock_selector_class.return_value = mock_selector
mock_selector.select.return_value = [] # Timeout - no data ready
with self.assertRaises(WebSocketTimeoutException) as cm:
recv(mock_sock, 100)
# Verify correct timeout exception and message
self.assertIn("Connection timed out waiting for data", str(cm.exception))
def test_bug_implicit_none_return_from_socket_error_fixed(self):
"""
BUG #5 FIX VERIFICATION: Test that socket.error with EAGAIN now handles timeout correctly
Bug was in _socket.py:102-105 - socket.error except block returned None implicitly
Fixed: Now properly handles timeout with WebSocketTimeoutException
"""
mock_sock = Mock()
# Create socket error with EAGAIN (should be retried)
eagain_error = OSError(errno.EAGAIN, "Resource temporarily unavailable")
# First call raises EAGAIN, selector times out on retry
mock_sock.recv.side_effect = eagain_error
mock_sock.gettimeout.return_value = 1.0
with patch("selectors.DefaultSelector") as mock_selector_class:
mock_selector = Mock()
mock_selector_class.return_value = mock_selector
mock_selector.select.return_value = [] # Timeout - no data ready
with self.assertRaises(WebSocketTimeoutException) as cm:
recv(mock_sock, 100)
# Verify correct timeout exception and message
self.assertIn("Connection timed out waiting for data", str(cm.exception))
def test_bug_wrong_exception_for_selector_timeout_fixed(self):
"""
BUG #6 FIX VERIFICATION: Test that selector timeout now raises correct exception type
Bug was in _socket.py:115 returning None for timeout, treated as connection error
Fixed: Now raises WebSocketTimeoutException directly
"""
mock_sock = Mock()
mock_sock.recv.side_effect = SSLWantReadError() # Trigger retry path
mock_sock.gettimeout.return_value = 1.0
with patch("selectors.DefaultSelector") as mock_selector_class:
mock_selector = Mock()
mock_selector_class.return_value = mock_selector
mock_selector.select.return_value = [] # TIMEOUT - this is key!
with self.assertRaises(WebSocketTimeoutException) as cm:
recv(mock_sock, 100)
# Verify it's the correct timeout exception with proper message
self.assertIn("Connection timed out waiting for data", str(cm.exception))
# This proves the fix works:
# 1. selector.select() returns [] (timeout)
# 2. _recv() now raises WebSocketTimeoutException directly
# 3. No more misclassification as connection closed error!
def test_socket_timeout_exception_handling(self):
"""
Test that socket.timeout exceptions are properly handled
"""
mock_sock = Mock()
mock_sock.gettimeout.return_value = 1.0
# Simulate a real socket.timeout scenario
mock_sock.recv.side_effect = socket.timeout("Operation timed out")
# This works correctly - socket.timeout raises WebSocketTimeoutException
with self.assertRaises(WebSocketTimeoutException) as cm:
recv(mock_sock, 100)
# In Python 3.10+, socket.timeout is a subclass of TimeoutError
# so it's caught by the TimeoutError handler with hardcoded message
# In Python 3.9, socket.timeout is caught by socket.timeout handler
# which preserves the original message
import sys
if sys.version_info >= (3, 10):
self.assertIn("Connection timed out", str(cm.exception))
else:
self.assertIn("Operation timed out", str(cm.exception))
def test_correct_ssl_want_read_retry_behavior(self):
"""Test the correct behavior when SSLWantReadError is properly handled"""
mock_sock = Mock()
# First call raises SSLWantReadError, second call succeeds
mock_sock.recv.side_effect = [SSLWantReadError(), b"data after retry"]
mock_sock.gettimeout.return_value = 1.0
with patch("selectors.DefaultSelector") as mock_selector_class:
mock_selector = Mock()
mock_selector_class.return_value = mock_selector
mock_selector.select.return_value = [True] # Data ready after wait
# This should work correctly
result = recv(mock_sock, 100)
self.assertEqual(result, b"data after retry")
# Selector should be used for retry
mock_selector.register.assert_called()
mock_selector.select.assert_called()
if __name__ == "__main__":
unittest.main()
|
