MINIFICPP-2679 Move HTTP tests to modular docker tests

Author: lordgamez

behave_framework/src/minifi_test_framework/containers/container.py

@@ -30,7 +30,7 @@
 
 
 class Container:
-    def __init__(self, image_name: str, container_name: str, network: Network, command: str | None = None):
+    def __init__(self, image_name: str, container_name: str, network: Network, command: str | None = None, entrypoint: str | None = None):
         self.image_name: str = image_name
         self.container_name: str = container_name
         self.network: Network = network
@@ -42,6 +42,7 @@ def __init__(self, image_name: str, container_name: str, network: Network, comma
         self.host_files: list[HostFile] = []
         self.volumes = {}
         self.command: str | None = command
+        self.entrypoint: str | None = entrypoint
         self._temp_dir: tempfile.TemporaryDirectory | None = None
         self.ports: dict[str, int] | None = None
         self.environment: list[str] = []
@@ -91,7 +92,7 @@ def deploy(self) -> bool:
             self.container = self.client.containers.run(
                 image=self.image_name, name=self.container_name, ports=self.ports,
                 environment=self.environment, volumes=self.volumes, network=self.network.name,
-                command=self.command, user=self.user, detach=True)
+                command=self.command, entrypoint=self.entrypoint, user=self.user, detach=True)
         except Exception as e:
             logging.error(f"Error starting container: {e}")
             raise
@@ -418,3 +419,25 @@ def directory_contains_file_with_json_content(self, directory_path: str, expecte
                 continue
 
         return False
+
+    def directory_contains_file_with_minimum_size(self, directory_path: str, expected_size: int) -> 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 -exec stat -c %s {{}} \\;"))
+
+        exit_code, output = self.exec_run(command)
+        if exit_code != 0:
+            logging.error(f"Error running command to get file sizes: {output}")
+            return False
+        sizes = output.strip().split('\n')
+        for size_str in sizes:
+            try:
+                size = int(size_str)
+                if size >= expected_size:
+                    return True
+            except ValueError:
+                logging.error(f"Error parsing size '{size_str}' as integer for file size comparison.")
+                continue
+
+        return False

behave_framework/src/minifi_test_framework/containers/minifi_container.py

@@ -20,20 +20,21 @@
 
 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, make_client_cert
+from minifi_test_framework.minifi.minifi_flow_definition import MinifiFlowDefinition
+from minifi_test_framework.core.ssl_utils import make_cert_without_extended_usage, make_client_cert, make_server_cert
 from .container import Container
 
 
 class MinifiContainer(Container):
     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.flow_definition = MinifiFlowDefinition()
         self.properties: dict[str, str] = {}
         self.log_properties: dict[str, str] = {}
 
         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("/usr/local/share/certs/ca-root-nss.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=test_context.root_ca_cert)))
         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)))
