codekingpro/portable-devtools

1# -*- coding: utf-8 -*-2import errno3import socket4import unittest5from unittest.mock import Mock, patch6 7from websocket._socket import recv8from websocket._ssl_compat import SSLWantReadError9from websocket._exceptions import (10    WebSocketTimeoutException,11    WebSocketConnectionClosedException,12)13 14"""15test_socket_bugs.py16websocket - WebSocket client library for Python17 18Copyright 2025 engn33r19 20Licensed under the Apache License, Version 2.0 (the "License");21you may not use this file except in compliance with the License.22You may obtain a copy of the License at23 24    http://www.apache.org/licenses/LICENSE-2.025 26Unless required by applicable law or agreed to in writing, software27distributed under the License is distributed on an "AS IS" BASIS,28WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.29See the License for the specific language governing permissions and30limitations under the License.31"""32 33class SocketBugsTest(unittest.TestCase):34    """Test bugs found in socket handling logic"""35 36    def test_bug_implicit_none_return_from_ssl_want_read_fixed(self):37        """38        BUG #5 FIX VERIFICATION: Test SSLWantReadError timeout now raises correct exception39 40        Bug was in _socket.py:100-101 - SSLWantReadError except block returned None implicitly41        Fixed: Now properly handles timeout with WebSocketTimeoutException42        """43        mock_sock = Mock()44        mock_sock.recv.side_effect = SSLWantReadError()45        mock_sock.gettimeout.return_value = 1.046 47        with patch("selectors.DefaultSelector") as mock_selector_class:48            mock_selector = Mock()49            mock_selector_class.return_value = mock_selector50            mock_selector.select.return_value = []  # Timeout - no data ready51 52            with self.assertRaises(WebSocketTimeoutException) as cm:53                recv(mock_sock, 100)54 55            # Verify correct timeout exception and message56            self.assertIn("Connection timed out waiting for data", str(cm.exception))57 58    def test_bug_implicit_none_return_from_socket_error_fixed(self):59        """60        BUG #5 FIX VERIFICATION: Test that socket.error with EAGAIN now handles timeout correctly61 62        Bug was in _socket.py:102-105 - socket.error except block returned None implicitly63        Fixed: Now properly handles timeout with WebSocketTimeoutException64        """65        mock_sock = Mock()66 67        # Create socket error with EAGAIN (should be retried)68        eagain_error = OSError(errno.EAGAIN, "Resource temporarily unavailable")69 70        # First call raises EAGAIN, selector times out on retry71        mock_sock.recv.side_effect = eagain_error72        mock_sock.gettimeout.return_value = 1.073 74        with patch("selectors.DefaultSelector") as mock_selector_class:75            mock_selector = Mock()76            mock_selector_class.return_value = mock_selector77            mock_selector.select.return_value = []  # Timeout - no data ready78 79            with self.assertRaises(WebSocketTimeoutException) as cm:80                recv(mock_sock, 100)81 82            # Verify correct timeout exception and message83            self.assertIn("Connection timed out waiting for data", str(cm.exception))84 85    def test_bug_wrong_exception_for_selector_timeout_fixed(self):86        """87        BUG #6 FIX VERIFICATION: Test that selector timeout now raises correct exception type88 89        Bug was in _socket.py:115 returning None for timeout, treated as connection error90        Fixed: Now raises WebSocketTimeoutException directly91        """92        mock_sock = Mock()93        mock_sock.recv.side_effect = SSLWantReadError()  # Trigger retry path94        mock_sock.gettimeout.return_value = 1.095 96        with patch("selectors.DefaultSelector") as mock_selector_class:97            mock_selector = Mock()98            mock_selector_class.return_value = mock_selector99            mock_selector.select.return_value = []  # TIMEOUT - this is key!100 101            with self.assertRaises(WebSocketTimeoutException) as cm:102                recv(mock_sock, 100)103 104            # Verify it's the correct timeout exception with proper message105            self.assertIn("Connection timed out waiting for data", str(cm.exception))106 107            # This proves the fix works:108            # 1. selector.select() returns [] (timeout)109            # 2. _recv() now raises WebSocketTimeoutException directly110            # 3. No more misclassification as connection closed error!111 112    def test_socket_timeout_exception_handling(self):113        """114        Test that socket.timeout exceptions are properly handled115        """116        mock_sock = Mock()117        mock_sock.gettimeout.return_value = 1.0118 119        # Simulate a real socket.timeout scenario120        mock_sock.recv.side_effect = socket.timeout("Operation timed out")121 122        # This works correctly - socket.timeout raises WebSocketTimeoutException123        with self.assertRaises(WebSocketTimeoutException) as cm:124            recv(mock_sock, 100)125 126        # In Python 3.10+, socket.timeout is a subclass of TimeoutError127        # so it's caught by the TimeoutError handler with hardcoded message128        # In Python 3.9, socket.timeout is caught by socket.timeout handler129        # which preserves the original message130        import sys131 132        if sys.version_info >= (3, 10):133            self.assertIn("Connection timed out", str(cm.exception))134        else:135            self.assertIn("Operation timed out", str(cm.exception))136 137    def test_correct_ssl_want_read_retry_behavior(self):138        """Test the correct behavior when SSLWantReadError is properly handled"""139        mock_sock = Mock()140 141        # First call raises SSLWantReadError, second call succeeds142        mock_sock.recv.side_effect = [SSLWantReadError(), b"data after retry"]143        mock_sock.gettimeout.return_value = 1.0144 145        with patch("selectors.DefaultSelector") as mock_selector_class:146            mock_selector = Mock()147            mock_selector_class.return_value = mock_selector148            mock_selector.select.return_value = [True]  # Data ready after wait149 150            # This should work correctly151            result = recv(mock_sock, 100)152            self.assertEqual(result, b"data after retry")153 154            # Selector should be used for retry155            mock_selector.register.assert_called()156            mock_selector.select.assert_called()157 158 159if __name__ == "__main__":160    unittest.main()161