Create SSH access replacements for calls to docker.exec_run()

doc/changes/changes_2.0.0.md

@@ -29,3 +29,4 @@ If you need further versions, please open an issue.
 * #308: Unified ports for database, BucketFS, and SSH
 * #322: Added additional tests for environment variable LOG_ENV_VARIABLE_NAME
 * #359: Fixed custom logging path not working if dir does not exist.
+* #304: Create SSH access replacements for calls to `docker.exec_run()`

exasol_integration_test_docker_environment/lib/base/db_os_executor.py

@@ -0,0 +1,132 @@
+from abc import abstractmethod
+import fabric
+import docker
+from docker import DockerClient
+from typing import Protocol, runtime_checkable
+from docker.models.containers import Container, ExecResult
+from exasol_integration_test_docker_environment \
+    .lib.base.ssh_access import SshKey
+from exasol_integration_test_docker_environment \
+    .lib.data.database_info import DatabaseInfo
+from exasol_integration_test_docker_environment.lib.docker \
+    import ContextDockerClient
+
+
+class DockerClientFactory:
+    """
+    Create a Docker client.
+    """
+    def __init__(self, timeout: int = 100000):
+        self._timeout = timeout
+
+    def client(self) -> DockerClient:
+        return ContextDockerClient(timeout=self._timeout)
+
+
+@runtime_checkable
+class DbOsExecutor(Protocol):
+    """
+    This class provides an abstraction to execute commands inside a Docker
+    Container.  See concrete implementations in sub-classes
+    ``DockerExecutor`` and ``SshExecutor``.
+    """
+    @abstractmethod
+    def exec(self, cmd: str) -> ExecResult:
+        ...
+
+
+class DockerExecutor(DbOsExecutor):
+    def __init__(self, docker_client: DockerClient, container_name: str):
+        self._client = docker_client
+        self._container_name = container_name
+        self._container = None
+
+    def __enter__(self):
+        self._container = self._client.containers.get(self._container_name)
+        return self
+
+    def __exit__(self, type_, value, traceback):
+        self.close()
+
+    def __del__(self):
+        self.close()
+
+    def exec(self, cmd: str):
+        return self._container.exec_run(cmd)
+
+    def close(self):
+        self._container = None
+        self._client.close()
+        self._client = None
+
+
+class SshExecutor(DbOsExecutor):
+    def __init__(self, connect_string: str, key_file: str):
+        self._connect_string = connect_string
+        self._key_file = key_file
+        self._connection = None
+
+    def __enter__(self):
+        key = SshKey.read_from(self._key_file)
+        self._connection = fabric.Connection(
+            self._connect_string,
+            connect_kwargs={ "pkey": key.private },
+        )
+        return self
+
+    def __exit__(self, type_, value, traceback):
+        self.close()
+
+    def __del__(self):
+        self.close()
+
+    def exec(self, cmd: str) -> ExecResult:
+        result = self._connection.run(cmd)
+        return ExecResult(result.exited, result.stdout)
+
+    def close(self):
+        if self._connection is not None:
+            self._connection.close()
+            self._connection = None
+
+
+@runtime_checkable
+class DbOsExecFactory(Protocol):
+    """
+    This class defines abstract method ``executor()`` to be implemented by
+    inheriting factories.
+    """
+
+    @abstractmethod
+    def executor(self) -> DbOsExecutor:
+        """Create an executor for executing commands inside a Docker
+        Container."""
+        ...
+
+
+class DockerExecFactory(DbOsExecFactory):
+    def __init__(self, container_name: str, client_factory: DockerClientFactory):
+        self._container_name = container_name
+        if client_factory is None:
+            client_factory = DockerClientFactory()
+        self._client_factory = client_factory
+
+    def executor(self) -> DbOsExecutor:
+        client = self._client_factory.client()
+        return DockerExecutor(client, self._container_name)
+
+
+class SshExecFactory(DbOsExecFactory):
+    @classmethod
+    def from_database_info(cls, info: DatabaseInfo):
+        return SshExecFactory(
+            f"{info.ssh_info.user}@{info.host}:{info.ports.ssh}",
+            info.ssh_info.key_file,
+        )
+
+    def __init__(self, connect_string: str, ssh_key_file: str):
+        self._connect_string = connect_string
+        self._key_file = ssh_key_file
+
+    def executor(self) -> DbOsExecutor:
+        return SshExecutor(self._connect_string, self._key_file)

exasol_integration_test_docker_environment/lib/test_environment/abstract_spawn_test_environment.py

@@ -144,24 +144,33 @@ def _create_network(self, attempt):
     def create_network_task(self, attempt: int):
         raise AbstractMethodException()
 
