Splitted s3 tests

Author: GrigoriyPA

ydb/tests/fq/s3/conftest.py

@@ -27,6 +27,19 @@
 S3_PID_FILE = "s3.pid"
 
 
+class TestCounter:
+    def __init__(self, tests_count_limit, error_string):
+        self.tests_count_limit = tests_count_limit
+        self.error_string = error_string
+        self.number_tests = 0
+
+    def on_test_start(self):
+        self.number_tests += 1
+        assert self.number_tests <= self.tests_count_limit, \
+            f"{self.error_string} exceeded limit {self.number_tests} vs {self.tests_count_limit}, " \
+            "this may lead timeouts on CI, please split this class"
+
+
 @pytest.fixture(scope="module")
 def mvp_external_ydb_endpoint(request) -> str:
     return request.param["endpoint"] if request is not None and hasattr(request, 'param') else None
@@ -61,12 +74,12 @@ def is_s3_ready():
         recipes_common.stop_daemon(pid)
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="class")
 def kikimr_settings(request: pytest.FixtureRequest):
     return getattr(request, "param", dict())
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="class")
 def kikimr_params(request: pytest.FixtureRequest):
     return request
 
@@ -85,15 +98,22 @@ def get_kikimr_extensions(s3: S3, yq_version: str, kikimr_settings, mvp_external
     ]
 
 
-@pytest.fixture(scope="module")
-def kikimr_yqv1(kikimr_params: pytest.FixtureRequest, s3: S3, kikimr_settings, mvp_external_ydb_endpoint):
+@pytest.fixture(scope="class")
+def kikimr_starts_counter():
+    return TestCounter(10, "Number kikimr restarts in one class")
+
+
+@pytest.fixture(scope="class")
+def kikimr_yqv1(kikimr_params: pytest.FixtureRequest, s3: S3, kikimr_settings, mvp_external_ydb_endpoint, kikimr_starts_counter):
+    kikimr_starts_counter.on_test_start()
     kikimr_extensions = get_kikimr_extensions(s3, YQV1_VERSION_NAME, kikimr_settings, mvp_external_ydb_endpoint)
     with start_kikimr(kikimr_params, kikimr_extensions) as kikimr:
         yield kikimr
 
 
-@pytest.fixture(scope="module")
-def kikimr_yqv2(kikimr_params: pytest.FixtureRequest, s3: S3, kikimr_settings, mvp_external_ydb_endpoint):
+@pytest.fixture(scope="class")
+def kikimr_yqv2(kikimr_params: pytest.FixtureRequest, s3: S3, kikimr_settings, mvp_external_ydb_endpoint, kikimr_starts_counter):
+    kikimr_starts_counter.on_test_start()
     kikimr_extensions = get_kikimr_extensions(s3, YQV2_VERSION_NAME, kikimr_settings, mvp_external_ydb_endpoint)
     with start_kikimr(kikimr_params, kikimr_extensions) as kikimr:
         yield kikimr
@@ -115,8 +135,14 @@ def kikimr(yq_version: str, kikimr_yqv1, kikimr_yqv2):
     return kikimr
 
 
+@pytest.fixture(scope="class")
+def tests_counter():
+    return TestCounter(200, "Number tests in one class")
+
+
 @pytest.fixture
-def client(kikimr, request=None):
+def client(kikimr, tests_counter, request=None):
+    tests_counter.on_test_start()
     client = FederatedQueryClient(
         request.param["folder_id"] if request is not None else "my_folder", streaming_over_kikimr=kikimr
     )
@@ -128,8 +154,5 @@ def client(kikimr, request=None):
 
 @pytest.fixture
 def unique_prefix(request: pytest.FixtureRequest):
-    name_hash = hash(request.node.name)
-    if name_hash >= 0:
-        return f"p{name_hash}_"
-    else:
-        return f"n{-name_hash}_"
+    name_hash = abs(hash(request.node.name))
+    return f"h{name_hash}_{request.function.__name__}"

ydb/tests/fq/s3/test_bindings.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+import time
 import boto3
 import logging
 import pytest
@@ -17,7 +18,7 @@
 }
 
 
-class TestBindings:
+class TestBindingsChunk1:
     @staticmethod
     def _preprocess_query(sql: str, yq_version: str) -> str:
         if yq_version == 'v1':
