MINIFICPP-2668 Move standard processor tests to modular docker tests

Author: lordgamez

behave_framework/src/minifi_test_framework/containers/container.py

@@ -137,6 +137,16 @@ def not_empty_dir_exists(self, directory_path: str) -> bool:
             "sh -c {}".format(shlex.quote(f'[ "$(ls -A {directory_path})" ]')))
         return dir_not_empty_ec == 0
 
+    def directory_contains_empty_file(self, directory_path: str) -> bool:
+        if not self.container or not self.not_empty_dir_exists(directory_path):
+            return False
+
+        command = "sh -c {}".format(shlex.quote(f"find {directory_path} -maxdepth 1 -type f -empty"))
+
+        exit_code, _ = self.exec_run(command)
+
+        return exit_code == 0
+
     def directory_contains_file_with_content(self, directory_path: str, expected_content: str) -> bool:
         if not self.container or not self.not_empty_dir_exists(directory_path):
             return False
@@ -250,25 +260,18 @@ def get_number_of_files(self, directory_path: str) -> int:
             logging.error(f"Error parsing output '{output}' from command '{count_command}'")
             return -1
 
-    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
-
+    def _get_contents_of_all_files_in_directory(self, directory_path: str) -> list[str] | None:
         safe_dir_path = shlex.quote(directory_path)
         list_files_command = f"find {safe_dir_path} -mindepth 1 -maxdepth 1 -type f -print0"
 
         exit_code, output = self.exec_run(f"sh -c \"{list_files_command}\"")
 
         if exit_code != 0:
             logging.error(f"Error running command '{list_files_command}': {output}")
-            return False
+            return None
 
         actual_filepaths = [path for path in output.split('\0') if path]
 
-        if len(actual_filepaths) != len(expected_contents):
-            logging.debug(f"Expected {len(expected_contents)} files, but found {len(actual_filepaths)}")
-            return False
-
         actual_file_contents = []
         for path in actual_filepaths:
             safe_path = shlex.quote(path)
