Add some tests

Author: DifferentialOrange

tarantool/__init__.py

@@ -2,7 +2,7 @@
 # pylint: disable=C0301,W0105,W0401,W0614
 
 from tarantool.connection import Connection
-from tarantool.connection_pool import ConnectionPool
+from tarantool.connection_pool import ConnectionPool, Mode
 from tarantool.const import (RECONNECT_DELAY, RECONNECT_MAX_ATTEMPTS,
                              SOCKET_TIMEOUT)
 from tarantool.error import DatabaseError, Error, NetworkError, NetworkWarning
@@ -60,4 +60,4 @@ def connectmesh(addrs=({'host': 'localhost', 'port': 3301},), user=None,
 
 __all__ = ['connect', 'Connection', 'connectmesh', 'MeshConnection', 'Schema',
            'Error', 'DatabaseError', 'NetworkError', 'NetworkWarning',
-           'SchemaError', 'ConnectionPool', 'dbapi']
+           'SchemaError', 'ConnectionPool', 'Mode', 'dbapi']

test/suites/test_pool.py

@@ -0,0 +1,157 @@
+# -*- coding: utf-8 -*-
+
+from __future__ import print_function
+
+import sys
+import unittest
+import warnings
+from time import sleep
+
+import tarantool
+from tarantool.error import (ClusterDiscoveryWarning, ConfigurationError,
+                             NetworkError, NetworkWarning)
+
+from .lib.tarantool_server import TarantoolServer
+
+
+def create_server(_id):
+    srv = TarantoolServer()
+    srv.script = 'test/suites/box.lua'
+    srv.start()
+    srv.admin("box.schema.user.create('test', {password = 'test', " +
+              "if_not_exists = true})")
+    srv.admin("box.schema.user.grant('test', 'execute', 'universe')")
+
+    # Create srv_id function (for testing purposes).
+    srv.admin("function srv_id() return %s end" % _id)
+    return srv
+
+
+@unittest.skipIf(sys.platform.startswith("win"),
+                 'Pool tests on windows platform are not supported')
+class TestSuite_Pool(unittest.TestCase):
+    # def define_cluster_function(self, func_name, servers):
+    #     addresses = [(srv.host, srv.args['primary']) for srv in servers]
+    #     addresses_lua = ",".join("'%s:%d'" % address for address in addresses)
+    #     func_body = """
+    #         function %s()
+    #             return {%s}
+    #         end
+    #     """ % (func_name, addresses_lua)
+    #     for srv in self.servers:
+    #         srv.admin(func_body)
+
+    # def define_custom_cluster_function(self, func_name, retval):
+    #     func_body = """
+    #         function %s()
+    #             return %s
+    #         end
+    #     """ % (func_name, retval)
+    #     for srv in self.servers:
+    #         srv.admin(func_body)
+
+    def set_ro(self, srv, read_only):
+        if read_only:
+            req = r'box.cfg{read_only = true}'
+        else:
+            req = r'box.cfg{read_only = false}'
+
+        srv.admin(req)
+
+
+    def set_cluster_ro(self, read_only_list):
+        assert len(self.servers) == len(read_only_list)
+
+        for i in range(len(self.servers)):
+            self.set_ro(self.servers[i], read_only_list[i])
+
+    @classmethod
+    def setUpClass(self):
+        print(' POOL '.center(70, '='), file=sys.stderr)
+        print('-' * 70, file=sys.stderr)
+
+    def setUp(self):
+        # Create five servers and extract helpful fields for tests.
+        self.servers = []
+        self.addrs = []
+        for i in range(5):
+            srv = create_server(i)
+            self.servers.append(srv)
+            self.addrs.append({'host': srv.host, 'port': srv.args['primary']})
+
+    def test_00_basic(self):
+        self.set_cluster_ro([False, False, True, False, True])
+
+        conn = tarantool.ConnectionPool(addrs=self.addrs, user='test', password='test')
+
+        self.assertSequenceEqual(conn.eval('return box.cfg.read_only', mode=tarantool.Mode.RW), [False])
+        self.assertSequenceEqual(conn.eval('return box.cfg.read_only', mode=tarantool.Mode.PREFER_RW), [False])
+        self.assertSequenceEqual(conn.eval('return box.cfg.read_only', mode=tarantool.Mode.PREFER_RO), [True])
+
+    def test_01_roundrobin(self):
+        self.set_cluster_ro([False, False, True, False, True])
+        RW_ports = set([str(self.addrs[0]['port']), str(self.addrs[1]['port']), str(self.addrs[3]['port'])])
+        RO_ports = set([str(self.addrs[2]['port']), str(self.addrs[4]['port'])])
+
+        conn = tarantool.ConnectionPool(addrs=self.addrs, user='test', password='test')
+
+        # Expect RW iterate through all RW instances.
+        RW_ports_result = set()
+        for i in range(len(self.servers)):
+            resp = conn.eval('return box.cfg.listen', mode=tarantool.Mode.RW)
+            RW_ports_result.add(resp.data[0])
+
+        self.assertSetEqual(RW_ports_result, RW_ports)
+
+        # Expect PREFER_RW iterate through all RW instances if there is at least one.
+        PREFER_RW_ports_result = set()
+        for i in range(len(self.servers)):
+            resp = conn.eval('return box.cfg.listen', mode=tarantool.Mode.PREFER_RW)
+            PREFER_RW_ports_result.add(resp.data[0])
+
+        self.assertSetEqual(PREFER_RW_ports_result, RW_ports)
+
+        # Expect PREFER_RO iterate through all RO instances if there is at least one.
+        PREFER_RO_ports_result = set()
+        for i in range(len(self.servers)):
+            resp = conn.eval('return box.cfg.listen', mode=tarantool.Mode.PREFER_RO)
+            PREFER_RO_ports_result.add(resp.data[0])
+
+        self.assertSetEqual(PREFER_RO_ports_result, RO_ports)
+
+    def test_02_roundrobin_all_ro(self):
+        self.set_cluster_ro([True, True, True, True, True])
+        all_ports = set()
+        for addr in self.addrs:
+            all_ports.add(str(addr['port']))
+
+        conn = tarantool.ConnectionPool(addrs=self.addrs, user='test', password='test')
+
+        # Expect PREFER_RW iterate through all instances if there are no RW.
+        PREFER_RW_ports_result = set()
+        for i in range(len(self.servers)):
+            resp = conn.eval('return box.cfg.listen', mode=tarantool.Mode.PREFER_RW)
+            PREFER_RW_ports_result.add(resp.data[0])
+
+        self.assertSetEqual(PREFER_RW_ports_result, all_ports)
+
+    def test_03_roundrobin_all_rw(self):
+        self.set_cluster_ro([False, False, False, False, False])
+        all_ports = set()
+        for addr in self.addrs:
+            all_ports.add(str(addr['port']))
+
+        conn = tarantool.ConnectionPool(addrs=self.addrs, user='test', password='test')
+
+        # Expect PREFER_RO iterate through all instances if there are no RO.
+        PREFER_RO_ports_result = set()
+        for i in range(len(self.servers)):
+            resp = conn.eval('return box.cfg.listen', mode=tarantool.Mode.PREFER_RO)
+            PREFER_RO_ports_result.add(resp.data[0])
+
+        self.assertSetEqual(PREFER_RO_ports_result, all_ports)
+
+    def tearDown(self):
+        for srv in self.servers:
+            srv.stop()
+            srv.clean()