@@ -42,6 +43,10 @@ def __init__(self, container_name: str, test_context: MinifiTestContext):
         self.files.append(File("/tmp/resources/clientuser.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=clientuser_cert)))
         self.files.append(File("/tmp/resources/clientuser.key", crypto.dump_privatekey(type=crypto.FILETYPE_PEM, pkey=clientuser_key)))
 
+        minifi_server_cert, minifi_server_key = make_server_cert(common_name=f"server-{test_context.scenario_id}", ca_cert=test_context.root_ca_cert, ca_key=test_context.root_ca_key)
+        self.files.append(File("/tmp/resources/minifi_server.crt",
+                               crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=minifi_server_cert) + crypto.dump_privatekey(type=crypto.FILETYPE_PEM, pkey=minifi_server_key)))
+
         self.is_fhs = 'MINIFI_INSTALLATION_TYPE=FHS' in str(self.client.images.get(test_context.minifi_container_image).history())
 
         self._fill_default_properties()

behave_framework/src/minifi_test_framework/containers/nifi_container.py

@@ -0,0 +1,89 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import gzip
+from typing import List, Optional
+from minifi_test_framework.containers.file import File
+from minifi_test_framework.containers.container import Container
+from minifi_test_framework.core.helpers import wait_for_condition
+from minifi_test_framework.core.minifi_test_context import MinifiTestContext
+from minifi_test_framework.minifi.nifi_flow_definition import NifiFlowDefinition
+
+
+class NifiContainer(Container):
+    NIFI_VERSION = '2.2.0'
+
+    def __init__(self, test_context: MinifiTestContext, command: Optional[List[str]] = None, use_ssl: bool = False):
+        self.flow_definition = NifiFlowDefinition()
+        name = f"nifi-{test_context.scenario_id}"
+        if use_ssl:
+            entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.remote.input.secure\)=.*/\1=true/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.sensitive.props.key\)=.*/\1=secret_key_12345/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.https.port\)=.*/\1=8443/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.https.host\)=.*/\1={name}/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystore\)=.*/\1=\/tmp\/resources\/keystore.jks/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystoreType\)=.*/\1=jks/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystorePasswd\)=.*/\1=passw0rd1!/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keyPasswd\)=.*/#\1=passw0rd1!/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststore\)=.*/\1=\/tmp\/resources\/truststore.jks/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststoreType\)=.*/\1=jks/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststorePasswd\)=.*/\1=passw0rd1!/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\1=10443/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"cp /tmp/nifi_config/flow.json.gz /opt/nifi/nifi-current/conf && /opt/nifi/nifi-current/bin/nifi.sh run & "
+                             r"nifi_pid=$! &&"
+                             r"tail -F --pid=${{nifi_pid}} /opt/nifi/nifi-current/logs/nifi-app.log").format(name=name)
+        else:
+            entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.sensitive.props.key\)=.*/\1=secret_key_12345/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.remote.input.secure\)=.*/\1=false/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.http.port\)=.*/\1=8080/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.https.port\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.https.host\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.web.http.host\)=.*/\1={name}/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystore\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystoreType\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keystorePasswd\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.keyPasswd\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststore\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststoreType\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.security.truststorePasswd\)=.*/\1=/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\1=10000/' /opt/nifi/nifi-current/conf/nifi.properties && "
+                             r"cp /tmp/nifi_config/flow.json.gz /opt/nifi/nifi-current/conf && /opt/nifi/nifi-current/bin/nifi.sh run & "
+                             r"nifi_pid=$! &&"
+                             r"tail -F --pid=${{nifi_pid}} /opt/nifi/nifi-current/logs/nifi-app.log").format(name=name)
+        if not command:
+            command = ["/bin/sh", "-c", entry_command]
+
+        super().__init__("apache/nifi:" + self.NIFI_VERSION, name, test_context.network, entrypoint=command)
+
+    def deploy(self):
+        flow_config = self.flow_definition.to_json()
+        buffer = io.BytesIO()
+
+        with gzip.GzipFile(fileobj=buffer, mode='wb') as gz_file:
+            gz_file.write(flow_config.encode())
+
+        gzipped_bytes = buffer.getvalue()
+        self.files.append(File("/tmp/nifi_config/flow.json.gz", gzipped_bytes))
+
+        super().deploy()
+        finished_str = "Started Application in"
+        return wait_for_condition(
+            condition=lambda: finished_str in self.get_logs(),
+            timeout_seconds=300,
+            bail_condition=lambda: self.exited,
+            context=None)

behave_framework/src/minifi_test_framework/minifi/flow_definition.py

@@ -15,7 +15,7 @@
 #  limitations under the License.
 #
 
-import yaml
+from abc import ABC
 
 from .connection import Connection
 from .controller_service import ControllerService
@@ -24,7 +24,7 @@
 from .processor import Processor
 
 
-class FlowDefinition:
+class FlowDefinition(ABC):
     def __init__(self, flow_name: str = "MiNiFi Flow"):
         self.flow_name = flow_name
         self.processors: list[Processor] = []
@@ -53,42 +53,10 @@ def add_connection(self, connection: Connection):
         self.connections.append(connection)
 
     def to_yaml(self) -> str:
-        """Serializes the entire flow definition into the MiNiFi YAML format."""
+        raise NotImplementedError("to_yaml method must be implemented in subclasses")
 
