IGNITE-14240 Re-factor tests (#29)

Handle authentication error.
Fix infinite recursion on failed connection on handshake.
Skip affinity test if server doesn't support protocol.
Remove travis.

(cherry picked from commit b2030be)

Co-authored-by: Ivan Dashchinskiy <ivandasch@gmail.com>

Author: isapego

pygridgain/connection/connection.py

@@ -170,7 +170,7 @@ def read_response(self) -> Union[dict, OrderedDict]:
             ('length', Int),
             ('op_code', Byte),
         ])
-        with BinaryStream(self, self.recv()) as stream:
+        with BinaryStream(self, self.recv(reconnect=False)) as stream:
             start_class = response_start.parse(stream)
             start = stream.read_ctype(start_class, direction=READ_BACKWARD)
             data = response_start.to_python(start)
@@ -266,7 +266,7 @@ def _connect_version(
 
         with BinaryStream(self) as stream:
             hs_request.from_python(stream)
-            self.send(stream.getbuffer())
+            self.send(stream.getbuffer(), reconnect=False)
 
         hs_response = self.read_response()
         if hs_response['op_code'] == 0:
@@ -314,12 +314,13 @@ def reconnect(self):
         except connection_errors:
             pass
 
-    def send(self, data: Union[bytes, bytearray, memoryview], flags=None):
+    def send(self, data: Union[bytes, bytearray, memoryview], flags=None, reconnect=True):
         """
         Send data down the socket.
 
         :param data: bytes to send,
         :param flags: (optional) OS-specific flags.
+        :param reconnect: (optional) reconnect on failure, default True.
         """
         if self.closed:
             raise SocketError('Attempt to use closed connection.')
@@ -335,7 +336,13 @@ def send(self, data: Union[bytes, bytearray, memoryview], flags=None):
             self.reconnect()
             raise
 
-    def recv(self, flags=None) -> bytearray:
+    def recv(self, flags=None, reconnect=True) -> bytearray:
+        """
+        Receive data from the socket.
+
+        :param flags: (optional) OS-specific flags.
+        :param reconnect: (optional) reconnect on failure, default True.
+        """
         def _recv(buffer, num_bytes):
             bytes_to_receive = num_bytes
             while bytes_to_receive > 0:
@@ -345,7 +352,8 @@ def _recv(buffer, num_bytes):
                         raise SocketError('Connection broken.')
                 except connection_errors:
                     self.failed = True
-                    self.reconnect()
+                    if reconnect:
+                        self.reconnect()
                     raise
 
                 buffer = buffer[bytes_rcvd:]

pygridgain/constants.py

@@ -52,7 +52,7 @@
 PROTOCOL_STRING_ENCODING = 'utf-8'
 PROTOCOL_CHAR_ENCODING = 'utf-16le'
 
-SSL_DEFAULT_VERSION = ssl.PROTOCOL_TLSv1_1
+SSL_DEFAULT_VERSION = ssl.PROTOCOL_TLSv1_2
 SSL_DEFAULT_CIPHERS = ssl._DEFAULT_CIPHERS
 
 FNV1_OFFSET_BASIS = 0x811c9dc5

requirements/install.txt

@@ -1,4 +1,4 @@
 # these pip packages are necessary for the pygridgain to run
 
-attrs==18.1.0
+attrs==20.3.0
 tzlocal==2.1

requirements/setup.txt

@@ -1,3 +1,3 @@
 # additional package for integrating pytest in setuptools
 
-pytest-runner==4.2
+pytest-runner==5.3.0

requirements/tests.txt

@@ -1,7 +1,7 @@
 # these packages are used for testing
 
-pytest==3.6.1
-pytest-cov==2.5.1
-teamcity-messages==1.21
-psutil==5.6.5
+pytest==6.2.2
+pytest-cov==2.11.1
+teamcity-messages==1.28
+psutil==5.8.0
 jinja2==2.11.3

tests/affinity/conftest.py

@@ -0,0 +1,72 @@
+#
+# Copyright 2021 GridGain Systems, Inc. and Contributors.
+#
+# Licensed under the GridGain Community Edition License (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+#
+# 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.
+#
+import pytest
+
+from pygridgain import Client
+from pygridgain.api import cache_create, cache_destroy
+from tests.util import start_ignite_gen
+
+
+@pytest.fixture(scope='module', autouse=True)
+def server1():
+    yield from start_ignite_gen(1)
+
+
+@pytest.fixture(scope='module', autouse=True)
+def server2():
+    yield from start_ignite_gen(2)
+
+
+@pytest.fixture(scope='module', autouse=True)
+def server3():
+    yield from start_ignite_gen(3)
+
+
+@pytest.fixture
+def client():
+    client = Client(partition_aware=True)
+
+    client.connect([('127.0.0.1', 10800 + i) for i in range(1, 4)])
+
+    yield client
+
+    client.close()
+
+
+@pytest.fixture
+def client_not_connected():
+    client = Client(partition_aware=True)
+    yield client
+    client.close()
+
+
+@pytest.fixture
+def cache(connected_client):
+    cache_name = 'my_bucket'
+    conn = connected_client.random_node
+
+    cache_create(conn, cache_name)
+    yield cache_name
+    cache_destroy(conn, cache_name)
+
+
+@pytest.fixture(scope='module', autouse=True)
+def skip_if_no_affinity(request, server1):
+    client = Client(partition_aware=True)
+    client.connect('127.0.0.1', 10801)
+
+    if not client.partition_awareness_supported_by_protocol:
+        pytest.skip(f'skipped {request.node.name}, partition awareness is not supported.')

tests/test_affinity.py tests/affinity/test_affinity.pytests/test_affinity.py renamed to tests/affinity/test_affinity.py

@@ -27,12 +27,11 @@
 from pygridgain.datatypes.prop_codes import *
 
 
-def test_get_node_partitions(client_partition_aware):
+def test_get_node_partitions(client):
+    conn = client.random_node
 
-    conn = client_partition_aware.random_node
-
-    cache_1 = client_partition_aware.get_or_create_cache('test_cache_1')
-    cache_2 = client_partition_aware.get_or_create_cache({
+    cache_1 = client.get_or_create_cache('test_cache_1')
+    cache_2 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_2',
         PROP_CACHE_KEY_CONFIGURATION: [
             {
@@ -41,9 +40,9 @@ def test_get_node_partitions(client_partition_aware):
             }
         ],
     })
-    cache_3 = client_partition_aware.get_or_create_cache('test_cache_3')
-    cache_4 = client_partition_aware.get_or_create_cache('test_cache_4')
-    cache_5 = client_partition_aware.get_or_create_cache('test_cache_5')
+    client.get_or_create_cache('test_cache_3')
+    client.get_or_create_cache('test_cache_4')
+    client.get_or_create_cache('test_cache_5')
 
     result = cache_get_node_partitions(
         conn,
@@ -115,9 +114,8 @@ def test_get_node_partitions(client_partition_aware):
 
     ],
 )