-    def _spawn_database_and_test_container(self,
-                                           network_info: DockerNetworkInfo,
-                                           certificate_volume_info: Optional[DockerVolumeInfo],
-                                           attempt: int) -> Tuple[DatabaseInfo, Optional[ContainerInfo]]:
-        certificate_volume_name = certificate_volume_info.volume_name if certificate_volume_info is not None else None
-        dependencies_tasks = {
-                DATABASE: self.create_spawn_database_task(network_info, certificate_volume_info, attempt)
-            }
-        if self.test_container_content is not None:
-            dependencies_tasks[TEST_CONTAINER] = \
-                self.create_spawn_test_container_task(network_info, certificate_volume_name, attempt)
-        database_and_test_container_info_future = yield from self.run_dependencies(dependencies_tasks)
-        database_and_test_container_info = \
-            self.get_values_from_futures(database_and_test_container_info_future)
-        test_container_info = None
+    def _spawn_database_and_test_container(
+            self,
+            network_info: DockerNetworkInfo,
+            certificate_volume_info: Optional[DockerVolumeInfo],
+            attempt: int,
+    ) -> Tuple[DatabaseInfo, Optional[ContainerInfo]]:
+        def volume_name(info):
+            return None if info is None else info.volume_name
+
+        child_tasks = {
+            DATABASE: self.create_spawn_database_task(
+                network_info,
+                certificate_volume_info,
+                attempt,
+            )
+        }
         if self.test_container_content is not None:
-            test_container_info = database_and_test_container_info[TEST_CONTAINER]
-        database_info = database_and_test_container_info[DATABASE]
+            certificate_volume_name = volume_name(certificate_volume_info)
+            child_tasks[TEST_CONTAINER] = self.create_spawn_test_container_task(
+                network_info,
+                certificate_volume_name,
+                attempt,
+            )
+        futures = yield from self.run_dependencies(child_tasks)
+        results = self.get_values_from_futures(futures)
+        database_info = results[DATABASE]
+        test_container_info = results[TEST_CONTAINER] if self.test_container_content is not None else None
         return database_info, test_container_info
 
     def create_spawn_database_task(self,

exasol_integration_test_docker_environment/lib/test_environment/create_certificates/create_ssl_certificates_task.py

@@ -112,26 +112,27 @@ def create_certificate(self, image_infos: Dict[str, ImageInfo]) -> None:
 
         with self._get_docker_client() as docker_client:
             try:
-                test_container = \
-                    docker_client.containers.create(
-                        image=certificate_container_image_info.get_target_complete_name(),
-                        name="certificate_resources",
-                        network_mode=None,
-                        command="sleep infinity",
-                        detach=True,
-                        volumes=volumes,
-                        labels={"test_environment_name": self.environment_name,
-                                "container_type": "certificate_resources"},
-                        runtime=self.docker_runtime
-                    )
-                test_container.start()
+                container = docker_client.containers.create(
+                    image=certificate_container_image_info.get_target_complete_name(),
+                    name="certificate_resources",
+                    network_mode=None,
+                    command="sleep infinity",
+                    detach=True,
+                    volumes=volumes,
+                    labels={
+                        "test_environment_name": self.environment_name,
+                        "container_type": "certificate_resources",
+                    },
+                    runtime=self.docker_runtime
+                )
+                container.start()
                 self.logger.info("Creating certificates...")
                 cmd = f"bash /scripts/create_certificates.sh " \
                       f"{self._construct_complete_host_name} {CERTIFICATES_MOUNT_PATH}"
-                exit_code, output = test_container.exec_run(cmd)
+                exit_code, output = container.exec_run(cmd)
                 self.logger.info(output.decode('utf-8'))
                 if exit_code != 0:
                     raise RuntimeError(f"Error creating certificates:'{output.decode('utf-8')}'")
             finally:
-                test_container.stop()
-                test_container.remove()
+                container.stop()
+                container.remove()

exasol_integration_test_docker_environment/lib/test_environment/database_setup/docker_db_log_based_bucket_sync_checker.py

@@ -3,8 +3,11 @@
 
 from docker.models.containers import Container
 
-from exasol_integration_test_docker_environment.lib.test_environment.database_setup.bucketfs_sync_checker import \
-    BucketFSSyncChecker
+from exasol_integration_test_docker_environment \
+    .lib.test_environment.database_setup.bucketfs_sync_checker \
+    import BucketFSSyncChecker
+from exasol_integration_test_docker_environment \
+    .lib.base.db_os_executor import DbOsExecFactory
 
 
 class DockerDBLogBasedBucketFSSyncChecker(BucketFSSyncChecker):
@@ -13,12 +16,14 @@ def __init__(self, logger,
                  database_container: Container,
                  pattern_to_wait_for: str,
                  log_file_to_check: str,
-                 bucketfs_write_password: str):
+                 bucketfs_write_password: str,
+                 executor_factory: DbOsExecFactory):
         self.logger = logger
         self.pattern_to_wait_for = pattern_to_wait_for
         self.log_file_to_check = log_file_to_check
         self.database_container = database_container
         self.bucketfs_write_password = bucketfs_write_password
