Merge pull request datastax#989 from datastax/long-python_catch-ssl-errs

PYTHON-1024: fix catch ssl err catching

Author: mambocab

CHANGELOG.rst

@@ -1,6 +1,10 @@
 3.15.2
 ======
 
+Bug Fixes
+---------
+* Improve and fix socket error-catching code in nonblocking-socket reactors (PYTHON-1024)
+
 Other
 -----
 * Fix tests when RF is not maintained if we decomission a node (PYTHON-1017)

cassandra/io/asyncorereactor.py

@@ -424,10 +424,14 @@ def handle_read(self):
                     break
         except socket.error as err:
             if ssl and isinstance(err, ssl.SSLError):
-                if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                    return
+                else:
                     self.defunct(err)
                     return
-            elif err.args[0] not in NONBLOCKING:
+            elif err.args[0] in NONBLOCKING:
+                return
+            else:
                 self.defunct(err)
                 return
 

cassandra/io/libevreactor.py

@@ -344,10 +344,14 @@ def handle_read(self, watcher, revents, errno=None):
                     break
         except socket.error as err:
             if ssl and isinstance(err, ssl.SSLError):
-                if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                    return
+                else:
                     self.defunct(err)
                     return
-            elif err.args[0] not in NONBLOCKING:
+            elif err.args[0] in NONBLOCKING:
+                return
+            else:
                 self.defunct(err)
                 return
 

tests/unit/io/utils.py

@@ -25,9 +25,11 @@
 from mock import Mock
 
 import errno
+import logging
 import math
 import os
 from socket import error as socket_error
+import ssl
 
 try:
     import unittest2 as unittest
@@ -37,6 +39,9 @@
 import time
 
 
+log = logging.getLogger(__name__)
+
+
 class TimerCallback(object):
 
     invoked = False
@@ -247,26 +252,38 @@ def test_successful_connection(self):
         return c
 
     def test_eagain_on_buffer_size(self):
+        self._check_error_recovery_on_buffer_size(errno.EAGAIN)
+
+    def test_ewouldblock_on_buffer_size(self):
+        self._check_error_recovery_on_buffer_size(errno.EWOULDBLOCK)
+
+    def test_sslwantread_on_buffer_size(self):
+        self._check_error_recovery_on_buffer_size(ssl.SSL_ERROR_WANT_READ)
+
+    def test_sslwantwrite_on_buffer_size(self):
+        self._check_error_recovery_on_buffer_size(ssl.SSL_ERROR_WANT_WRITE)
+
+    def _check_error_recovery_on_buffer_size(self, error_code):
         c = self.test_successful_connection()
 
         header = six.b('\x00\x00\x00\x00') + int32_pack(20000)
         responses = [
             header + (six.b('a') * (4096 - len(header))),
             six.b('a') * 4096,
-            socket_error(errno.EAGAIN),
+            socket_error(error_code),
             six.b('a') * 100,
-            socket_error(errno.EAGAIN)]
+            socket_error(error_code)]
 
         def side_effect(*args):
             response = responses.pop(0)
+            log.debug('about to mock return {}'.format(response))
             if isinstance(response, socket_error):
                 raise response
             else:
                 return response
 
         self.get_socket(c).recv.side_effect = side_effect
         c.handle_read(*self.null_handle_function_args)
-        self.assertEqual(c._current_frame.end_pos, 20000 + len(header))
         # the EAGAIN prevents it from reading the last 100 bytes
         c._iobuf.seek(0, os.SEEK_END)
         pos = c._iobuf.tell()