@@ -279,10 +282,24 @@ def _verify_file_contents_in_running_container(self, directory_path: str, expect
             if exit_code != 0:
                 error_message = f"Command to read file '{path}' failed with exit code {exit_code}"
                 logging.error(error_message)
-                return False
+                return None
 
             actual_file_contents.append(content)
 
+        return actual_file_contents
+
+    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
+
+        actual_file_contents = self._get_contents_of_all_files_in_directory(directory_path)
+        if actual_file_contents is None:
+            return False
+
+        if len(actual_file_contents) != len(expected_contents):
+            logging.debug(f"Expected {len(expected_contents)} files, but found {len(actual_file_contents)}")
+            return False
+
         return sorted(actual_file_contents) == sorted(expected_contents)
 
     def _verify_file_contents_in_stopped_container(self, directory_path: str, expected_contents: list[str]) -> bool:
@@ -347,6 +364,7 @@ def log_app_output(self) -> bool:
 
     def verify_path_with_json_content(self, directory_path: str, expected_str: str) -> bool:
         if not self.container or not self.not_empty_dir_exists(directory_path):
+            logging.warning(f"Container not running or directory does not exist: {directory_path}")
             return False
 
         count_command = f"sh -c 'find {directory_path} -maxdepth 1 -type f | wc -l'"
@@ -378,3 +396,25 @@ def verify_path_with_json_content(self, directory_path: str, expected_str: str)
         expected_json = json.loads(expected_str)
 
         return actual_json == expected_json
+
+    def directory_contains_file_with_json_content(self, directory_path: str, expected_content: str) -> bool:
+        if not self.container or not self.not_empty_dir_exists(directory_path):
+            logging.warning(f"Container not running or directory does not exist: {directory_path}")
+            return False
+
+        actual_file_contents = self._get_contents_of_all_files_in_directory(directory_path)
+        if actual_file_contents is None:
+            return False
+
+        for file_content in actual_file_contents:
+            try:
+                actual_json = json.loads(file_content)
+                expected_json = json.loads(expected_content)
+                if actual_json == expected_json:
+                    return True
+                logging.warning(f"File content does not match expected JSON: {file_content}")
+            except json.JSONDecodeError:
+                logging.error("Error decoding JSON content from file.")
+                continue
+
+        return False

behave_framework/src/minifi_test_framework/containers/minifi_container.py

@@ -66,6 +66,9 @@ def set_property(self, key: str, value: str):
     def set_log_property(self, key: str, value: str):
         self.log_properties[key] = value
 
+    def enable_openssl_fips_mode(self):
+        self.properties["nifi.openssl.fips.support.enable"] = "true"
+
     def _fill_default_properties(self):
         if self.is_fhs:
             self.properties["nifi.flow.configuration.file"] = "/etc/nifi-minifi-cpp/config.yml"

behave_framework/src/minifi_test_framework/steps/checking_steps.py

@@ -154,8 +154,8 @@ def step_impl(context: MinifiTestContext, directory: str, timeout: str, contents
                               context=context)
 
 
-@then("a flowfile with the JSON content \"{content}\" is placed in {directory} in less than {duration}")
-@then("a flowfile with the JSON content '{content}' is placed in {directory} in less than {duration}")
+@then("a file with the JSON content \"{content}\" is placed in the \"{directory}\" directory in less than {duration}")
+@then("a file with the JSON content '{content}' is placed in the '{directory}' directory in less than {duration}")
 def step_impl(context: MinifiTestContext, content: str, directory: str, duration: str):
     timeout_in_seconds = humanfriendly.parse_timespan(duration)
     assert wait_for_condition(
@@ -173,6 +173,15 @@ def step_impl(context: MinifiTestContext, max_increase: str, duration: str):
     assert final_memory_usage - initial_memory_usage <= max_increase_in_bytes
 
 
+@then("at least one file with the JSON content \"{content}\" is placed in the \"{directory}\" directory in less than {duration}")
+@then("at least one file with the JSON content '{content}' is placed in the '{directory}' directory in less than {duration}")
+def step_impl(context: MinifiTestContext, content: str, directory: str, duration: str):
+    timeout_in_seconds = humanfriendly.parse_timespan(duration)
+    assert wait_for_condition(
+        condition=lambda: context.get_default_minifi_container().directory_contains_file_with_json_content(directory, content),
+        timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.get_default_minifi_container().exited, context=context)
+
+
 @then('after a wait of {duration}, at least {lower_bound:d} and at most {upper_bound:d} flowfiles are produced and placed in the "{directory}" directory')
 def step_impl(context, lower_bound, upper_bound, duration, directory):
     duration_seconds = humanfriendly.parse_timespan(duration)
@@ -196,3 +205,11 @@ def step_impl(context, directory, duration, contents):
 @then('exactly these files are in the "{directory}" directory in less than {duration}: ""')
 def step_impl(context, directory, duration):
     context.execute_steps(f'then no files are placed in the "{directory}" directory in {duration} of running time')
+
+
+@then("at least one empty file is placed in the \"{directory}\" directory in less than {duration}")
+def step_impl(context, directory, duration):
+    timeout_in_seconds = humanfriendly.parse_timespan(duration)
+    assert wait_for_condition(
+        condition=lambda: context.get_default_minifi_container().directory_contains_empty_file(directory),
+        timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.get_default_minifi_container().exited, context=context)

behave_framework/src/minifi_test_framework/steps/core_steps.py

@@ -20,6 +20,7 @@
 import string
 import os
 import time
+import uuid
 
 import humanfriendly
 from behave import when, step, given
@@ -58,6 +59,17 @@ def step_impl(context: MinifiTestContext, file_name: str, content: str, path: st
     context.get_or_create_default_minifi_container().files.append(File(os.path.join(path, file_name), new_content))
 
 
+@step('a file with the content "{content}" is present in "{path}"')
+def step_impl(context: MinifiTestContext, content: str, path: str):
+    new_content = content.replace("\\n", "\n")
+    context.get_or_create_default_minifi_container().files.append(File(os.path.join(path, str(uuid.uuid4())), new_content))
+
+
+@given("an empty file is present in \"{path}\"")
+def step_impl(context, path):
+    context.get_or_create_default_minifi_container().files.append(File(os.path.join(path, str(uuid.uuid4())), ""))
+
+
 @given('a host resource file "{filename}" is bound to the "{container_path}" path in the MiNiFi container "{container_name}"')
 def step_impl(context: MinifiTestContext, filename: str, container_path: str, container_name: str):
     path = os.path.join(context.resource_dir, filename)
@@ -83,3 +95,8 @@ def step_impl(context: MinifiTestContext):
 @when("MiNiFi is restarted")
 def step_impl(context: MinifiTestContext):
     context.get_or_create_default_minifi_container().restart()
+
+
+@given("OpenSSL FIPS mode is enabled in MiNiFi")
+def step_impl(context):
+    context.get_or_create_default_minifi_container().enable_openssl_fips_mode()

docker/test/integration/cluster/ContainerStore.py

@@ -31,13 +31,11 @@
 from .containers.SyslogUdpClientContainer import SyslogUdpClientContainer
 from .containers.SyslogTcpClientContainer import SyslogTcpClientContainer
 from .containers.MinifiAsPodInKubernetesCluster import MinifiAsPodInKubernetesCluster
-from .containers.TcpClientContainer import TcpClientContainer
 from .containers.PrometheusContainer import PrometheusContainer
 from .containers.MinifiC2ServerContainer import MinifiC2ServerContainer
 from .containers.GrafanaLokiContainer import GrafanaLokiContainer
 from .containers.GrafanaLokiContainer import GrafanaLokiOptions
 from .containers.ReverseProxyContainer import ReverseProxyContainer
-from .containers.DiagSlave import DiagSlave
 from .containers.CouchbaseServerContainer import CouchbaseServerContainer
 from .FeatureContext import FeatureContext
 
@@ -216,14 +214,6 @@ def acquire_container(self, context, container_name: str, engine='minifi-cpp', c
                                                   network=self.network,
                                                   image_store=self.image_store,
                                                   command=command))
-        elif engine == "tcp-client":
-            return self.containers.setdefault(container_name,
-                                              TcpClientContainer(feature_context=feature_context,
-                                                                 name=container_name,
-                                                                 vols=self.vols,
-                                                                 network=self.network,
-                                                                 image_store=self.image_store,
-                                                                 command=command))
         elif engine == "prometheus":
             return self.containers.setdefault(container_name,
                                               PrometheusContainer(feature_context=feature_context,
@@ -276,14 +266,6 @@ def acquire_container(self, context, container_name: str, engine='minifi-cpp', c
                                                                     network=self.network,
                                                                     image_store=self.image_store,
                                                                     command=command))
-        elif engine == "diag-slave-tcp":
-            return self.containers.setdefault(container_name,
-                                              DiagSlave(feature_context=feature_context,
-                                                        name=container_name,
-                                                        vols=self.vols,
-                                                        network=self.network,
-                                                        image_store=self.image_store,
-                                                        command=command))
         elif engine == "couchbase-server":
             return self.containers.setdefault(container_name,
                                               CouchbaseServerContainer(feature_context=feature_context,

docker/test/integration/cluster/ImageStore.py

@@ -71,8 +71,6 @@ def get_image(self, container_engine):
             image = self.__build_kinesis_image()
         elif container_engine == "reverse-proxy":
             image = self.__build_reverse_proxy_image()
-        elif container_engine == "diag-slave-tcp":
-            image = self.__build_diagslave_image()
         else:
             raise Exception("There is no associated image for " + container_engine)
 
@@ -303,9 +301,6 @@ def __build_splunk_image(self):
     def __build_reverse_proxy_image(self):
         return self.__build_image_by_path(self.test_dir + "/resources/reverse-proxy", 'reverse-proxy')
 
-    def __build_diagslave_image(self):
-        return self.__build_image_by_path(self.test_dir + "/resources/diagslave", 'diag-slave-tcp')
-
     def __build_image(self, dockerfile, context_files=[]):
         conf_dockerfile_buffer = BytesIO()
         docker_context_buffer = BytesIO()

docker/test/integration/cluster/containers/DiagSlave.py

@@ -1297,18 +1297,6 @@ def step_impl(context, parameter_context_name):
     container.set_parameter_context_name(parameter_context_name)
 
 
-# Modbus
-@given(u'there is an accessible PLC with modbus enabled')
-def step_impl(context):
-    context.test.acquire_container(context=context, name="diag-slave-tcp", engine="diag-slave-tcp")
-    context.test.start('diag-slave-tcp')
-
-
-@given(u'PLC register has been set with {modbus_cmd} command')
-def step_impl(context, modbus_cmd):
-    context.test.set_value_on_plc_with_modbus(context.test.get_container_name_with_postfix('diag-slave-tcp'), modbus_cmd)
-
-
 # Couchbase
 @when(u'a Couchbase server is started')
 def step_impl(context):