tests: support testenv setup in non-anonymous mode (#15130)

Add support for fully authenticated setup operations to `tests/library/`.
Before that library could only execute cluster setup, configuration and database manipulation in anonymous mode or in the mode when every user is a cluster admin.
Now `tests/library/` can operate when `administration_allowed_lists` is not empty and `enforce_user_token_requirement=True`.

6 files changed, 121 insertions, 44 deletions

diff --git a/ydb/tests/library/clients/kikimr_client.py b/ydb/tests/library/clients/kikimr_client.py
index 013ed5385d..87d7150dfd 100644
--- a/ydb/tests/library/clients/kikimr_client.py
+++ b/ydb/tests/library/clients/kikimr_client.py
@@ -254,8 +254,10 @@ class KiKiMRMessageBusClient(object):
             raise RuntimeError('console_request failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         return response
 
-    def add_config_item(self, config, cookie=None, raise_on_error=True):
+    def add_config_item(self, config, cookie=None, raise_on_error=True, token=None):
         request = msgbus.TConsoleRequest()
+        if token is not None:
+            request.SecurityToken = token
         action = request.ConfigureRequest.Actions.add()
         item = action.AddConfigItem.ConfigItem
         if isinstance(config, str) or isinstance(config, bytes):
diff --git a/ydb/tests/library/common/protobuf_console.py b/ydb/tests/library/common/protobuf_console.py
index 2706cfc1b0..6b8fe411bc 100644
--- a/ydb/tests/library/common/protobuf_console.py
+++ b/ydb/tests/library/common/protobuf_console.py
@@ -2,7 +2,7 @@
 # -*- coding: utf-8 -*-
 import ydb.core.protos.msgbus_pb2 as msgbus
 
-from ydb.tests.library.common.protobuf import AbstractProtobufBuilder
+from ydb.tests.library.common.protobuf import AbstractProtobufBuilder, to_bytes
 
 
 class CreateTenantRequest(AbstractProtobufBuilder):
@@ -14,6 +14,10 @@ class CreateTenantRequest(AbstractProtobufBuilder):
         super(CreateTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.CreateTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.CreateTenantRequest.UserToken = to_bytes(token)
+
     def add_storage_pool(self, pool_type, pool_size):
         pool = self.protobuf.CreateTenantRequest.Request.resources.storage_units.add()
         pool.unit_kind = pool_type
@@ -78,6 +82,10 @@ class AlterTenantRequest(AbstractProtobufBuilder):
         super(AlterTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.AlterTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.AlterTenantRequest.UserToken = to_bytes(token)
+
     def set_schema_quotas(self, schema_quotas):
         quotas = self.protobuf.AlterTenantRequest.Request.schema_operation_quotas
         quotas.SetInParent()
@@ -106,6 +114,10 @@ class GetTenantStatusRequest(AbstractProtobufBuilder):
         super(GetTenantStatusRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.GetTenantStatusRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.GetTenantStatusRequest.UserToken = to_bytes(token)
+
 
 class RemoveTenantRequest(AbstractProtobufBuilder):
     """
@@ -116,6 +128,10 @@ class RemoveTenantRequest(AbstractProtobufBuilder):
         super(RemoveTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.RemoveTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.RemoveTenantRequest.UserToken = to_bytes(token)
+
 
 class SetConfigRequest(AbstractProtobufBuilder):
     """
@@ -158,3 +174,6 @@ class GetOperationRequest(AbstractProtobufBuilder):
     def __init__(self, op_id):
         super(GetOperationRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.GetOperationRequest.id = op_id
+
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
diff --git a/ydb/tests/library/harness/kikimr_cluster_interface.py b/ydb/tests/library/harness/kikimr_cluster_interface.py
index 994b1dbb5a..51972fc4d2 100644
--- a/ydb/tests/library/harness/kikimr_cluster_interface.py
+++ b/ydb/tests/library/harness/kikimr_cluster_interface.py
@@ -110,35 +110,41 @@ class KiKiMRClusterInterface(object):
             )
         return self.__scheme_client
 
-    def get_database_status(self, database_name):
-        response = self.client.send_request(
-            GetTenantStatusRequest(database_name).protobuf,
-            method='ConsoleRequest'
-        ).GetTenantStatusResponse
+    def _send_get_tenant_status_request(self, database_name, token=None):
+        req = GetTenantStatusRequest(database_name)
+
+        if token is not None:
+            req.set_user_token(token)
+
+        return self.client.send_request(req.protobuf, method='ConsoleRequest').GetTenantStatusResponse
+
+    def get_database_status(self, database_name, token=None):
+        response = self._send_get_tenant_status_request(database_name, token=token)
 
         if response.Response.operation.status != StatusIds.SUCCESS:
             logger.critical("Console response status: %s", str(response.Response.operation.status))
             assert False
-            return False
 
         result = cms_tenants_pb.GetDatabaseStatusResult()
         response.Response.operation.result.Unpack(result)
         return result
 
-    def wait_tenant_up(self, database_name):
+    def wait_tenant_up(self, database_name, token=None):
         self.__wait_tenant_up(
             database_name,
-            expected_computational_units=1
+            expected_computational_units=1,
+            token=token,
         )
 
     def __wait_tenant_up(
             self,
             database_name,
             expected_computational_units=None,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None
     ):
         def predicate():
-            result = self.get_database_status(database_name)
+            result = self.get_database_status(database_name, token=token)
 
             if expected_computational_units is None:
                 expected = set([2])
@@ -154,21 +160,25 @@ class KiKiMRClusterInterface(object):
         )
         assert tenant_running
 
-    def __get_console_op(self, op_id):
+    def __get_console_op(self, op_id, token=None):
         req = GetOperationRequest(op_id)
+
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.GetOperationResponse.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('get_console_op failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         return operation
 
-    def __wait_console_op(self, op_id, timeout_seconds, step_seconds=0.5):
+    def __wait_console_op(self, op_id, timeout_seconds, step_seconds=0.5, token=None):
         deadline = time.time() + timeout_seconds
         while True:
             time.sleep(step_seconds)
             if time.time() >= deadline:
                 raise RuntimeError('wait_console_op: deadline exceeded')
-            operation = self.__get_console_op(op_id)
+            operation = self.__get_console_op(op_id, token=token)
             if operation.ready:
                 return operation
 
@@ -177,7 +187,8 @@ class KiKiMRClusterInterface(object):
             database_name,
             storage_pool_units_count,
             disable_external_subdomain=False,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
         for storage_pool_type_name, units_count in storage_pool_units_count.items():
@@ -189,19 +200,23 @@ class KiKiMRClusterInterface(object):
         if disable_external_subdomain:
             req.disable_external_subdomain()
 
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.CreateTenantResponse.Response.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_database failed: %s, %s' % (operation.status, ydb.issues._format_issues(operation.issues)))
 
         self.__wait_tenant_up(
             database_name,
             expected_computational_units=0,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
@@ -209,7 +224,8 @@ class KiKiMRClusterInterface(object):
             self,
             database_name,
             storage_pool_units_count,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
         for storage_pool_type_name, units_count in storage_pool_units_count.items():
@@ -218,19 +234,23 @@ class KiKiMRClusterInterface(object):
                 units_count,
             )
 
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.CreateTenantResponse.Response.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_hostel_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_hostel_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
             expected_computational_units=0,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
@@ -241,10 +261,14 @@ class KiKiMRClusterInterface(object):
             timeout_seconds=120,
             schema_quotas=None,
             disk_quotas=None,
-            attributes=None
+            attributes=None,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
 
+        if token is not None:
+            req.set_user_token(token)
+
         req.share_resources_with(hostel_db)
 
         if schema_quotas is not None:
@@ -262,13 +286,14 @@ class KiKiMRClusterInterface(object):
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_serverless_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_serverless_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
@@ -278,9 +303,13 @@ class KiKiMRClusterInterface(object):
             schema_quotas=None,
             disk_quotas=None,
             timeout_seconds=120,
+            token=None,
     ):
         req = AlterTenantRequest(database_name)
 
+        if token is not None:
+            req.set_user_token(token)
+
         assert schema_quotas is not None or disk_quotas is not None
 
         if schema_quotas is not None:
@@ -294,33 +323,39 @@ class KiKiMRClusterInterface(object):
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('alter_serverless_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('alter_serverless_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
     def remove_database(
             self,
             database_name,
-            timeout_seconds=20
+            timeout_seconds=20,
+            token=None,
     ):
         logger.debug(database_name)
 
-        operation_id = self._remove_database_send_op(database_name)
-        self._remove_database_wait_op(database_name, operation_id, timeout_seconds=timeout_seconds)
-        self._remove_database_wait_tenant_gone(database_name, timeout_seconds=timeout_seconds)
+        operation_id = self._remove_database_send_op(database_name, token=token)
+        self._remove_database_wait_op(database_name, operation_id, timeout_seconds=timeout_seconds, token=token)
+        self._remove_database_wait_tenant_gone(database_name, timeout_seconds=timeout_seconds, token=token)
 
         return database_name
 
-    def _remove_database_send_op(self, database_name):
-        logger.debug('%s: send console operation', database_name)
+    def _remove_database_send_op(self, database_name, token=None):
+        logger.debug('%s: send console operation, token %s', database_name, token)
 
         req = RemoveTenantRequest(database_name)
+
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.RemoveTenantResponse.Response.operation
         logger.debug('%s: response from console: %s', database_name, response)
@@ -330,20 +365,19 @@ class KiKiMRClusterInterface(object):
 
         return operation.id
 
-    def _remove_database_wait_op(self, database_name, operation_id, timeout_seconds=20):
+    def _remove_database_wait_op(self, database_name, operation_id, timeout_seconds=20, token=None):
         logger.debug('%s: wait console operation done', database_name)
-        operation = self.__wait_console_op(operation_id, timeout_seconds=timeout_seconds)
+        operation = self.__wait_console_op(operation_id, timeout_seconds=timeout_seconds, token=token)
         logger.debug('%s: console operation done', database_name)
 
         if operation.status not in (StatusIds.SUCCESS, StatusIds.NOT_FOUND):
             raise RuntimeError('remove_database failed: %s' % (operation.status,))
 
-    def _remove_database_wait_tenant_gone(self, database_name, timeout_seconds=20):
+    def _remove_database_wait_tenant_gone(self, database_name, timeout_seconds=20, token=None):
         logger.debug('%s: wait tenant gone', database_name)
 
         def predicate():
-            response = self.client.send_request(
-                GetTenantStatusRequest(database_name).protobuf, method='ConsoleRequest').GetTenantStatusResponse
+            response = self._send_get_tenant_status_request(database_name, token=token)
             return response.Response.operation.status == StatusIds.NOT_FOUND
 
         tenant_not_found = wait_for(
diff --git a/ydb/tests/library/harness/kikimr_config.py b/ydb/tests/library/harness/kikimr_config.py
index 3d58060ec6..f53b87bdb5 100644
--- a/ydb/tests/library/harness/kikimr_config.py
+++ b/ydb/tests/library/harness/kikimr_config.py
@@ -164,6 +164,8 @@ class KikimrConfigGenerator(object):
             column_shard_config=None,
             use_config_store=False,
             separate_node_configs=False,
+            default_clusteradmin=None,
+            enable_resource_pools=None,
     ):
         if extra_feature_flags is None:
             extra_feature_flags = []
@@ -270,6 +272,8 @@ class KikimrConfigGenerator(object):
 
         # for faster shutdown: there is no reason to wait while tablets are drained before whole cluster is stopping
         self.yaml_config["feature_flags"]["enable_drain_on_shutdown"] = False
+        if enable_resource_pools is not None:
+            self.yaml_config["feature_flags"]["enable_resource_pools"] = enable_resource_pools
         for extra_feature_flag in extra_feature_flags:
             self.yaml_config["feature_flags"][extra_feature_flag] = True
         if enable_alter_database_create_hive_first:
@@ -457,6 +461,16 @@ class KikimrConfigGenerator(object):
         self.use_config_store = use_config_store
         self.separate_node_configs = separate_node_configs
 
+        self.__default_clusteradmin = default_clusteradmin
+        if self.__default_clusteradmin is not None:
+            security_config = self.yaml_config["domains_config"]["security_config"]
+            security_config.setdefault("administration_allowed_sids", []).append(self.__default_clusteradmin)
+            security_config.setdefault("default_access", []).append('+F:{}'.format(self.__default_clusteradmin))
+
+    @property
+    def default_clusteradmin(self):
+        return self.__default_clusteradmin
+
     @property
     def pdisks_info(self):
         return self._pdisks_info
diff --git a/ydb/tests/library/harness/kikimr_runner.py b/ydb/tests/library/harness/kikimr_runner.py
index 45336d6cd7..60c83b5b45 100644
--- a/ydb/tests/library/harness/kikimr_runner.py
+++ b/ydb/tests/library/harness/kikimr_runner.py
@@ -289,7 +289,7 @@ class KiKiMR(kikimr_cluster_interface.KiKiMRClusterInterface):
     def server(self):
         return self.__server
 
-    def __call_kikimr_new_cli(self, cmd, connect_to_server=True):
+    def __call_kikimr_new_cli(self, cmd, connect_to_server=True, token=None):
         server = 'grpc://{server}:{port}'.format(server=self.server, port=self.nodes[1].port)
         binary_path = self.__configurator.get_binary_path(0)
         full_command = [binary_path]
@@ -297,9 +297,15 @@ class KiKiMR(kikimr_cluster_interface.KiKiMRClusterInterface):
             full_command += ["--server={server}".format(server=server)]
         full_command += cmd
 
+        env = None
+        token = token or self.__configurator.default_clusteradmin
+        if token is not None:
+            env = os.environ.copy()
+            env['YDB_TOKEN'] = token
+
         logger.debug("Executing command = {}".format(full_command))
         try:
-            return yatest.common.execute(full_command)
+            return yatest.common.execute(full_command, env=env)
         except yatest.common.ExecutionError as e:
             logger.exception("KiKiMR command '{cmd}' failed with error: {e}\n\tstdout: {out}\n\tstderr: {err}".format(
                 cmd=" ".join(str(x) for x in full_command),
@@ -366,7 +372,7 @@ class KiKiMR(kikimr_cluster_interface.KiKiMRClusterInterface):
         logger.info("Cluster started and initialized")
 
         if bs_needed:
-            self.client.add_config_item(read_binary(__name__, "resources/default_profile.txt"))
+            self.client.add_config_item(read_binary(__name__, "resources/default_profile.txt"), token=self.__configurator.default_clusteradmin)
 
     def __run_node(self, node_id):
         """
diff --git a/ydb/tests/library/harness/ydb_fixtures.py b/ydb/tests/library/harness/ydb_fixtures.py
index 58b33f2b88..17004ec63b 100644
--- a/ydb/tests/library/harness/ydb_fixtures.py
+++ b/ydb/tests/library/harness/ydb_fixtures.py
@@ -79,23 +79,25 @@ def ydb_database_ctx(ydb_cluster, database_path, node_count=1, timeout_seconds=2
     '''???'''
     assert os.path.abspath(database_path), 'database_path should be an (absolute) path, not a database name'
 
-    ydb_cluster.remove_database(database_path, timeout_seconds=timeout_seconds)
+    token = ydb_cluster.config.default_clusteradmin
+
+    ydb_cluster.remove_database(database_path, timeout_seconds=timeout_seconds, token=token)
 
     logger.debug("create database %s: create path and declare internals", database_path)
 
-    ydb_cluster.create_database(database_path, storage_pool_units_count=storage_pools, timeout_seconds=timeout_seconds)
+    ydb_cluster.create_database(database_path, storage_pool_units_count=storage_pools, timeout_seconds=timeout_seconds, token=token)
 
     logger.debug("create database %s: start nodes and construct internals", database_path)
     database_nodes = ydb_cluster.register_and_start_slots(database_path, node_count)
 
     logger.debug("create database %s: wait construction done", database_path)
-    ydb_cluster.wait_tenant_up(database_path)
+    ydb_cluster.wait_tenant_up(database_path, token=token)
 
     logger.debug("create database %s: database up", database_path)
     yield database_path
 
     logger.debug("destroy database %s: remove path and dismantle internals", database_path)
-    ydb_cluster.remove_database(database_path, timeout_seconds=timeout_seconds)
+    ydb_cluster.remove_database(database_path, timeout_seconds=timeout_seconds, token=token)
 
     logger.debug("destroy database %s: stop nodes", database_path)
     ydb_cluster.unregister_and_stop_slots(database_nodes)