-def test_affinity(client_partition_aware, key, key_hint):
-
-    cache_1 = client_partition_aware.get_or_create_cache({
+def test_affinity(client, key, key_hint):
+    cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
         PROP_CACHE_MODE: CacheMode.PARTITIONED,
     })
@@ -126,7 +124,7 @@ def test_affinity(client_partition_aware, key, key_hint):
 
     best_node = cache_1.get_best_node(key, key_hint=key_hint)
 
-    for node in filter(lambda n: n.alive, client_partition_aware._nodes):
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key, key_hint=key_hint,
         )
@@ -142,9 +140,8 @@ def test_affinity(client_partition_aware, key, key_hint):
     cache_1.destroy()
 
 
-def test_affinity_for_generic_object(client_partition_aware):
-
-    cache_1 = client_partition_aware.get_or_create_cache({
+def test_affinity_for_generic_object(client):
+    cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
         PROP_CACHE_MODE: CacheMode.PARTITIONED,
     })
@@ -166,7 +163,7 @@ class KeyClass(
 
     best_node = cache_1.get_best_node(key, key_hint=BinaryObject)
 
-    for node in filter(lambda n: n.alive, client_partition_aware._nodes):
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key, key_hint=BinaryObject,
         )
@@ -182,16 +179,8 @@ class KeyClass(
     cache_1.destroy()
 
 
-def test_affinity_for_generic_object_without_type_hints(client_partition_aware):
-
-    if not client_partition_aware.partition_awareness_supported_by_protocol:
-        pytest.skip(
-            'Best effort affinity is not supported by the protocol {}.'.format(
-                client_partition_aware.protocol_version
-            )
-        )
-
-    cache_1 = client_partition_aware.get_or_create_cache({
+def test_affinity_for_generic_object_without_type_hints(client):
+    cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
         PROP_CACHE_MODE: CacheMode.PARTITIONED,
     })
@@ -213,7 +202,7 @@ class KeyClass(
 
     best_node = cache_1.get_best_node(key)
 
-    for node in filter(lambda n: n.alive, client_partition_aware._nodes):
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key
         )

tests/test_affinity_bad_servers.py …ts/affinity/test_affinity_bad_servers.pytests/test_affinity_bad_servers.py renamed to tests/affinity/test_affinity_bad_servers.py tests/test_affinity_bad_servers.py renamed to tests/affinity/test_affinity_bad_servers.py

@@ -16,22 +16,20 @@
 import pytest
 
 from pygridgain.exceptions import ReconnectError
-from tests.util import *
+from tests.util import start_ignite, kill_process_tree
 
 
-def test_client_with_multiple_bad_servers(start_client):
-    client = start_client(partition_aware=True)
+def test_client_with_multiple_bad_servers(client_not_connected):
     with pytest.raises(ReconnectError) as e_info:
-        client.connect([("127.0.0.1", 10900), ("127.0.0.1", 10901)])
+        client_not_connected.connect([("127.0.0.1", 10900), ("127.0.0.1", 10901)])
     assert str(e_info.value) == "Can not connect."
 
 
-def test_client_with_failed_server(request, start_ignite_server, start_client):
-    srv = start_ignite_server(4)
+def test_client_with_failed_server(request, client_not_connected):
+    srv = start_ignite(idx=4)
     try:
-        client = start_client()
-        client.connect([("127.0.0.1", 10804)])
-        cache = client.get_or_create_cache(request.node.name)
+        client_not_connected.connect([("127.0.0.1", 10804)])
+        cache = client_not_connected.get_or_create_cache(request.node.name)
         cache.put(1, 1)
         kill_process_tree(srv.pid)
         with pytest.raises(ConnectionResetError):
@@ -40,17 +38,16 @@ def test_client_with_failed_server(request, start_ignite_server, start_client):
         kill_process_tree(srv.pid)
 
 
-def test_client_with_recovered_server(request, start_ignite_server, start_client):
-    srv = start_ignite_server(4)
+def test_client_with_recovered_server(request, client_not_connected):
+    srv = start_ignite(idx=4)
     try:
-        client = start_client()
-        client.connect([("127.0.0.1", 10804)])
-        cache = client.get_or_create_cache(request.node.name)
+        client_not_connected.connect([("127.0.0.1", 10804)])
+        cache = client_not_connected.get_or_create_cache(request.node.name)
         cache.put(1, 1)
 
         # Kill and restart server
         kill_process_tree(srv.pid)
-        srv = start_ignite_server(4)
+        srv = start_ignite(idx=4)
 
         # First request fails
         with pytest.raises(Exception):

tests/test_affinity_request_routing.py …ffinity/test_affinity_request_routing.pytests/test_affinity_request_routing.py renamed to tests/affinity/test_affinity_request_routing.py tests/test_affinity_request_routing.py renamed to tests/affinity/test_affinity_request_routing.py

@@ -70,10 +70,8 @@ def check_grid_idx():
 
 @pytest.mark.parametrize("key,grid_idx", [(1, 1), (2, 2), (3, 3), (4, 1), (5, 1), (6, 2), (11, 1), (13, 1), (19, 1)])
 @pytest.mark.parametrize("backups", [0, 1, 2, 3])
-def test_cache_operation_on_primitive_key_routes_request_to_primary_node(
-        request, key, grid_idx, backups, client_partition_aware):
-
-    cache = client_partition_aware.get_or_create_cache({
+def test_cache_operation_on_primitive_key_routes_request_to_primary_node(request, key, grid_idx, backups, client):
+    cache = client.get_or_create_cache({
         PROP_NAME: request.node.name + str(backups),
         PROP_BACKUPS_NUMBER: backups,
     })
@@ -132,8 +130,7 @@ def test_cache_operation_on_complex_key_routes_request_to_primary_node():
 
 @pytest.mark.parametrize("key,grid_idx", [(1, 2), (2, 1), (3, 1), (4, 2), (5, 2), (6, 3)])
 @pytest.mark.skip(reason="GG-25823: Custom key objects are not supported yet")
-def test_cache_operation_on_custom_affinity_key_routes_request_to_primary_node(
-        request, client_partition_aware, key, grid_idx):
+def test_cache_operation_on_custom_affinity_key_routes_request_to_primary_node(request, client, key, grid_idx):
     class AffinityTestType1(
         metaclass=GenericObjectMeta,
         type_name='AffinityTestType1',
@@ -153,7 +150,7 @@ class AffinityTestType1(
             },
         ],
     }
-    cache = client_partition_aware.create_cache(cache_config)
+    cache = client.create_cache(cache_config)
 
     # noinspection PyArgumentList
     key_obj = AffinityTestType1(
@@ -167,17 +164,18 @@ class AffinityTestType1(
     assert requests.pop() == grid_idx
 
 
-def test_cache_operation_routed_to_new_cluster_node(request, start_ignite_server, start_client):
-    client = start_client(partition_aware=True)
-    client.connect([("127.0.0.1", 10801), ("127.0.0.1", 10802), ("127.0.0.1", 10803), ("127.0.0.1", 10804)])
-    cache = client.get_or_create_cache(request.node.name)
+def test_cache_operation_routed_to_new_cluster_node(request, client_not_connected):
+    client_not_connected.connect(
+        [("127.0.0.1", 10801), ("127.0.0.1", 10802), ("127.0.0.1", 10803), ("127.0.0.1", 10804)]
+    )
+    cache = client_not_connected.get_or_create_cache(request.node.name)
     key = 12
     wait_for_affinity_distribution(cache, key, 3)
     cache.put(key, key)
     cache.put(key, key)
     assert requests.pop() == 3
 
-    srv = start_ignite_server(4)
+    srv = start_ignite(idx=4)
     try:
         # Wait for rebalance and partition map exchange
         wait_for_affinity_distribution(cache, key, 4)
@@ -190,8 +188,8 @@ def test_cache_operation_routed_to_new_cluster_node(request, start_ignite_server
         kill_process_tree(srv.pid)
 
 
-def test_replicated_cache_operation_routed_to_random_node(request, client_partition_aware):
-    cache = client_partition_aware.get_or_create_cache({
+def test_replicated_cache_operation_routed_to_random_node(request, client):
+    cache = client.get_or_create_cache({
         PROP_NAME: request.node.name,
         PROP_CACHE_MODE: CacheMode.REPLICATED,
     })