+        self.executor_factory = executor_factory
 
     def prepare_upload(self):
         self.start_exit_code, self.start_output = self.find_pattern_in_logfile()

exasol_integration_test_docker_environment/lib/test_environment/database_setup/find_exaplus_in_db_container.py

@@ -1,9 +1,14 @@
 from pathlib import PurePath
 
 import docker.models.containers
+from exasol_integration_test_docker_environment.lib.base.db_os_executor import \
+    DbOsExecFactory
 
 
-def find_exaplus(db_container: docker.models.containers.Container) -> PurePath:
+def find_exaplus(
+        db_container: docker.models.containers.Container,
+        executor_factory: DbOsExecFactory,
+) -> PurePath:
     """
     Tries to find path of exaplus in given container in directories where exaplus is normally installed.
     :db_container Container where to search for exaplus

exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py

@@ -15,11 +15,16 @@
 from exasol_integration_test_docker_environment.lib.base.json_pickle_parameter import JsonPickleParameter
 from exasol_integration_test_docker_environment.lib.base.still_running_logger import StillRunningLogger, \
     StillRunningLoggerThread
-from exasol_integration_test_docker_environment.lib.data.environment_info import EnvironmentInfo
-from exasol_integration_test_docker_environment.lib.test_environment.database_setup.docker_db_log_based_bucket_sync_checker import \
-    DockerDBLogBasedBucketFSSyncChecker
-from exasol_integration_test_docker_environment.lib.test_environment.database_setup.time_based_bucketfs_sync_waiter import \
-    TimeBasedBucketFSSyncWaiter
+from exasol_integration_test_docker_environment.lib.data.environment_info \
+    import EnvironmentInfo
+from exasol_integration_test_docker_environment \
+    .lib.test_environment.database_setup.docker_db_log_based_bucket_sync_checker \
+    import DockerDBLogBasedBucketFSSyncChecker
+from exasol_integration_test_docker_environment \
+    .lib.test_environment.database_setup.time_based_bucketfs_sync_waiter \
+    import TimeBasedBucketFSSyncWaiter
+from exasol_integration_test_docker_environment \
+    .lib.base.db_os_executor import DbOsExecFactory
 
 
 @dataclasses.dataclass
@@ -35,6 +40,7 @@ class UploadFileToBucketFS(DockerBaseTask):
     reuse_uploaded = luigi.BoolParameter(False, significant=False)
     bucketfs_write_password = luigi.Parameter(
         significant=False, visibility=luigi.parameter.ParameterVisibility.HIDDEN)
+    executor_factory=JsonPickleParameter(DbOsExecFactory, significant=False)
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -103,7 +109,8 @@ def get_sync_checker(self, database_container: Container,
                 log_file_to_check=log_file,
                 pattern_to_wait_for=pattern_to_wait_for,
                 logger=self.logger,
-                bucketfs_write_password=str(self.bucketfs_write_password)
+                bucketfs_write_password=str(self.bucketfs_write_password),
+                executor_factory=self.executor_factory,
             )
         else:
             return TimeBasedBucketFSSyncWaiter(sync_time_estimation)

exasol_integration_test_docker_environment/lib/test_environment/database_waiters/is_database_ready_thread.py

@@ -8,6 +8,8 @@
 from exasol_integration_test_docker_environment.lib.data.database_info import DatabaseInfo
 from exasol_integration_test_docker_environment.lib.test_environment.database_setup.find_exaplus_in_db_container import \
     find_exaplus
+from exasol_integration_test_docker_environment.lib.base.db_os_executor import \
+    DbOsExecFactory
 
 
 class IsDatabaseReadyThread(Thread):
@@ -17,7 +19,8 @@ def __init__(self,
                  database_info: DatabaseInfo,
                  database_container: Container,
                  database_credentials: DatabaseCredentials,
-                 docker_db_image_version: str):
+                 docker_db_image_version: str,
+                 executor_factory: DbOsExecFactory):
         super().__init__()
         self.logger = logger
         self.database_credentials = database_credentials
@@ -28,6 +31,7 @@ def __init__(self,
         self.output_db_connection = None
         self.output_bucketfs_connection = None
         self.docker_db_image_version = docker_db_image_version
+        self.executor_factory = executor_factory
 
     def stop(self):
         self.logger.info("Stop IsDatabaseReadyThread")
@@ -37,7 +41,7 @@ def run(self):
         db_connection_command = ""
         bucket_fs_connection_command = ""
         try:
-            exaplus_path = find_exaplus(self._db_container)
+            exaplus_path = find_exaplus(self._db_container, self.executor_factory)
             db_connection_command = self.create_db_connection_command(exaplus_path)
             bucket_fs_connection_command = self.create_bucketfs_connection_command()
         except RuntimeError as e: