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

Co-authored-by: Torsten Kilias <[email protected]>
Co-authored-by: Torsten Kilias <[email protected]>

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/api/spawn_test_environment.py

@@ -71,6 +71,7 @@ def spawn_test_environment(
     """
     def str_or_none(x: any) -> str:
         return str(x) if x is not None else None
+
     parsed_db_mem_size = humanfriendly.parse_size(db_mem_size)
     if parsed_db_mem_size < humanfriendly.parse_size("1 GiB"):
         raise ArgumentConstraintError("db_mem_size", "needs to be at least 1 GiB")

exasol_integration_test_docker_environment/lib/base/db_os_executor.py

@@ -0,0 +1,142 @@
+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:
+        with ContextDockerClient(timeout=self._timeout) as client:
+            return client
+
+
+# Avoid TypeError: Instance and class checks can only be
+# used with @runtime_checkable protocols
+# raised by unit tests
+@runtime_checkable
+class DbOsExecutor(Protocol):
+    """
+    This class provides an abstraction to execute operating system
+    commands on the database host, e.g. 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) -> ExecResult:
+        return self._container.exec_run(cmd)
+
+    def close(self):
+        self._container = None
+        if self._client is not 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)
+        output = result.stdout.encode("utf-8")
+        return ExecResult(result.exited, output)
+
+    def close(self):
+        if self._connection is not None:
+            self._connection.close()
+            self._connection = None
+
+
+# Avoid TypeError: Instance and class checks can only be
+# used with @runtime_checkable protocols
+# raised by integration tests
+@runtime_checkable
+class DbOsExecFactory(Protocol):
+    """
+    This class defines an abstract method ``executor()`` to be implemented by
+    inheriting factories.
+    """
+
+    @abstractmethod
+    def executor(self) -> DbOsExecutor:
+        """
+        Create an executor for executing commands inside of the operating
+        system of the database.
+        """
+        ...
+
+
+class DockerExecFactory(DbOsExecFactory):
+    def __init__(self, container_name: str, client_factory: DockerClientFactory):
+        self._container_name = container_name
+        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 DbOsExecutor
 
 
 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: DbOsExecutor):
         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 = executor
 
     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
+import docker
+from exasol_integration_test_docker_environment.lib.base.db_os_executor \
+    import DbOsExecutor
 
 
-def find_exaplus(db_container: docker.models.containers.Container) -> PurePath:
+def find_exaplus(
+        db_container: docker.models.containers.Container,
+        os_executor: DbOsExecutor,
+) -> 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