-        # Create a quick lookup map of processor names to their objects
-        # This is crucial for finding the source/destination IDs for connections
-        processors_by_name = {p.name: p for p in self.processors}
-        funnels_by_name = {f.name: f for f in self.funnels}
-
-        connectables_by_name = {**processors_by_name, **funnels_by_name}
-
-        if len(self.parameter_contexts) > 0:
-            parameter_context_name = self.parameter_contexts[0].name
-        else:
-            parameter_context_name = ''
-        # Build the final dictionary structure
-        config = {'MiNiFi Config Version': 3, 'Flow Controller': {'name': self.flow_name},
-                  'Parameter Contexts': [p.to_yaml_dict() for p in self.parameter_contexts],
-                  'Processors': [p.to_yaml_dict() for p in self.processors],
-                  'Funnels': [f.to_yaml_dict() for f in self.funnels], 'Connections': [],
-                  'Controller Services': [c.to_yaml_dict() for c in self.controller_services],
-                  'Remote Processing Groups': [], 'Parameter Context Name': parameter_context_name}
-
-        # Build the connections list by looking up processor IDs
-        for conn in self.connections:
-            source_proc = connectables_by_name.get(conn.source_name)
-            dest_proc = connectables_by_name.get(conn.target_name)
-
-            if not source_proc or not dest_proc:
-                raise ValueError(
-                    f"Could not find processors for connection from '{conn.source_name}' to '{conn.target_name}'")
-
-            config['Connections'].append(
-                {'name': f"{conn.source_name}/{conn.source_relationship}/{conn.target_name}", 'id': conn.id,
-                 'source id': source_proc.id, 'source relationship name': conn.source_relationship,
-                 'destination id': dest_proc.id})
-
-        return yaml.dump(config, sort_keys=False, indent=2, width=120)
+    def to_json(self) -> str:
+        raise NotImplementedError("to_json method must be implemented in subclasses")
 
     def __repr__(self):
         return f"FlowDefinition(Processors: {self.processors}, Controller Services: {self.controller_services})"

behave_framework/src/minifi_test_framework/minifi/minifi_flow_definition.py

@@ -0,0 +1,63 @@
+#
+#  Licensed to the Apache Software Foundation (ASF) under one or more
+#  contributor license agreements.  See the NOTICE file distributed with
+#  this work for additional information regarding copyright ownership.
+#  The ASF licenses this file to You under the Apache License, Version 2.0
+#  (the "License"); you may not use this file except in compliance with
+#  the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+import yaml
+
+from .flow_definition import FlowDefinition
+
+
+class MinifiFlowDefinition(FlowDefinition):
+    def __init__(self, flow_name: str = "MiNiFi Flow"):
+        super().__init__(flow_name)
+
+    def to_yaml(self) -> str:
+        """Serializes the entire flow definition into the MiNiFi YAML format."""
+
+        # Create a quick lookup map of processor names to their objects
+        # This is crucial for finding the source/destination IDs for connections
+        processors_by_name = {p.name: p for p in self.processors}
+        funnels_by_name = {f.name: f for f in self.funnels}
+
+        connectables_by_name = {**processors_by_name, **funnels_by_name}
+
+        if len(self.parameter_contexts) > 0:
+            parameter_context_name = self.parameter_contexts[0].name
+        else:
+            parameter_context_name = ''
+        # Build the final dictionary structure
+        config = {'MiNiFi Config Version': 3, 'Flow Controller': {'name': self.flow_name},
+                  'Parameter Contexts': [p.to_yaml_dict() for p in self.parameter_contexts],
+                  'Processors': [p.to_yaml_dict() for p in self.processors],
+                  'Funnels': [f.to_yaml_dict() for f in self.funnels], 'Connections': [],
+                  'Controller Services': [c.to_yaml_dict() for c in self.controller_services],
+                  'Remote Processing Groups': [], 'Parameter Context Name': parameter_context_name}
+
+        # Build the connections list by looking up processor IDs
+        for conn in self.connections:
+            source_proc = connectables_by_name.get(conn.source_name)
+            dest_proc = connectables_by_name.get(conn.target_name)
+
+            if not source_proc or not dest_proc:
+                raise ValueError(
+                    f"Could not find processors for connection from '{conn.source_name}' to '{conn.target_name}'")
+
+            config['Connections'].append(
+                {'name': f"{conn.source_name}/{conn.source_relationship}/{conn.target_name}", 'id': conn.id,
+                 'source id': source_proc.id, 'source relationship name': conn.source_relationship,
+                 'destination id': dest_proc.id})
+
+        return yaml.dump(config, sort_keys=False, indent=2, width=120)