@@ -40,7 +41,7 @@ def _assert_bindings(client: FederatedQueryClient, expected_binding_names: list[
     @staticmethod
     def _assert_query_results(client: FederatedQueryClient, sql: str, yq_version: str, expected_result_set):
         query_id = client.create_query(
-            "simple", TestBindings._preprocess_query(sql, yq_version), type=fq.QueryContent.QueryType.ANALYTICS
+            "simple", TestBindingsChunk1._preprocess_query(sql, yq_version), type=fq.QueryContent.QueryType.ANALYTICS
         ).result.query_id
         client.wait_query_status(query_id, fq.QueryMeta.COMPLETED)
 
@@ -248,6 +249,7 @@ def test_modify_connection_with_a_lot_of_bindings(
     @yq_all
     @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
     def test_name_uniqueness_constraint(self, kikimr, client: FederatedQueryClient, unique_prefix):
+        time.sleep(1000)
         # Test connection & binding creation
         kikimr.control_plane.wait_bootstrap(1)
         storage_connection_name = unique_prefix + "connection_name"
@@ -347,6 +349,8 @@ def test_name_uniqueness_constraint(self, kikimr, client: FederatedQueryClient,
         )
         assert modify_binding_result.issues[0].severity == 1
 
+
+class TestBindingsChunk2:
     @yq_all
     @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
     @pytest.mark.parametrize("kikimr_settings", [{"bindings_mode": "BM_DROP_WITH_WARNING"}], indirect=True)

ydb/tests/fq/s3/test_explicit_partitioning.py

@@ -13,7 +13,7 @@
 from ydb.tests.tools.fq_runner.kikimr_utils import yq_all
 
 
-class TestS3(TestYdsBase):
+class TestS3Chunk1(TestYdsBase):
     @yq_all
     @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
     @pytest.mark.parametrize("runtime_listing", ["false", "true"])
@@ -866,6 +866,8 @@ def test_projection_enum_type_invalid_validation(
                 describe_result.query.issue
             )
 
+
+class TestS3Chunk2(TestYdsBase):
     @yq_all
     @pytest.mark.parametrize(
         "client, column_type, is_correct",

ydb/tests/fq/s3/test_s3.py

@@ -13,7 +13,7 @@
 from ydb.tests.tools.fq_runner.kikimr_utils import yq_v1, yq_v2, yq_all
 
 
-class TestS3(TestYdsBase):
+class TestS3Chunk1(TestYdsBase):
     @yq_all
     @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
     @pytest.mark.parametrize("runtime_listing", ["false", "true"])
@@ -455,6 +455,8 @@ def wait_checkpoints(require_query_is_on=False):
         client.abort_query(query_id)
         client.wait_query(query_id)
 
+
+class TestS3Chunk2(TestYdsBase):
     @yq_v1  # v2 compute with multiple nodes is not supported yet
     @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
     @pytest.mark.parametrize("kikimr_params", [{"compute": 3}], indirect=True)

ydb/tests/fq/s3/ya.make

@@ -1,6 +1,7 @@
 PY3TEST()
 
 FORK_TEST_FILES()
+FORK_TESTS()
 
 INCLUDE(${ARCADIA_ROOT}/ydb/tests/tools/fq_runner/ydb_runner_with_datastreams.inc)
 

ydb/tests/library/harness/kikimr_runner.py

@@ -421,10 +421,13 @@ def unregister_and_stop_slots(self, slots):
         for i in slots:
             i.stop()
 
-    def __stop_node(self, node):
+    def __stop_node(self, node, kill=False):
         ret = None
         try:
-            node.stop()
+            if kill:
+                node.kill()
+            else:
+                node.stop()
         except daemon.DaemonError as exceptions:
             ret = exceptions
         else:
@@ -434,16 +437,16 @@ def __stop_node(self, node):
                 shutil.rmtree(self.__common_udfs_dir, ignore_errors=True)
         return ret
 
-    def stop(self):
+    def stop(self, kill=False):
         saved_exceptions = []
 
         for slot in self.slots.values():
-            exception = self.__stop_node(slot)
+            exception = self.__stop_node(slot, kill)
             if exception is not None:
                 saved_exceptions.append(exception)
 
         for node in self.nodes.values():
-            exception = self.__stop_node(node)
+            exception = self.__stop_node(node, kill)
             if exception is not None:
                 saved_exceptions.append(exception)
 

ydb/tests/tools/fq_runner/kikimr_runner.py

@@ -52,7 +52,7 @@ def start(self):
 
     def stop(self):
         if self.kikimr_cluster:
-            self.kikimr_cluster.stop()
+            self.kikimr_cluster.stop(kill=True)
 
     def endpoint(self, node_index=None):
         return "localhost:{}".format(