Refactor harness 2 (#11220)

Author: maximyurchuk

ydb/public/tools/lib/cmds/__init__.py

@@ -41,6 +41,15 @@ def __init__(self):
         self.enabled_grpc_services = []
 
 
+def _get_build_path(path):
+    try:
+        result = yatest.common.build_path(path)
+    except (AttributeError, yatest.common.NoRuntimeFormed):
+        result = path
+
+    return result
+
+
 def ensure_path_exists(path):
     if not os.path.isdir(path):
         os.makedirs(path)
@@ -358,9 +367,8 @@ def deploy(arguments):
         domain_name='local',
         pq_client_service_types=pq_client_service_types(arguments),
         enable_pqcd=enable_pqcd(arguments),
-        load_udfs=True,
         suppress_version_check=arguments.suppress_version_check,
-        udfs_path=arguments.ydb_udfs_dir,
+        udfs_path=arguments.ydb_udfs_dir or _get_build_path("yql/udfs"),
         additional_log_configs=additional_log_configs,
         port_allocator=port_allocator,
         use_in_memory_pdisks=use_in_memory_pdisks_flag(arguments.ydb_working_dir),

ydb/tests/functional/api/test_crud.py

@@ -5,6 +5,8 @@
 
 from hamcrest import assert_that, equal_to, raises
 
+import yatest
+
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
 from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
 from ydb.tests.oss.ydb_sdk_import import ydb
@@ -73,7 +75,7 @@ def test_create_and_upsert_data_with_repetitions(self, repetitions, column_count
 class TestCRUDOperations(object):
     @classmethod
     def setup_class(cls):
-        cls.cluster = KiKiMR(KikimrConfigGenerator(load_udfs=True))
+        cls.cluster = KiKiMR(KikimrConfigGenerator(udfs_path=yatest.common.build_path("yql/udfs")))
         cls.cluster.start()
         cls.driver = ydb.Driver(
             ydb.DriverConfig(
@@ -134,7 +136,7 @@ def callee():
 class TestSelect(object):
     @classmethod
     def setup_class(cls):
-        cls.cluster = KiKiMR(KikimrConfigGenerator(load_udfs=True))
+        cls.cluster = KiKiMR(KikimrConfigGenerator(udfs_path=yatest.common.build_path("yql/udfs")))
         cls.cluster.start()
         cls.driver = ydb.Driver(
             ydb.DriverConfig(

ydb/tests/functional/canonical/test_sql.py

@@ -111,7 +111,7 @@ def setup_class(cls):
         cls.database = '/local'
         cls.cluster = KiKiMR(
             KikimrConfigGenerator(
-                load_udfs=True,
+                udfs_path=yatest.common.build_path("yql/udfs"),
                 domain_name='local',
                 use_in_memory_pdisks=True,
                 disable_iterator_reads=True,

ydb/tests/functional/suite_tests/test_base.py

@@ -245,7 +245,7 @@ class BaseSuiteRunner(object):
     def setup_class(cls):
         cls.cluster = KiKiMR(
             KikimrConfigGenerator(
-                load_udfs=True,
+                udfs_path=yatest.common.build_path("yql/udfs"),
                 use_in_memory_pdisks=True,
                 disable_iterator_reads=True,
                 disable_iterator_lookups=True,

ydb/tests/library/harness/daemon.py

@@ -6,7 +6,6 @@
 import sys
 import subprocess
 
-import yatest
 from yatest.common import process
 import six
 
@@ -53,23 +52,15 @@ def __init__(self, exceptions):
         super(SeveralDaemonErrors, self).__init__("\n".join(str(x) for x in exceptions))
 
 
-def _work_path(name):
-    # TODO: remove yatest dependency from harness
-    try:
-        return yatest.common.work_path(name)
-    except (AttributeError, yatest.common.NoRuntimeFormed):
-        return name
-
-
 class Daemon(object):
     """Local process executed as process in current host"""
     def __init__(
         self,
         command,
         cwd,
         timeout,
-        stdout_file=_work_path('stdout'),
-        stderr_file=_work_path('stderr'),
+        stdout_file,
+        stderr_file,
         stderr_on_error_lines=0,
         core_pattern=None,
     ):

ydb/tests/library/harness/kikimr_config.py

@@ -89,13 +89,9 @@ def _load_default_yaml(default_tablet_node_ids, ydb_domain_name, static_erasure,
     return yaml_dict
 
 
-def _read_file(filename):
-    with open(filename, "r") as f:
-        return f.read()
-
-
 def _load_yaml_config(filename):
-    return yaml.safe_load(_read_file(filename))
+    with open(filename, "r") as f:
+        return yaml.safe_load(f)
 
 
 def _use_in_memory_pdisks_var(pdisk_store_path, use_in_memory_pdisks):
@@ -108,16 +104,6 @@ def _use_in_memory_pdisks_var(pdisk_store_path, use_in_memory_pdisks):
     return use_in_memory_pdisks
 
 
-def _get_build_path(path):
-    # TODO: remove yatest dependency from harness
-    try:
-        result = yatest.common.build_path(path)
-    except (AttributeError, yatest.common.NoRuntimeFormed):
-        result = path
-
-    return result
-
-
 class KikimrConfigGenerator(object):
     def __init__(
             self,
@@ -126,7 +112,6 @@ def __init__(
             nodes=None,
             additional_log_configs=None,
             port_allocator=None,
-            load_udfs=False,
             udfs_path=None,
             output_path=None,
             enable_pq=True,
@@ -217,7 +202,6 @@ def __init__(
         self.static_erasure = erasure
         self.domain_name = domain_name
         self.__number_of_pdisks_per_node = 1 + len(dynamic_pdisks)
-        self.__load_udfs = load_udfs
         self.__udfs_path = udfs_path
         self._dcs = [1]
         if erasure == Erasure.MIRROR_3_DC:
@@ -236,7 +220,11 @@ def __init__(
 
         self.__dynamic_pdisks = dynamic_pdisks
 
-        self.__output_path = output_path or yatest.common.output_path()
+        self.__working_dir = output_path or yatest.common.test_output_path()
+
+        if not os.path.isdir(self.__working_dir):
+            os.makedirs(self.__working_dir)
+
         self.node_kind = node_kind
         self.yq_tenant = yq_tenant
         self.dc_mapping = dc_mapping
@@ -296,8 +284,10 @@ def __init__(
         # NOTE(shmel1k@): change to 'true' after migration to YDS scheme
         self.yaml_config['sqs_config']['enable_sqs'] = enable_sqs
         self.yaml_config['pqcluster_discovery_config']['enabled'] = enable_pqcd
-        self.yaml_config["net_classifier_config"]["net_data_file_path"] = os.path.join(self.__output_path,
-                                                                                       'netData.tsv')
+        self.yaml_config["net_classifier_config"]["net_data_file_path"] = os.path.join(
+            self.__working_dir,
+            'netData.tsv',
+        )
         with open(self.yaml_config["net_classifier_config"]["net_data_file_path"], "w") as net_data_file:
             net_data_file.write("")
 
@@ -482,37 +472,13 @@ def names_txt(self):
         return self.naming_config.NameserviceConfig
 
     def __set_enable_metering(self):
-        def ensure_path_exists(path):
-            if not os.path.isdir(path):
-                os.makedirs(path)
-            return path
-
-        def get_cwd_for_test(output_path):
-            test_name = yatest.common.context.test_name or ""
-            test_name = test_name.replace(':', '_')
-            return os.path.join(output_path, test_name)
-
-        cwd = get_cwd_for_test(self.__output_path)
-        ensure_path_exists(cwd)
-        metering_file_path = os.path.join(cwd, 'metering.txt')
+        metering_file_path = os.path.join(self.__working_dir, 'metering.txt')
         with open(metering_file_path, "w") as metering_file:
             metering_file.write('')
         self.yaml_config['metering_config'] = {'metering_file_path': metering_file_path}
 
     def __set_enable_audit_log(self):
-        def ensure_path_exists(path):
-            if not os.path.isdir(path):
-                os.makedirs(path)
-            return path
-
-        def get_cwd_for_test(output_path):
-            test_name = yatest.common.context.test_name or ""
-            test_name = test_name.replace(':', '_')
-            return os.path.join(output_path, test_name)
-
-        cwd = get_cwd_for_test(self.__output_path)
-        ensure_path_exists(cwd)
-        audit_file_path = os.path.join(cwd, 'audit.txt')
+        audit_file_path = os.path.join(self.__working_dir, 'audit.txt')
         with open(audit_file_path, "w") as audit_file:
             audit_file.write('')
         self.yaml_config['audit_config'] = dict(
@@ -534,8 +500,8 @@ def sqs_service_enabled(self):
         return self.yaml_config['sqs_config']['enable_sqs']
 
     @property
-    def output_path(self):
-        return self.__output_path
+    def working_dir(self):
+        return self.__working_dir
 
     def get_binary_path(self, node_id):
         binary_paths = self.__binary_paths
@@ -573,9 +539,9 @@ def clone_grpc_as_ext_endpoint(self, port, endpoint_id=None):
         self.yaml_config['grpc_config']['ext_endpoints'].append(cur_grpc_config)
 
     def get_yql_udfs_to_load(self):
-        if not self.__load_udfs:
+        if self.__udfs_path is None:
             return []
-        udfs_path = self.__udfs_path or _get_build_path("yql/udfs")
+        udfs_path = self.__udfs_path
         result = []
         for dirpath, dnames, fnames in os.walk(udfs_path):
             is_loaded = False

ydb/tests/library/harness/kikimr_runner.py

@@ -24,17 +24,6 @@
 logger = logging.getLogger(__name__)
 
 
-def get_unique_path_for_current_test(output_path, sub_folder):
-    # TODO: remove yatest dependency from harness
-    try:
-        test_name = yatest.common.context.test_name
-    except (AttributeError, yatest.common.NoRuntimeFormed):
-        test_name = ""
-    test_name = test_name.replace(':', '_')
-
-    return os.path.join(output_path, test_name, sub_folder)
-
-
 def ensure_path_exists(path):
     # NOTE: can't switch to os.makedirs(path, exist_ok=True) as some tests
     # are still running under python2 (exist_ok was added in py3.2)
@@ -43,14 +32,6 @@ def ensure_path_exists(path):
     return path
 
 
-def join(a, b):
-    if a is None:
-        a = ''
-    if b is None:
-        b = ''
-    return os.path.join(a, b)
-
-
 class KiKiMRNode(daemon.Daemon, kikimr_node_interface.NodeInterface):
     def __init__(self, node_id, config_path, port_allocator, cluster_name, configurator,
                  udfs_dir=None, role='node', node_broker_port=None, tenant_affiliation=None, encryption_key=None,
@@ -59,7 +40,6 @@ def __init__(self, node_id, config_path, port_allocator, cluster_name, configura
         super(kikimr_node_interface.NodeInterface, self).__init__()
         self.node_id = node_id
         self.data_center = data_center
-        self.__cwd = None
         self.__config_path = config_path
         self.__cluster_name = cluster_name
         self.__configurator = configurator
@@ -80,33 +60,35 @@ def __init__(self, node_id, config_path, port_allocator, cluster_name, configura
         self.__role = role
         self.__node_broker_port = node_broker_port
 
+        self.__working_dir = ensure_path_exists(
+            os.path.join(
+                self.__configurator.working_dir,
+                self.__cluster_name,
+                "{}_{}".format(
+                    self.__role,
+                    self.node_id
+                )
+            )
+        )
+
         if configurator.use_log_files:
-            self.__log_file = tempfile.NamedTemporaryFile(dir=self.cwd, prefix="logfile_", suffix=".log", delete=False)
-            kwargs = {}
+            self.__log_file = tempfile.NamedTemporaryFile(dir=self.__working_dir, prefix="logfile_", suffix=".log", delete=False)
+            kwargs = {
+                "stdout_file": os.path.join(self.__working_dir, "stdout"),
+                "stderr_file": os.path.join(self.__working_dir, "stderr")
+                }
         else:
             self.__log_file = None
             kwargs = {
                 "stdout_file": "/dev/stdout",
                 "stderr_file": "/dev/stderr"
                 }
 
-        daemon.Daemon.__init__(self, self.command, cwd=self.cwd, timeout=180, stderr_on_error_lines=240, **kwargs)
+        daemon.Daemon.__init__(self, self.command, cwd=self.__working_dir, timeout=180, stderr_on_error_lines=240, **kwargs)
 
     @property
     def cwd(self):
-        if self.__cwd is None:
-            self.__cwd = ensure_path_exists(
-                get_unique_path_for_current_test(
-                    self.__configurator.output_path,
-                    join(
-                        self.__cluster_name, "{}_{}".format(
-                            self.__role,
-                            self.node_id
-                        )
-                    )
-                )
-            )
-        return self.__cwd
+        return self.__working_dir
 
     @property
     def binary_path(self):
@@ -171,7 +153,7 @@ def __make_run_command(self):
 
         command.extend(
             [
-                "--yaml-config=%s" % join(self.__config_path, "config.yaml"),
+                "--yaml-config=%s" % os.path.join(self.__config_path, "config.yaml"),
                 "--grpc-port=%s" % self.grpc_port,
                 "--mon-port=%d" % self.mon_port,
                 "--ic-port=%d" % self.ic_port,
@@ -232,7 +214,7 @@ def start(self):
 
 
 class KiKiMR(kikimr_cluster_interface.KiKiMRClusterInterface):
-    def __init__(self, configurator=None, cluster_name=''):
+    def __init__(self, configurator=None, cluster_name='cluster'):
         super(KiKiMR, self).__init__()
 
         self.__tmpdir = tempfile.mkdtemp(prefix="kikimr_" + cluster_name + "_")
@@ -243,11 +225,10 @@ def __init__(self, configurator=None, cluster_name=''):
         self._nodes = {}
         self._slots = {}
         self.__server = 'localhost'
-        self.__client = None
-        self.__kv_client = None
-        self.__scheme_client = None
         self.__storage_pool_id_allocator = itertools.count(1)
-        self.__config_path = None
+        self.__config_path = ensure_path_exists(
+            os.path.join(self.__configurator.working_dir, self.__cluster_name, "kikimr_configs")
+        )
         self._slot_index_allocator = itertools.count(1)
         self._node_index_allocator = itertools.count(1)
         self.default_channel_bindings = None
@@ -313,9 +294,6 @@ def prepare(self):
             return
 
         self.__initialy_prepared = True
-        self.__client = None
-        self.__kv_client = None
-        self.__scheme_client = None
         self.__instantiate_udfs_dir()
         self.__write_configs()
         for _ in self.__configurator.all_node_ids():
@@ -463,15 +441,6 @@ def restart_nodes(self):
 
     @property
     def config_path(self):
-        if self.__config_path is None:
-            self.__config_path = ensure_path_exists(
-                get_unique_path_for_current_test(
-                    self.__configurator.output_path,
-                    join(
-                        self.__cluster_name, "kikimr_configs"
-                    )
-                )
-            )
         return self.__config_path
 
     def __write_configs(self):