MINIFICPP-2666 Move Kafka tests to modular docker tests

behave_framework/pyproject.toml

@@ -7,7 +7,10 @@ dependencies = [
     "behavex==4.6.0",
     "docker==7.1.0",
     "PyYAML==6.0.3",
-    "humanfriendly==10.0"
+    "humanfriendly==10.0",
+    "m2crypto==0.41.0",
+    "pyopenssl==25.0.0",
+    "pyjks==20.0.0"
 ]
 
 [tool.setuptools]

behave_framework/src/minifi_test_framework/containers/container.py

@@ -19,6 +19,7 @@
 import os
 import shlex
 import tempfile
+import tarfile
 
 import docker
 from docker.models.networks import Network
@@ -48,23 +49,34 @@ def __init__(self, image_name: str, container_name: str, network: Network, comma
     def add_host_file(self, host_path: str, container_path: str, mode: str = "ro"):
         self.host_files.append(HostFile(container_path, host_path, mode))
 
-    def deploy(self) -> bool:
-        self._temp_dir = tempfile.TemporaryDirectory()
+    def _write_content_to_file(self, filepath: str, permissions: int | None, content: str | bytes):
+        write_mode = "w"
+        if isinstance(content, bytes):
+            write_mode = "wb"
+        with open(filepath, write_mode) as file:
+            file.write(content)
+        if permissions:
+            os.chmod(filepath, permissions)
 
+    def _configure_volumes_of_container_files(self):
         for file in self.files:
-            file_name = os.path.basename(file.path)
-            temp_path = os.path.join(self._temp_dir.name, file_name)
-            with open(temp_path, "w") as temp_file:
-                temp_file.write(file.content)
+            temp_path = os.path.join(self._temp_dir.name, os.path.basename(file.path))
+            self._write_content_to_file(temp_path, file.permissions, file.content)
             self.volumes[temp_path] = {"bind": file.path, "mode": file.mode}
+
+    def _configure_volumes_of_container_dirs(self):
         for directory in self.dirs:
             temp_path = self._temp_dir.name + directory.path
             os.makedirs(temp_path, exist_ok=True)
             for file_name, content in directory.files.items():
                 file_path = os.path.join(temp_path, file_name)
-                with open(file_path, "w") as temp_file:
-                    temp_file.write(content)
+                self._write_content_to_file(file_path, None, content)
             self.volumes[temp_path] = {"bind": directory.path, "mode": directory.mode}
+
+    def deploy(self) -> bool:
+        self._temp_dir = tempfile.TemporaryDirectory()
+        self._configure_volumes_of_container_files()
+        self._configure_volumes_of_container_dirs()
         for host_file in self.host_files:
             self.volumes[host_file.host_path] = {"bind": host_file.container_path, "mode": host_file.mode}
 
@@ -85,6 +97,30 @@ def deploy(self) -> bool:
             raise
         return True
 
+    def start(self):
+        if self.container:
+            self.container.start()
+        else:
+            logging.error("Container does not exist. Cannot start.")
+
+    def stop(self):
+        if self.container:
+            self.container.stop()
+        else:
+            logging.error("Container does not exist. Cannot stop.")
+
+    def kill(self):
+        if self.container:
+            self.container.kill()
+        else:
+            logging.error("Container does not exist. Cannot kill.")
+
+    def restart(self):
+        if self.container:
+            self.container.restart()
+        else:
+            logging.error("Container does not exist. Cannot restart.")
+
     def clean_up(self):
         if self.container:
             try:
@@ -203,9 +239,14 @@ def exited(self) -> bool:
             return False
 
     def get_number_of_files(self, directory_path: str) -> int:
-        if not self.container or not self.not_empty_dir_exists(directory_path):
+        if not self.container:
+            logging.warning("Container not running")
             return -1
 
+        if not self.not_empty_dir_exists(directory_path):
+            logging.warning(f"Container directory does not exist: {directory_path}")
+            return 0
+
         count_command = f"sh -c 'find {directory_path} -maxdepth 1 -type f | wc -l'"
         exit_code, output = self.exec_run(count_command)
 
@@ -214,13 +255,15 @@ def get_number_of_files(self, directory_path: str) -> int:
             return -1
 
         try:
-            return int(output.strip())
+            file_count = int(output.strip())
+            logging.debug(f"Number of files in '{directory_path}': {file_count}")
+            return file_count
         except (ValueError, IndexError):
             logging.error(f"Error parsing output '{output}' from command '{count_command}'")
             return -1
 
-    def verify_file_contents(self, directory_path: str, expected_contents: list[str]) -> bool:
-        if not self.container or not self.not_empty_dir_exists(directory_path):
+    def _verify_file_contents_in_running_container(self, directory_path: str, expected_contents: list[str]) -> bool:
+        if not self.not_empty_dir_exists(directory_path):
             return False
 
         safe_dir_path = shlex.quote(directory_path)
@@ -254,6 +297,59 @@ def verify_file_contents(self, directory_path: str, expected_contents: list[str]
 
         return sorted(actual_file_contents) == sorted(expected_contents)
 
+    def _verify_file_contents_in_stopped_container(self, directory_path: str, expected_contents: list[str]) -> bool:
+        if not self.container:
+            return False
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            extracted_dir = self._extract_directory_from_container(directory_path, temp_dir)
+            if not extracted_dir:
+                return False
+
+            actual_file_contents = self._read_files_from_directory(extracted_dir)
+            if actual_file_contents is None:
+                return False
+
+            return sorted(actual_file_contents) == sorted(expected_contents)
+
+    def _extract_directory_from_container(self, directory_path: str, temp_dir: str) -> str | None:
+        try:
+            bits, _ = self.container.get_archive(directory_path)
+            temp_tar_path = os.path.join(temp_dir, "archive.tar")
+
+            with open(temp_tar_path, 'wb') as f:
+                for chunk in bits:
+                    f.write(chunk)
+
+            with tarfile.open(temp_tar_path) as tar:
+                tar.extractall(path=temp_dir)
+
+            return os.path.join(temp_dir, os.path.basename(directory_path.strip('/')))
+        except Exception as e:
+            logging.error(f"Error extracting files from directory path '{directory_path}' from container '{self.container_name}': {e}")
+            return None
+
+    def _read_files_from_directory(self, directory_path: str) -> list[str] | None:
+        try:
+            file_contents = []
+            for entry in os.scandir(directory_path):
+                if entry.is_file():
+                    with open(entry.path, 'r') as f:
+                        file_contents.append(f.read())
+            return file_contents
+        except Exception as e:
+            logging.error(f"Error reading extracted files: {e}")
+            return None
+
+    def verify_file_contents(self, directory_path: str, expected_contents: list[str]) -> bool:
+        if not self.container:
+            return False
+
+        if self.container.status == "running":
+            return self._verify_file_contents_in_running_container(directory_path, expected_contents)
+
+        return self._verify_file_contents_in_stopped_container(directory_path, expected_contents)
+
     def log_app_output(self) -> bool:
         logs = self.get_logs()
         logging.info("Logs of container '%s':", self.container_name)

behave_framework/src/minifi_test_framework/containers/file.py

@@ -16,7 +16,8 @@
 #
 
 class File:
-    def __init__(self, path, content):
+    def __init__(self, path, content: str | bytes, mode="rw", permissions=None):
         self.path = path
         self.content = content
-        self.mode = "rw"
+        self.mode = mode
+        self.permissions = permissions

behave_framework/src/minifi_test_framework/containers/minifi_container.py

@@ -16,34 +16,43 @@
 #
 
 import logging
-from docker.models.networks import Network
+from OpenSSL import crypto
 
+from minifi_test_framework.core.minifi_test_context import MinifiTestContext
 from minifi_test_framework.containers.file import File
 from minifi_test_framework.minifi.flow_definition import FlowDefinition
+from minifi_test_framework.core.ssl_utils import make_cert_without_extended_usage
 from .container import Container
 
 
 class MinifiContainer(Container):
-    def __init__(self, image_name: str, container_name: str, scenario_id: str, network: Network):
-        super().__init__(image_name, f"{container_name}-{scenario_id}", network)
+    def __init__(self, container_name: str, test_context: MinifiTestContext):
+        super().__init__(test_context.minifi_container_image, f"{container_name}-{test_context.scenario_id}", test_context.network)
         self.flow_config_str: str = ""
         self.flow_definition = FlowDefinition()
         self.properties: dict[str, str] = {}
         self.log_properties: dict[str, str] = {}
 
-        self.is_fhs = 'MINIFI_INSTALLATION_TYPE=FHS' in str(self.client.images.get(image_name).history())
+        minifi_client_cert, minifi_client_key = make_cert_without_extended_usage(common_name=self.container_name, ca_cert=test_context.root_ca_cert, ca_key=test_context.root_ca_key)
+        self.files.append(File("/tmp/resources/root_ca.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=test_context.root_ca_cert)))
+        self.files.append(File("/tmp/resources/minifi_client.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=minifi_client_cert)))
+        self.files.append(File("/tmp/resources/minifi_client.key", crypto.dump_privatekey(type=crypto.FILETYPE_PEM, pkey=minifi_client_key)))
+
+        self.is_fhs = 'MINIFI_INSTALLATION_TYPE=FHS' in str(self.client.images.get(test_context.minifi_container_image).history())
 
         self._fill_default_properties()
         self._fill_default_log_properties()
 
     def deploy(self) -> bool:
+        flow_config = self.flow_definition.to_yaml()
+        logging.info(f"Deploying MiNiFi container '{self.container_name}' with flow configuration:\n{flow_config}")
         if self.is_fhs:
-            self.files.append(File("/etc/nifi-minifi-cpp/config.yml", self.flow_definition.to_yaml()))
+            self.files.append(File("/etc/nifi-minifi-cpp/config.yml", flow_config))
             self.files.append(File("/etc/nifi-minifi-cpp/minifi.properties", self._get_properties_file_content()))
             self.files.append(
                 File("/etc/nifi-minifi-cpp/minifi-log.properties", self._get_log_properties_file_content()))
         else:
-            self.files.append(File("/opt/minifi/minifi-current/conf/config.yml", self.flow_definition.to_yaml()))
+            self.files.append(File("/opt/minifi/minifi-current/conf/config.yml", flow_config))
             self.files.append(
                 File("/opt/minifi/minifi-current/conf/minifi.properties", self._get_properties_file_content()))
             self.files.append(File("/opt/minifi/minifi-current/conf/minifi-log.properties",

behave_framework/src/minifi_test_framework/core/helpers.py

@@ -31,6 +31,15 @@ def log_due_to_failure(context: MinifiTestContext | None):
             container.log_app_output()
 
 
+def check_condition_after_wait(condition: Callable[[], bool], context: MinifiTestContext | None, wait_time: int) -> bool:
+    time.sleep(wait_time)
+    if not condition():
+        logging.warning("Condition not met after wait")
+        log_due_to_failure(context)
+        return False
+    return True
+
+
 def wait_for_condition(condition: Callable[[], bool], timeout_seconds: float, bail_condition: Callable[[], bool],
                        context: MinifiTestContext | None) -> bool:
     if bail_condition():
@@ -50,8 +59,8 @@ def wait_for_condition(condition: Callable[[], bool], timeout_seconds: float, ba
             sleep_time = min(1.0, remaining_time)
             if sleep_time > 0:
                 time.sleep(sleep_time)
-    except (Exception,):
-        logging.warning("Exception while waiting for condition")
+    except Exception as ex:
+        logging.warning("Exception while waiting for condition: %s", ex)
         log_due_to_failure(context)
         return False
     logging.warning("Timed out after %d seconds while waiting for condition", timeout_seconds)

behave_framework/src/minifi_test_framework/core/hooks.py

@@ -23,6 +23,7 @@
 from behave.model import Scenario
 from behave.runner import Context
 
+from minifi_test_framework.core.ssl_utils import make_self_signed_cert
 from minifi_test_framework.core.minifi_test_context import MinifiTestContext
 
 
@@ -74,6 +75,7 @@ def common_before_scenario(context: Context, scenario: Scenario):
     context.network = docker_client.networks.create(network_name)
     context.containers = {}
     context.resource_dir = None
+    context.root_ca_cert, context.root_ca_key = make_self_signed_cert("root CA")
 
     for step in scenario.steps:
         inject_scenario_id(context, step)

behave_framework/src/minifi_test_framework/core/minifi_test_context.py

@@ -15,11 +15,16 @@
 #  limitations under the License.
 #
 
+from __future__ import annotations
+from typing import TYPE_CHECKING
 from behave.runner import Context
 from docker.models.networks import Network
 
 from minifi_test_framework.containers.container import Container
-from minifi_test_framework.containers.minifi_container import MinifiContainer
+from OpenSSL import crypto
+
+if TYPE_CHECKING:
+    from minifi_test_framework.containers.minifi_container import MinifiContainer
 
 DEFAULT_MINIFI_CONTAINER_NAME = "minifi-primary"
 
@@ -30,10 +35,13 @@ class MinifiTestContext(Context):
     network: Network
     minifi_container_image: str
     resource_dir: str | None
+    root_ca_key: crypto.PKey
+    root_ca_cert: crypto.X509
 
     def get_or_create_minifi_container(self, container_name: str) -> MinifiContainer:
         if container_name not in self.containers:
-            self.containers[container_name] = MinifiContainer(self.minifi_container_image, container_name, self.scenario_id, self.network)
+            from minifi_test_framework.containers.minifi_container import MinifiContainer
+            self.containers[container_name] = MinifiContainer(container_name, self)
         return self.containers[container_name]
 
     def get_or_create_default_minifi_container(self) -> MinifiContainer: