test: add test cases for reconnections

Fixes #125.

Author: Totktonada

unit/suites/__init__.py

@@ -7,8 +7,10 @@
 from .test_schema import TestSuite_Schema
 from .test_dml import TestSuite_Request
 from .test_protocol import TestSuite_Protocol
+from .test_reconnect import TestSuite_Reconnect
 
-test_cases = (TestSuite_Schema, TestSuite_Request, TestSuite_Protocol)
+test_cases = (TestSuite_Schema, TestSuite_Request, TestSuite_Protocol,
+              TestSuite_Reconnect)
 
 def load_tests(loader, tests, pattern):
     suite = unittest.TestSuite()

unit/suites/test_reconnect.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+
+import unittest
+import warnings
+import tarantool
+from .lib.tarantool_server import TarantoolServer
+
+
+class TestSuite_Reconnect(unittest.TestCase):
+    @classmethod
+    def setUpClass(self):
+        print(' RECONNECT '.center(70, '='))
+        print('-' * 70)
+        self.srv = TarantoolServer()
+        self.srv.script = 'unit/suites/box.lua'
+
+    def test_01_simple(self):
+        # Create a connection, but don't connect it.
+        con = tarantool.Connection('localhost', self.srv.args['primary'],
+                                   connect_now=False)
+
+        # Trigger a reconnection due to server unavailability.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            with self.assertRaises(tarantool.error.NetworkError):
+                con.ping()
+
+        # Start a server and verify that the reconnection
+        # succeeds.
+        self.srv.start()
+        self.assertIs(con.ping(notime=True), "Success")
+
+        # Close the connection and stop the server.
+        con.close()
+        self.srv.stop()
+
+    def test_02_wrong_auth(self):
+        # Create a connection with wrong credentials, but don't
+        # connect it.
+        con = tarantool.Connection('localhost', self.srv.args['primary'],
+                                   connect_now=False, user='not_exist')
+
+        # Start a server.
+        self.srv.start()
+
+        # Trigger a reconnection due to wrong credentials.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            with self.assertRaises(tarantool.error.DatabaseError):
+                con.ping()
+
+        # Set right credentials and verify that the reconnection
+        # succeeds.
+        con.user = None
+        self.assertIs(con.ping(notime=True), "Success")
+
+        # Close the connection and stop the server.
+        con.close()
+        self.srv.stop()
+
+    @classmethod
+    def tearDownClass(self):
+        self.srv.clean()