You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by lo...@apache.org on 2026/02/24 18:57:35 UTC
(nifi-minifi-cpp) branch main updated (2ac1f1b86 -> 3190ca615)
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 2ac1f1b86 MINIFICPP-2721 Fix rocky linux docker image name retrieval in CI
new d24d32400 MINIFICPP-2683 Move C2 docker tests to modular docker tests
new e730451c9 MINIFICPP-2685 Move MQTT tests to modular docker tests
new ebc89c186 MINIFICPP-2256 Remove LoggerConfiguration::newInstance
new 3b1334649 MINIFICPP-2722 C API enum explicit type
new 3190ca615 MINIFICPP-2724 Add MinifiStatus return type to C api function that might fail
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../minifi_test_framework/containers/container.py | 25 ++
.../containers/minifi_container.py | 10 +
.../minifi_test_framework/steps/checking_steps.py | 26 +-
.../src/minifi_test_framework/steps/core_steps.py | 57 ++-
.../steps/flow_building_steps.py | 10 +-
docker/RunBehaveTests.sh | 3 +-
docker/requirements.txt | 1 -
docker/test/integration/cluster/ContainerStore.py | 37 --
.../test/integration/cluster/DockerTestCluster.py | 5 -
docker/test/integration/cluster/ImageStore.py | 11 -
.../cluster/containers/MinifiC2ServerContainer.py | 112 ------
.../cluster/containers/MinifiContainer.py | 31 +-
.../cluster/containers/MqttBrokerContainer.py | 40 ---
.../features/MiNiFi_integration_test_driver.py | 8 -
docker/test/integration/features/steps/steps.py | 59 ---
.../integration/minifi/processors/ConsumeMQTT.py | 30 --
.../integration/minifi/processors/PublishMQTT.py | 30 --
.../include/api/core/ProcessContext.h | 2 -
.../include/api/utils/ProcessorConfigUtils.h | 24 +-
.../cpp-extension-lib/src/core/ProcessContext.cpp | 10 -
.../cpp-extension-lib/src/core/ProcessSession.cpp | 22 +-
.../include/utils/ProcessorConfigUtils.h | 24 +-
.../azure/tests/FetchAzureDataLakeStorageTests.cpp | 2 +-
.../features/steps/couchbase_server_container.py | 4 +-
.../tests/features/steps/grafana_loki_container.py | 4 +-
.../llamacpp/tests/RunLlamaCppInferenceTests.cpp | 4 +-
extensions/mqtt/tests/ConsumeMQTTTests.cpp | 4 +-
extensions/mqtt/tests/PublishMQTTTests.cpp | 2 +-
.../tests/features/environment.py | 21 +-
.../mqtt/tests}/features/mqtt.feature | 398 ++++++++++++---------
.../features/resources/publish_mqtt_message.py | 20 +-
.../tests/features/steps/mqtt_broker_container.py | 63 ++++
extensions/mqtt/tests/features/steps/steps.py | 76 ++++
.../tests}/features/minifi_c2_server.feature | 15 +-
.../features/steps/minifi_c2_server_container.py | 76 ++++
.../resources/minifi-c2-server/authorizations.yaml | 0
.../resources/minifi-c2-server/c2.properties | 0
.../steps}/resources/minifi-c2-server/config.yml | 0
.../tests/features/steps/steps.py | 54 ++-
.../tests/unit/SegmentContentTests.cpp | 4 +-
.../tests/unit/SplitContentTests.cpp | 2 +-
.../tests/unit/SplitTextTests.cpp | 2 +-
.../include/core/logging/LoggerConfiguration.h | 9 +-
libminifi/src/minifi-c.cpp | 31 +-
libminifi/test/libtest/unit/TestBase.cpp | 7 +-
libminifi/test/unit/LoggerTests.cpp | 84 ++---
minifi-api/include/minifi-c/minifi-c.h | 15 +-
minifi-api/minifi-c-api.def | 1 -
48 files changed, 768 insertions(+), 707 deletions(-)
delete mode 100644 docker/test/integration/cluster/containers/MinifiC2ServerContainer.py
delete mode 100644 docker/test/integration/cluster/containers/MqttBrokerContainer.py
delete mode 100644 docker/test/integration/minifi/processors/ConsumeMQTT.py
delete mode 100644 docker/test/integration/minifi/processors/PublishMQTT.py
copy extensions/{grafana-loki => mqtt}/tests/features/environment.py (71%)
rename {docker/test/integration => extensions/mqtt/tests}/features/mqtt.feature (60%)
rename docker/test/integration/cluster/checkers/MqttHelper.py => extensions/mqtt/tests/features/resources/publish_mqtt_message.py (62%)
create mode 100644 extensions/mqtt/tests/features/steps/mqtt_broker_container.py
create mode 100644 extensions/mqtt/tests/features/steps/steps.py
rename {docker/test/integration => extensions/standard-processors/tests}/features/minifi_c2_server.feature (82%)
create mode 100644 extensions/standard-processors/tests/features/steps/minifi_c2_server_container.py
rename {docker/test/integration => extensions/standard-processors/tests/features/steps}/resources/minifi-c2-server/authorizations.yaml (100%)
rename {docker/test/integration => extensions/standard-processors/tests/features/steps}/resources/minifi-c2-server/c2.properties (100%)
rename {docker/test/integration => extensions/standard-processors/tests/features/steps}/resources/minifi-c2-server/config.yml (100%)
(nifi-minifi-cpp) 01/05: MINIFICPP-2683 Move C2 docker tests to modular docker tests
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit d24d3240030f25655710bd056b3e9c38e12c2299
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Jan 20 14:13:54 2026 +0100
MINIFICPP-2683 Move C2 docker tests to modular docker tests
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #2089
---
.../containers/minifi_container.py | 10 ++
.../minifi_test_framework/steps/checking_steps.py | 12 +--
.../src/minifi_test_framework/steps/core_steps.py | 11 +-
docker/test/integration/cluster/ContainerStore.py | 28 ------
.../cluster/containers/MinifiC2ServerContainer.py | 112 ---------------------
.../cluster/containers/MinifiContainer.py | 31 +-----
.../features/MiNiFi_integration_test_driver.py | 5 -
docker/test/integration/features/steps/steps.py | 32 ------
.../tests}/features/minifi_c2_server.feature | 15 ++-
.../features/steps/minifi_c2_server_container.py | 76 ++++++++++++++
.../resources/minifi-c2-server/authorizations.yaml | 0
.../resources/minifi-c2-server/c2.properties | 0
.../steps}/resources/minifi-c2-server/config.yml | 0
.../tests/features/steps/steps.py | 54 +++++++++-
14 files changed, 161 insertions(+), 225 deletions(-)
diff --git a/behave_framework/src/minifi_test_framework/containers/minifi_container.py b/behave_framework/src/minifi_test_framework/containers/minifi_container.py
index be72155e9..8d2b5f992 100644
--- a/behave_framework/src/minifi_test_framework/containers/minifi_container.py
+++ b/behave_framework/src/minifi_test_framework/containers/minifi_container.py
@@ -35,6 +35,7 @@ class MinifiContainer(Container):
self.flow_definition = MinifiFlowDefinition()
self.properties: dict[str, str] = {}
self.log_properties: dict[str, str] = {}
+ self.scenario_id = test_context.scenario_id
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)))
@@ -96,6 +97,15 @@ class MinifiContainer(Container):
def enable_openssl_fips_mode(self):
self.properties["nifi.openssl.fips.support.enable"] = "true"
+ def fetch_flow_config_from_flow_url(self):
+ self.properties["nifi.c2.flow.url"] = f"http://minifi-c2-server-{self.scenario_id}:10090/c2/config?class=minifi-test-class"
+
+ def set_up_ssl_properties(self):
+ self.properties["nifi.remote.input.secure"] = "true"
+ self.properties["nifi.security.client.certificate"] = "/tmp/resources/minifi_client.crt"
+ self.properties["nifi.security.client.private.key"] = "/tmp/resources/minifi_client.key"
+ self.properties["nifi.security.client.ca.certificate"] = "/tmp/resources/root_ca.crt"
+
def _fill_default_properties(self):
if self.is_fhs:
self.properties["nifi.flow.configuration.file"] = "/etc/nifi-minifi-cpp/config.yml"
diff --git a/behave_framework/src/minifi_test_framework/steps/checking_steps.py b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
index 8a2eafae1..ef5b1a338 100644
--- a/behave_framework/src/minifi_test_framework/steps/checking_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
@@ -196,7 +196,7 @@ def step_impl(context: MinifiTestContext, content: str, directory: str, duration
@then('after a wait of {duration}, at least {lower_bound:d} and at most {upper_bound:d} files are produced and placed in the "{directory}" directory')
-def step_impl(context, lower_bound, upper_bound, duration, directory):
+def step_impl(context: MinifiTestContext, lower_bound: int, upper_bound: int, duration: str, directory: str):
duration_seconds = humanfriendly.parse_timespan(duration)
assert check_condition_after_wait(condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].get_number_of_files(directory) >= lower_bound
and context.containers[DEFAULT_MINIFI_CONTAINER_NAME].get_number_of_files(directory) <= upper_bound,
@@ -204,7 +204,7 @@ def step_impl(context, lower_bound, upper_bound, duration, directory):
@then('exactly these files are in the "{directory}" directory in less than {duration}: "{contents}"')
-def step_impl(context, directory, duration, contents):
+def step_impl(context: MinifiTestContext, directory: str, duration: str, contents: str):
if not contents:
context.execute_steps(f'then no files are placed in the "{directory}" directory in {duration} of running time')
return
@@ -221,7 +221,7 @@ def step_impl(context, directory, duration):
@then("at least one empty file is placed in the \"{directory}\" directory in less than {duration}")
-def step_impl(context, directory, duration):
+def step_impl(context: MinifiTestContext, directory: str, duration: str):
timeout_in_seconds = humanfriendly.parse_timespan(duration)
assert wait_for_condition(
condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].directory_contains_empty_file(directory),
@@ -229,7 +229,7 @@ def step_impl(context, directory, duration):
@then("in the \"{container_name}\" container at least one empty file is placed in the \"{directory}\" directory in less than {duration}")
-def step_impl(context, container_name, directory, duration):
+def step_impl(context: MinifiTestContext, container_name: str, directory: str, duration: str):
timeout_in_seconds = humanfriendly.parse_timespan(duration)
assert wait_for_condition(
condition=lambda: context.containers[container_name].directory_contains_empty_file(directory),
@@ -237,7 +237,7 @@ def step_impl(context, container_name, directory, duration):
@then("in the \"{container_name}\" container at least one file with minimum size of \"{size}\" is placed in the \"{directory}\" directory in less than {duration}")
-def step_impl(context, container_name: str, directory: str, size: str, duration: str):
+def step_impl(context: MinifiTestContext, container_name: str, directory: str, size: str, duration: str):
timeout_in_seconds = humanfriendly.parse_timespan(duration)
size_in_bytes = humanfriendly.parse_size(size)
assert wait_for_condition(
@@ -246,6 +246,6 @@ def step_impl(context, container_name: str, directory: str, size: str, duration:
@then("at least one file with minimum size of \"{size}\" is placed in the \"{directory}\" directory in less than {duration}")
-def step_impl(context, directory: str, size: str, duration: str):
+def step_impl(context: MinifiTestContext, directory: str, size: str, duration: str):
context.execute_steps(
f'Then in the "{DEFAULT_MINIFI_CONTAINER_NAME}" container at least one file with minimum size of "{size}" is placed in the "{directory}" directory in less than {duration}')
diff --git a/behave_framework/src/minifi_test_framework/steps/core_steps.py b/behave_framework/src/minifi_test_framework/steps/core_steps.py
index 836b58834..6b488587c 100644
--- a/behave_framework/src/minifi_test_framework/steps/core_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/core_steps.py
@@ -100,15 +100,20 @@ def step_impl(context: MinifiTestContext):
@given("OpenSSL FIPS mode is enabled in MiNiFi")
-def step_impl(context):
+def step_impl(context: MinifiTestContext):
context.get_or_create_default_minifi_container().enable_openssl_fips_mode()
@step("the http proxy server is set up")
-def step_impl(context):
+def step_impl(context: MinifiTestContext):
context.containers["http-proxy"] = HttpProxy(context)
@step("a NiFi container is set up")
-def step_impl(context):
+def step_impl(context: MinifiTestContext):
context.containers["nifi"] = NifiContainer(context)
+
+
+@given("flow configuration path is set up in flow url property")
+def step_impl(context: MinifiTestContext):
+ context.get_or_create_default_minifi_container().fetch_flow_config_from_flow_url()
diff --git a/docker/test/integration/cluster/ContainerStore.py b/docker/test/integration/cluster/ContainerStore.py
index 084bbc45d..f139acd21 100644
--- a/docker/test/integration/cluster/ContainerStore.py
+++ b/docker/test/integration/cluster/ContainerStore.py
@@ -26,7 +26,6 @@ from .containers.SyslogUdpClientContainer import SyslogUdpClientContainer
from .containers.SyslogTcpClientContainer import SyslogTcpClientContainer
from .containers.MinifiAsPodInKubernetesCluster import MinifiAsPodInKubernetesCluster
from .containers.PrometheusContainer import PrometheusContainer
-from .containers.MinifiC2ServerContainer import MinifiC2ServerContainer
from .FeatureContext import FeatureContext
@@ -172,24 +171,6 @@ class ContainerStore:
image_store=self.image_store,
command=command,
ssl=True))
- elif engine == "minifi-c2-server":
- return self.containers.setdefault(container_name,
- MinifiC2ServerContainer(feature_context=feature_context,
- name=container_name,
- vols=self.vols,
- network=self.network,
- image_store=self.image_store,
- command=command,
- ssl=False))
- elif engine == "minifi-c2-server-ssl":
- return self.containers.setdefault(container_name,
- MinifiC2ServerContainer(feature_context=feature_context,
- name=container_name,
- vols=self.vols,
- network=self.network,
- image_store=self.image_store,
- command=command,
- ssl=True))
else:
raise Exception('invalid flow engine: \'%s\'' % engine)
@@ -231,15 +212,6 @@ class ContainerStore:
def enable_provenance_repository_in_minifi(self):
self.minifi_options.enable_provenance = True
- def enable_c2_in_minifi(self):
- self.minifi_options.enable_c2 = True
-
- def enable_c2_with_ssl_in_minifi(self):
- self.minifi_options.enable_c2_with_ssl = True
-
- def fetch_flow_config_from_c2_url_in_minifi(self):
- self.minifi_options.use_flow_config_from_url = True
-
def set_ssl_context_properties_in_minifi(self):
self.minifi_options.set_ssl_context_properties = True
diff --git a/docker/test/integration/cluster/containers/MinifiC2ServerContainer.py b/docker/test/integration/cluster/containers/MinifiC2ServerContainer.py
deleted file mode 100644
index 9be1cdeb1..000000000
--- a/docker/test/integration/cluster/containers/MinifiC2ServerContainer.py
+++ /dev/null
@@ -1,112 +0,0 @@
-# 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 os
-import tempfile
-
-import docker.types
-import jks
-
-from .Container import Container
-from OpenSSL import crypto
-from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption, load_pem_private_key
-from cryptography import x509
-from ssl_utils.SSL_cert_utils import make_server_cert
-
-
-class MinifiC2ServerContainer(Container):
- def __init__(self, feature_context, name, vols, network, image_store, command=None, ssl=False):
- engine = "minifi-c2-server-ssl" if ssl else "minifi-c2-server"
- super().__init__(feature_context, name, engine, vols, network, image_store, command)
- self.ssl = ssl
- if ssl:
- c2_cert, c2_key = make_server_cert(f"minifi-c2-server-{feature_context.id}", feature_context.root_ca_cert, feature_context.root_ca_key)
- pke = jks.PrivateKeyEntry.new('c2-server-cert', [crypto.dump_certificate(crypto.FILETYPE_ASN1, c2_cert)], crypto.dump_privatekey(crypto.FILETYPE_ASN1, c2_key), 'rsa_raw')
- server_keystore = jks.KeyStore.new('jks', [pke])
- self.server_keystore_file = tempfile.NamedTemporaryFile(delete=False)
- server_keystore.save(self.server_keystore_file.name, 'abcdefgh')
- os.chmod(self.server_keystore_file.name, 0o644)
-
- self.server_truststore_file = tempfile.NamedTemporaryFile(delete=False)
- private_key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, feature_context.root_ca_key)
- private_key = load_pem_private_key(private_key_pem, password=None)
- certificate_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, feature_context.root_ca_cert)
- certificate = x509.load_pem_x509_certificate(certificate_pem)
- pkcs12_data = pkcs12.serialize_key_and_certificates(
- name=None,
- key=private_key,
- cert=certificate,
- cas=None,
- encryption_algorithm=BestAvailableEncryption(b'abcdefgh')
- )
- self.server_truststore_file.write(pkcs12_data)
- self.server_truststore_file.close()
- os.chmod(self.server_truststore_file.name, 0o644)
-
- self.authorities_file = tempfile.NamedTemporaryFile(delete=False)
- authorities_file_content = """
-CN=minifi-cpp-flow-{feature_id}:
- - CLASS_MINIFI_CPP
-""".format(feature_id=self.feature_context.id)
- self.authorities_file.write(authorities_file_content.encode())
- self.authorities_file.close()
- os.chmod(self.authorities_file.name, 0o644)
-
- def get_startup_finished_log_entry(self):
- return "Server Started"
-
- def deploy(self):
- if not self.set_deployed():
- return
-
- test_dir = os.environ['TEST_DIRECTORY'] # Based on DockerVerify.sh
- mounts = [docker.types.Mount(
- type='bind',
- source=os.path.join(test_dir, "resources/minifi-c2-server/config.yml"),
- target='/opt/minifi-c2/minifi-c2-current/files/minifi-test-class/config.text.yml.v1'
- )]
-
- if self.ssl:
- mounts.append(docker.types.Mount(
- type='bind',
- source=self.authorities_file.name,
- target='/opt/minifi-c2/minifi-c2-current/conf/authorities.yaml'
- ))
- mounts.append(docker.types.Mount(
- type='bind',
- source=os.path.join(test_dir, "resources/minifi-c2-server/authorizations.yaml"),
- target='/opt/minifi-c2/minifi-c2-current/conf/authorizations.yaml'
- ))
- mounts.append(docker.types.Mount(
- type='bind',
- source=os.path.join(test_dir, "resources/minifi-c2-server/c2.properties"),
- target='/opt/minifi-c2/minifi-c2-current/conf/c2.properties'
- ))
- mounts.append(docker.types.Mount(
- type='bind',
- source=self.server_keystore_file.name,
- target='/opt/minifi-c2/minifi-c2-current/certs/minifi-c2-server-keystore.jks'
- ))
- mounts.append(docker.types.Mount(
- type='bind',
- source=self.server_truststore_file.name,
- target='/opt/minifi-c2/minifi-c2-current/certs/minifi-c2-server-truststore.p12'
- ))
- self.docker_container = self.client.containers.run(
- image='apache/nifi-minifi-c2:1.27.0',
- detach=True,
- name=self.name,
- network=self.network.name,
- mounts=mounts,
- entrypoint=self.command)
diff --git a/docker/test/integration/cluster/containers/MinifiContainer.py b/docker/test/integration/cluster/containers/MinifiContainer.py
index 15e88021a..fcb73c0a4 100644
--- a/docker/test/integration/cluster/containers/MinifiContainer.py
+++ b/docker/test/integration/cluster/containers/MinifiContainer.py
@@ -27,8 +27,6 @@ from minifi.flow_serialization.Minifi_flow_json_serializer import Minifi_flow_js
class MinifiOptions:
def __init__(self):
- self.enable_c2 = False
- self.enable_c2_with_ssl = False
self.enable_provenance = False
self.enable_prometheus = False
self.enable_prometheus_with_ssl = False
@@ -39,7 +37,6 @@ class MinifiOptions:
self.remove_python_requirements_txt = False
self.use_nifi_python_processors_without_dependencies = False
self.config_format = "json"
- self.use_flow_config_from_url = False
self.set_ssl_context_properties = False
self.enable_log_metrics_publisher = False
self.enable_example_minifi_python_processors = False
@@ -133,25 +130,6 @@ class MinifiContainer(FlowContainer):
f.write("nifi.database.content.repository.directory.default={minifi_home}/content_repository\n".format(minifi_home=MinifiContainer.MINIFI_LOCATIONS.minifi_home))
f.write("nifi.python.processor.dir={minifi_home}/minifi-python\n".format(minifi_home=MinifiContainer.MINIFI_LOCATIONS.minifi_home))
- if self.options.enable_c2:
- f.write("nifi.c2.enable=true\n")
- f.write(f"nifi.c2.rest.url=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/heartbeat\n")
- f.write(f"nifi.c2.rest.url.ack=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/acknowledge\n")
- f.write(f"nifi.c2.flow.base.url=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/\n")
- f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation\n")
- f.write("nifi.c2.full.heartbeat=false\n")
- f.write("nifi.c2.agent.class=minifi-test-class\n")
- f.write("nifi.c2.agent.identifier=minifi-test-id\n")
- elif self.options.enable_c2_with_ssl:
- f.write("nifi.c2.enable=true\n")
- f.write(f"nifi.c2.rest.url=https://minifi-c2-server-{self.feature_context.id}:10090/c2/config/heartbeat\n")
- f.write(f"nifi.c2.rest.url.ack=https://minifi-c2-server-{self.feature_context.id}:10090/c2/config/acknowledge\n")
- f.write(f"nifi.c2.flow.base.url=https://minifi-c2-server-{self.feature_context.id}:10090/c2/config/\n")
- f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation\n")
- f.write("nifi.c2.full.heartbeat=false\n")
- f.write("nifi.c2.agent.class=minifi-test-class\n")
- f.write("nifi.c2.agent.identifier=minifi-test-id\n")
-
if self.options.set_ssl_context_properties:
f.write("nifi.remote.input.secure=true\n")
f.write("nifi.security.client.certificate=/tmp/resources/minifi_client.crt\n")
@@ -180,9 +158,6 @@ class MinifiContainer(FlowContainer):
if metrics_publisher_classes:
f.write("nifi.metrics.publisher.class=" + ",".join(metrics_publisher_classes) + "\n")
- if self.options.use_flow_config_from_url:
- f.write(f"nifi.c2.flow.url=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config?class=minifi-test-class\n")
-
if self.options.use_nifi_python_processors_with_virtualenv or self.options.remove_python_requirements_txt or self.options.use_nifi_python_processors_without_dependencies:
f.write("nifi.python.virtualenv.directory={minifi_python_venv_parent}/venv\n".format(minifi_python_venv_parent=MinifiContainer.MINIFI_LOCATIONS.minifi_python_venv_parent))
elif self.options.use_nifi_python_processors_with_virtualenv_packages_installed:
@@ -198,10 +173,8 @@ class MinifiContainer(FlowContainer):
def _setup_config(self):
self._create_properties()
- if not self.options.use_flow_config_from_url:
- self._create_config()
- self.vols[os.path.join(self.container_specific_config_dir, 'config.yml')] = {"bind": MinifiContainer.MINIFI_LOCATIONS.config_path, "mode": "rw"}
-
+ self._create_config()
+ self.vols[os.path.join(self.container_specific_config_dir, 'config.yml')] = {"bind": MinifiContainer.MINIFI_LOCATIONS.config_path, "mode": "rw"}
self.vols[os.path.join(self.container_specific_config_dir, 'minifi.properties')] = {"bind": MinifiContainer.MINIFI_LOCATIONS.properties_path, "mode": "rw"}
self.vols[os.path.join(self.container_specific_config_dir, 'minifi-log.properties')] = {"bind": MinifiContainer.MINIFI_LOCATIONS.log_properties_path, "mode": "rw"}
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index 73941ecda..7a36309cd 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -63,11 +63,6 @@ class MiNiFi_integration_test:
def acquire_transient_minifi(self, context, name, engine='minifi-cpp'):
return self.cluster.acquire_transient_minifi(context=context, name=name, engine=engine)
- def start_minifi_c2_server(self, context):
- self.cluster.acquire_container(context=context, name="minifi-c2-server", engine="minifi-c2-server")
- self.cluster.deploy_container('minifi-c2-server')
- assert self.cluster.wait_for_container_startup_to_finish('minifi-c2-server') or self.cluster.log_app_output()
-
def start_nifi(self, context):
self.cluster.acquire_container(context=context, name='nifi', engine='nifi')
self.cluster.deploy_container('nifi')
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 606d5d76d..ce222141b 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -827,43 +827,11 @@ def step_impl(context):
context.test.check_all_prometheus_metric_types_are_defined_once()
-# MiNiFi C2 Server
-@given("ssl properties are set up for MiNiFi C2 server")
-def step_impl(context):
- context.test.enable_c2_with_ssl_in_minifi()
- context.test.set_ssl_context_properties_in_minifi()
-
-
@given("SSL properties are set in MiNiFi")
def step_impl(context):
context.test.set_ssl_context_properties_in_minifi()
-@given(u'a MiNiFi C2 server is set up')
-def step_impl(context):
- context.test.acquire_container(context=context, name="minifi-c2-server", engine="minifi-c2-server")
-
-
-@given(u'a MiNiFi C2 server is started')
-def step_impl(context):
- context.test.start_minifi_c2_server(context)
-
-
-@then("the MiNiFi C2 server logs contain the following message: \"{log_message}\" in less than {duration}")
-def step_impl(context, log_message, duration):
- context.test.check_container_log_contents("minifi-c2-server", log_message, humanfriendly.parse_timespan(duration))
-
-
-@then("the MiNiFi C2 SSL server logs contain the following message: \"{log_message}\" in less than {duration}")
-def step_impl(context, log_message, duration):
- context.test.check_container_log_contents("minifi-c2-server-ssl", log_message, humanfriendly.parse_timespan(duration))
-
-
-@given(u'a MiNiFi C2 server is set up with SSL')
-def step_impl(context):
- context.test.acquire_container(context=context, name="minifi-c2-server", engine="minifi-c2-server-ssl")
-
-
@given(u'flow configuration path is set up in flow url property')
def step_impl(context):
context.test.acquire_container(context=context, name="minifi-cpp-flow", engine="minifi-cpp")
diff --git a/docker/test/integration/features/minifi_c2_server.feature b/extensions/standard-processors/tests/features/minifi_c2_server.feature
similarity index 82%
rename from docker/test/integration/features/minifi_c2_server.feature
rename to extensions/standard-processors/tests/features/minifi_c2_server.feature
index 2bbffc9f4..29d11295a 100644
--- a/docker/test/integration/features/minifi_c2_server.feature
+++ b/extensions/standard-processors/tests/features/minifi_c2_server.feature
@@ -16,9 +16,6 @@
@CORE
Feature: MiNiFi can communicate with Apache NiFi MiNiFi C2 server
- Background:
- Given the content of "/tmp/output" is monitored
-
Scenario: MiNiFi flow config is updated from MiNiFi C2 server
Given a GetFile processor with the name "GetFile1" and the "Input Directory" property set to "/tmp/non-existent"
And C2 is enabled in MiNiFi
@@ -26,7 +23,7 @@ Feature: MiNiFi can communicate with Apache NiFi MiNiFi C2 server
And a MiNiFi C2 server is set up
When all instances start up
Then the MiNiFi C2 server logs contain the following message: "acknowledged with a state of FULLY_APPLIED(DONE)" in less than 30 seconds
- And a flowfile with the content "test" is placed in the monitored directory in less than 10 seconds
+ And a single file with the content "test" is placed in the "/tmp/output" directory in less than 10 seconds
And the Minifi logs do not contain the following message: "Failed to parse json response: The document is empty. at 0" after 0 seconds
Scenario: MiNiFi can get flow config from C2 server through flow url when it is not available at start
@@ -34,9 +31,9 @@ Feature: MiNiFi can communicate with Apache NiFi MiNiFi C2 server
And C2 is enabled in MiNiFi
And a file with the content "test" is present in "/tmp/input"
And a MiNiFi C2 server is started
- When all instances start up
+ When the MiNiFi instance starts up
Then the MiNiFi C2 server logs contain the following message: "acknowledged with a state of FULLY_APPLIED(DONE)" in less than 30 seconds
- And a flowfile with the content "test" is placed in the monitored directory in less than 10 seconds
+ And a single file with the content "test" is placed in the "/tmp/output" directory in less than 10 seconds
Scenario: MiNiFi flow config is updated from MiNiFi C2 server through SSL with SSL properties
Given a file with the content "test" is present in "/tmp/input"
@@ -44,8 +41,8 @@ Feature: MiNiFi can communicate with Apache NiFi MiNiFi C2 server
And ssl properties are set up for MiNiFi C2 server
And a MiNiFi C2 server is set up with SSL
When all instances start up
- Then the MiNiFi C2 SSL server logs contain the following message: "acknowledged with a state of FULLY_APPLIED(DONE)" in less than 60 seconds
- And a flowfile with the content "test" is placed in the monitored directory in less than 10 seconds
+ Then the MiNiFi C2 server logs contain the following message: "acknowledged with a state of FULLY_APPLIED(DONE)" in less than 60 seconds
+ And a single file with the content "test" is placed in the "/tmp/output" directory in less than 10 seconds
Scenario: MiNiFi flow config is updated from MiNiFi C2 server with overriden parameter context
Given a GetFile processor with the name "GetFile1" and the "Input Directory" property set to "${INPUT_DIR}"
@@ -56,5 +53,5 @@ Feature: MiNiFi can communicate with Apache NiFi MiNiFi C2 server
And a MiNiFi C2 server is set up
When all instances start up
Then the MiNiFi C2 server logs contain the following message: "acknowledged with a state of FULLY_APPLIED(DONE)" in less than 30 seconds
- And a flowfile with the content "test" is placed in the monitored directory in less than 10 seconds
+ And a single file with the content "test" is placed in the "/tmp/output" directory in less than 10 seconds
And the Minifi logs do not contain the following message: "Failed to parse json response: The document is empty. at 0" after 0 seconds
diff --git a/extensions/standard-processors/tests/features/steps/minifi_c2_server_container.py b/extensions/standard-processors/tests/features/steps/minifi_c2_server_container.py
new file mode 100644
index 000000000..6c25d7c77
--- /dev/null
+++ b/extensions/standard-processors/tests/features/steps/minifi_c2_server_container.py
@@ -0,0 +1,76 @@
+#
+# 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 jks
+import os
+from OpenSSL import crypto
+from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption, load_pem_private_key
+from cryptography import x509
+from pathlib import Path
+
+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.core.ssl_utils import make_server_cert
+from minifi_test_framework.containers.file import File
+from minifi_test_framework.containers.host_file import HostFile
+
+
+class MinifiC2Server(Container):
+ def __init__(self, test_context: MinifiTestContext, ssl: bool = False):
+ super().__init__("apache/nifi-minifi-c2:1.27.0", f"minifi-c2-server-{test_context.scenario_id}", test_context.network)
+ if ssl:
+ c2_cert, c2_key = make_server_cert(f"minifi-c2-server-{test_context.scenario_id}", test_context.root_ca_cert, test_context.root_ca_key)
+ pke = jks.PrivateKeyEntry.new('c2-server-cert', [crypto.dump_certificate(crypto.FILETYPE_ASN1, c2_cert)], crypto.dump_privatekey(crypto.FILETYPE_ASN1, c2_key), 'rsa_raw')
+ server_keystore = jks.KeyStore.new('jks', [pke])
+ server_keystore_content = server_keystore.saves('abcdefgh')
+ self.files.append(File("/opt/minifi-c2/minifi-c2-current/certs/minifi-c2-server-keystore.jks", server_keystore_content, permissions=0o644))
+
+ private_key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, test_context.root_ca_key)
+ private_key = load_pem_private_key(private_key_pem, password=None)
+ certificate_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, test_context.root_ca_cert)
+ certificate = x509.load_pem_x509_certificate(certificate_pem)
+ pkcs12_data = pkcs12.serialize_key_and_certificates(
+ name=None,
+ key=private_key,
+ cert=certificate,
+ cas=None,
+ encryption_algorithm=BestAvailableEncryption(b'abcdefgh')
+ )
+ self.files.append(File("/opt/minifi-c2/minifi-c2-current/certs/minifi-c2-server-truststore.p12", pkcs12_data, permissions=0o644))
+
+ authorities_file_content = """
+ CN=minifi-primary-{scenario_id}:
+ - CLASS_MINIFI_CPP
+ """.format(scenario_id=test_context.scenario_id)
+ self.files.append(File("/opt/minifi-c2/minifi-c2-current/conf/authorities.yaml", authorities_file_content, permissions=0o644))
+
+ resource_dir = Path(__file__).resolve().parent / "resources" / "minifi-c2-server"
+ self.host_files.append(HostFile("/opt/minifi-c2/minifi-c2-current/files/minifi-test-class/config.text.yml.v1", os.path.join(resource_dir, "config.yml")))
+ if ssl:
+ self.host_files.append(HostFile("/opt/minifi-c2/minifi-c2-current/conf/authorizations.yaml", os.path.join(resource_dir, "authorizations.yaml")))
+ self.host_files.append(HostFile("/opt/minifi-c2/minifi-c2-current/conf/c2.properties", os.path.join(resource_dir, "c2.properties")))
+
+ def deploy(self):
+ super().deploy()
+ finished_str = "Server Started"
+ return wait_for_condition(
+ condition=lambda: finished_str in self.get_logs(),
+ timeout_seconds=60,
+ bail_condition=lambda: self.exited,
+ context=None
+ )
diff --git a/docker/test/integration/resources/minifi-c2-server/authorizations.yaml b/extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/authorizations.yaml
similarity index 100%
rename from docker/test/integration/resources/minifi-c2-server/authorizations.yaml
rename to extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/authorizations.yaml
diff --git a/docker/test/integration/resources/minifi-c2-server/c2.properties b/extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/c2.properties
similarity index 100%
rename from docker/test/integration/resources/minifi-c2-server/c2.properties
rename to extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/c2.properties
diff --git a/docker/test/integration/resources/minifi-c2-server/config.yml b/extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/config.yml
similarity index 100%
rename from docker/test/integration/resources/minifi-c2-server/config.yml
rename to extensions/standard-processors/tests/features/steps/resources/minifi-c2-server/config.yml
diff --git a/extensions/standard-processors/tests/features/steps/steps.py b/extensions/standard-processors/tests/features/steps/steps.py
index 85158cb9e..517bff039 100644
--- a/extensions/standard-processors/tests/features/steps/steps.py
+++ b/extensions/standard-processors/tests/features/steps/steps.py
@@ -15,16 +15,19 @@
# limitations under the License.
#
-from behave import step
+import humanfriendly
+from behave import step, then, given
from minifi_test_framework.steps import checking_steps # noqa: F401
from minifi_test_framework.steps import configuration_steps # noqa: F401
from minifi_test_framework.steps import core_steps # noqa: F401
from minifi_test_framework.steps import flow_building_steps # noqa: F401
from minifi_test_framework.core.minifi_test_context import MinifiTestContext
+from minifi_test_framework.core.helpers import wait_for_condition
from syslog_container import SyslogContainer
from diag_slave_container import DiagSlave
from tcp_client_container import TcpClientContainer
+from minifi_c2_server_container import MinifiC2Server
@step("a Syslog client with TCP protocol is setup to send logs to minifi")
@@ -51,3 +54,52 @@ def step_impl(context: MinifiTestContext, modbus_cmd: str):
@step('a TCP client is set up to send a test TCP message to minifi')
def step_impl(context: MinifiTestContext):
context.containers["tcp-client"] = TcpClientContainer(context)
+
+
+@given("C2 is enabled in MiNiFi")
+def step_impl(context: MinifiTestContext):
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.enable", "true")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.rest.url", f"http://minifi-c2-server-{context.scenario_id}:10090/c2/config/heartbeat")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.rest.url.ack", f"http://minifi-c2-server-{context.scenario_id}:10090/c2/config/acknowledge")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.flow.base.url", f"http://minifi-c2-server-{context.scenario_id}:10090/c2/config/")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.root.classes", "DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.full.heartbeat", "false")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.agent.class", "minifi-test-class")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.agent.identifier", "minifi-test-id")
+
+
+@given("ssl properties are set up for MiNiFi C2 server")
+def step_impl(context: MinifiTestContext):
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.enable", "true")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.rest.url", f"https://minifi-c2-server-{context.scenario_id}:10090/c2/config/heartbeat")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.rest.url.ack", f"https://minifi-c2-server-{context.scenario_id}:10090/c2/config/acknowledge")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.flow.base.url", f"https://minifi-c2-server-{context.scenario_id}:10090/c2/config/")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.root.classes", "DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.full.heartbeat", "false")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.agent.class", "minifi-test-class")
+ context.get_or_create_default_minifi_container().set_property("nifi.c2.agent.identifier", "minifi-test-id")
+ context.get_or_create_default_minifi_container().set_up_ssl_properties()
+
+
+@given("a MiNiFi C2 server is set up")
+def step_impl(context: MinifiTestContext):
+ context.containers["minifi-c2-server"] = MinifiC2Server(context)
+
+
+@given("a MiNiFi C2 server is set up with SSL")
+def step_impl(context: MinifiTestContext):
+ context.containers["minifi-c2-server"] = MinifiC2Server(context, ssl=True)
+
+
+@given("a MiNiFi C2 server is started")
+def step_impl(context: MinifiTestContext):
+ context.containers["minifi-c2-server"] = MinifiC2Server(context)
+ assert context.containers["minifi-c2-server"].deploy()
+
+
+@then("the MiNiFi C2 server logs contain the following message: \"{log_message}\" in less than {duration}")
+def step_impl(context: MinifiTestContext, log_message: str, duration: str):
+ duration_seconds = humanfriendly.parse_timespan(duration)
+ assert wait_for_condition(condition=lambda: log_message in context.containers["minifi-c2-server"].get_logs(),
+ timeout_seconds=duration_seconds, bail_condition=lambda: context.containers["minifi-c2-server"].exited,
+ context=context)
(nifi-minifi-cpp) 02/05: MINIFICPP-2685 Move MQTT tests to modular docker tests
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit e730451c933bf184796014fd538dd4e9ae8debc7
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Jan 20 17:18:25 2026 +0100
MINIFICPP-2685 Move MQTT tests to modular docker tests
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #2091
---
.../minifi_test_framework/containers/container.py | 25 ++
.../minifi_test_framework/steps/checking_steps.py | 14 +-
.../src/minifi_test_framework/steps/core_steps.py | 46 ++-
.../steps/flow_building_steps.py | 10 +-
docker/RunBehaveTests.sh | 3 +-
docker/requirements.txt | 1 -
docker/test/integration/cluster/ContainerStore.py | 9 -
.../test/integration/cluster/DockerTestCluster.py | 5 -
docker/test/integration/cluster/ImageStore.py | 11 -
.../cluster/containers/MqttBrokerContainer.py | 40 ---
.../features/MiNiFi_integration_test_driver.py | 3 -
docker/test/integration/features/steps/steps.py | 27 --
.../integration/minifi/processors/ConsumeMQTT.py | 30 --
.../integration/minifi/processors/PublishMQTT.py | 30 --
.../features/steps/couchbase_server_container.py | 4 +-
.../tests/features/steps/grafana_loki_container.py | 4 +-
extensions/mqtt/tests/features/environment.py | 44 +++
.../mqtt/tests}/features/mqtt.feature | 398 ++++++++++++---------
.../features/resources/publish_mqtt_message.py | 20 +-
.../tests/features/steps/mqtt_broker_container.py | 63 ++++
extensions/mqtt/tests/features/steps/steps.py | 76 ++++
21 files changed, 521 insertions(+), 342 deletions(-)
diff --git a/behave_framework/src/minifi_test_framework/containers/container.py b/behave_framework/src/minifi_test_framework/containers/container.py
index 8c59586f5..f3268d2dd 100644
--- a/behave_framework/src/minifi_test_framework/containers/container.py
+++ b/behave_framework/src/minifi_test_framework/containers/container.py
@@ -20,6 +20,7 @@ import os
import shlex
import tempfile
import tarfile
+import uuid
import docker
from docker.models.networks import Network
@@ -50,6 +51,24 @@ class Container:
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 add_file_to_running_container(self, content: str, path: str):
+ if not self.container:
+ logging.error("Container is not running. Cannot add file.")
+ raise RuntimeError("Container is not running. Cannot add file.")
+
+ mkdir_command = f"mkdir -p {shlex.quote(path)}"
+ exit_code, output = self.exec_run(mkdir_command)
+ if exit_code != 0:
+ logging.error(f"Error creating directory '{path}' in container: {output}")
+ raise RuntimeError(f"Error creating directory '{path}' in container: {output}")
+
+ full_path = os.path.join(path, str(uuid.uuid4()))
+ pipe_command = f"printf %s {shlex.quote(content)} > {shlex.quote(full_path)}"
+ exit_code, output = self.exec_run(f"sh -c {shlex.quote(pipe_command)}")
+ if exit_code != 0:
+ logging.error(f"Error adding file to running container: {output}")
+ raise RuntimeError(f"Error adding file to running container: {output}")
+
def _write_content_to_file(self, filepath: str, permissions: int | None, content: str | bytes):
write_mode = "w"
if isinstance(content, bytes):
@@ -74,7 +93,13 @@ class Container:
self._write_content_to_file(file_path, None, content)
self.volumes[temp_path] = {"bind": directory.path, "mode": directory.mode}
+ def is_deployed(self) -> bool:
+ return self.container is not None
+
def deploy(self) -> bool:
+ if self.is_deployed():
+ logging.info(f"Container '{self.container_name}' is already deployed.")
+ return True
self._temp_dir = tempfile.TemporaryDirectory()
self._configure_volumes_of_container_files()
self._configure_volumes_of_container_dirs()
diff --git a/behave_framework/src/minifi_test_framework/steps/checking_steps.py b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
index ef5b1a338..c6593c1bf 100644
--- a/behave_framework/src/minifi_test_framework/steps/checking_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
@@ -36,13 +36,18 @@ def step_impl(context: MinifiTestContext, content: str, path: str, duration: str
timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].exited, context=context)
-@then('a single file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
-def step_impl(context: MinifiTestContext, content: str, directory: str, duration: str):
+@then('in the "{container_name}" container a single file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
+def step_impl(context: MinifiTestContext, container_name: str, content: str, directory: str, duration: str):
new_content = content.replace("\\n", "\n")
timeout_in_seconds = humanfriendly.parse_timespan(duration)
assert wait_for_condition(
- condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].directory_has_single_file_with_content(directory, new_content),
- timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].exited, context=context)
+ condition=lambda: context.containers[container_name].directory_has_single_file_with_content(directory, new_content),
+ timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.containers[container_name].exited, context=context)
+
+
+@then('a single file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
+def step_impl(context: MinifiTestContext, content: str, directory: str, duration: str):
+ context.execute_steps(f'then in the "{DEFAULT_MINIFI_CONTAINER_NAME}" container a single file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
@then('in the "{container_name}" container at least one file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
@@ -147,6 +152,7 @@ def step_impl(context: MinifiTestContext, directory: str, regex_str: str, durati
@then('files with contents "{content_one}" and "{content_two}" are placed in the "{directory}" directory in less than {timeout}')
+@then("files with contents '{content_one}' and '{content_two}' are placed in the '{directory}' directory in less than {timeout}")
def step_impl(context: MinifiTestContext, directory: str, timeout: str, content_one: str, content_two: str):
timeout_seconds = humanfriendly.parse_timespan(timeout)
c1 = content_one.replace("\\n", "\n")
diff --git a/behave_framework/src/minifi_test_framework/steps/core_steps.py b/behave_framework/src/minifi_test_framework/steps/core_steps.py
index 6b488587c..b332b7e46 100644
--- a/behave_framework/src/minifi_test_framework/steps/core_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/core_steps.py
@@ -61,10 +61,27 @@ 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}"')
+@given('a file with the content "{content}" is present in "{path}" in the "{container_name}" flow')
+def step_impl(context: MinifiTestContext, content: str, path: str, container_name: str):
+ new_content = content.replace("\\n", "\n")
+ context.get_or_create_minifi_container(container_name).files.append(File(os.path.join(path, str(uuid.uuid4())), new_content))
+
+
+@given('a file with the content "{content}" is present in "{path}"')
+@given("a file with the content '{content}' is present in '{path}'")
def step_impl(context: MinifiTestContext, content: str, path: str):
+ context.execute_steps(f"given a file with the content \"{content}\" is present in \"{path}\" in the \"{DEFAULT_MINIFI_CONTAINER_NAME}\" flow")
+
+
+@when('a file with the content "{content}" is placed in "{path}" in the "{container_name}" flow')
+def step_impl(context: MinifiTestContext, content: str, path: str, container_name: 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))
+ context.containers[container_name].add_file_to_running_container(new_content, path)
+
+
+@when('a file with the content "{content}" is placed in "{path}"')
+def step_impl(context: MinifiTestContext, content: str, path: str):
+ context.execute_steps(f"when a file with the content \"{content}\" is placed in \"{path}\" in the \"{DEFAULT_MINIFI_CONTAINER_NAME}\" flow")
@given("an empty file is present in \"{path}\"")
@@ -94,11 +111,31 @@ def step_impl(context: MinifiTestContext):
context.get_or_create_default_minifi_container().stop()
+@when("the \"{container_name}\" flow is stopped")
+def step_impl(context: MinifiTestContext, container_name: str):
+ context.get_or_create_minifi_container(container_name).stop()
+
+
@when("MiNiFi is restarted")
def step_impl(context: MinifiTestContext):
context.get_or_create_default_minifi_container().restart()
+@when("the \"{container_name}\" flow is restarted")
+def step_impl(context: MinifiTestContext, container_name: str):
+ context.get_or_create_minifi_container(container_name).restart()
+
+
+@when("the \"{container_name}\" flow is started")
+def step_impl(context: MinifiTestContext, container_name: str):
+ context.get_or_create_minifi_container(container_name).start()
+
+
+@when("the \"{container_name}\" flow is killed")
+def step_impl(context: MinifiTestContext, container_name: str):
+ context.get_or_create_minifi_container(container_name).kill()
+
+
@given("OpenSSL FIPS mode is enabled in MiNiFi")
def step_impl(context: MinifiTestContext):
context.get_or_create_default_minifi_container().enable_openssl_fips_mode()
@@ -117,3 +154,8 @@ def step_impl(context: MinifiTestContext):
@given("flow configuration path is set up in flow url property")
def step_impl(context: MinifiTestContext):
context.get_or_create_default_minifi_container().fetch_flow_config_from_flow_url()
+
+
+@step("{duration} later")
+def step_impl(context: MinifiTestContext, duration: str):
+ time.sleep(humanfriendly.parse_timespan(duration))
diff --git a/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py b/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
index 474dd5a0a..770f222a7 100644
--- a/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
@@ -70,10 +70,16 @@ def step_impl(context: MinifiTestContext, processor_type: str, property_name: st
context.containers["nifi"].flow_definition.add_processor(processor)
+@given('a {processor_type} processor with the name "{processor_name}" in the "{minifi_container_name}" flow')
+def step_impl(context: MinifiTestContext, processor_type: str, processor_name: str, minifi_container_name: str):
+ processor = Processor(processor_type, processor_name)
+ context.get_or_create_minifi_container(minifi_container_name).flow_definition.add_processor(processor)
+
+
@given('a {processor_type} processor with the name "{processor_name}"')
def step_impl(context: MinifiTestContext, processor_type: str, processor_name: str):
- processor = Processor(processor_type, processor_name)
- context.get_or_create_default_minifi_container().flow_definition.add_processor(processor)
+ context.execute_steps(
+ f'given a {processor_type} processor with the name "{processor_name}" in the "{DEFAULT_MINIFI_CONTAINER_NAME}" flow')
@given("a {processor_type} processor in the \"{minifi_container_name}\" flow")
diff --git a/docker/RunBehaveTests.sh b/docker/RunBehaveTests.sh
index dff7fccc9..f982e3c6a 100755
--- a/docker/RunBehaveTests.sh
+++ b/docker/RunBehaveTests.sh
@@ -207,4 +207,5 @@ exec \
"${docker_dir}/../extensions/gcp/tests/features" \
"${docker_dir}/../extensions/grafana-loki/tests/features" \
"${docker_dir}/../extensions/lua/tests/features/" \
- "${docker_dir}/../extensions/civetweb/tests/features/"
+ "${docker_dir}/../extensions/civetweb/tests/features/" \
+ "${docker_dir}/../extensions/mqtt/tests/features/"
diff --git a/docker/requirements.txt b/docker/requirements.txt
index a50c768a0..0092b5e25 100644
--- a/docker/requirements.txt
+++ b/docker/requirements.txt
@@ -10,4 +10,3 @@ azure-storage-blob==12.24.1
prometheus-api-client==0.5.5
humanfriendly==10.0
requests<2.29 # https://github.com/docker/docker-py/issues/3113
-paho-mqtt==2.1.0
diff --git a/docker/test/integration/cluster/ContainerStore.py b/docker/test/integration/cluster/ContainerStore.py
index f139acd21..4b1b69a29 100644
--- a/docker/test/integration/cluster/ContainerStore.py
+++ b/docker/test/integration/cluster/ContainerStore.py
@@ -21,7 +21,6 @@ from .containers.NifiContainer import NiFiOptions
from .containers.AzureStorageServerContainer import AzureStorageServerContainer
from .containers.HttpProxyContainer import HttpProxyContainer
from .containers.PostgreSQLServerContainer import PostgreSQLServerContainer
-from .containers.MqttBrokerContainer import MqttBrokerContainer
from .containers.SyslogUdpClientContainer import SyslogUdpClientContainer
from .containers.SyslogTcpClientContainer import SyslogTcpClientContainer
from .containers.MinifiAsPodInKubernetesCluster import MinifiAsPodInKubernetesCluster
@@ -128,14 +127,6 @@ class ContainerStore:
network=self.network,
image_store=self.image_store,
command=command))
- elif engine == 'mqtt-broker':
- return self.containers.setdefault(container_name,
- MqttBrokerContainer(feature_context=feature_context,
- name=container_name,
- vols=self.vols,
- network=self.network,
- image_store=self.image_store,
- command=command))
elif engine == "syslog-udp-client":
return self.containers.setdefault(container_name,
SyslogUdpClientContainer(
diff --git a/docker/test/integration/cluster/DockerTestCluster.py b/docker/test/integration/cluster/DockerTestCluster.py
index 458e0bd88..006aa97c5 100644
--- a/docker/test/integration/cluster/DockerTestCluster.py
+++ b/docker/test/integration/cluster/DockerTestCluster.py
@@ -23,7 +23,6 @@ from .checkers.AzureChecker import AzureChecker
from .checkers.PostgresChecker import PostgresChecker
from .checkers.PrometheusChecker import PrometheusChecker
from .checkers.ModbusChecker import ModbusChecker
-from .checkers.MqttHelper import MqttHelper
from utils import get_peak_memory_usage, get_minifi_pid, get_memory_usage
@@ -37,7 +36,6 @@ class DockerTestCluster:
self.postgres_checker = PostgresChecker(self.container_communicator)
self.prometheus_checker = PrometheusChecker()
self.modbus_checker = ModbusChecker(self.container_communicator)
- self.mqtt_helper = MqttHelper()
def cleanup(self):
self.container_store.cleanup()
@@ -278,6 +276,3 @@ class DockerTestCluster:
def enable_ssl_in_nifi(self):
self.container_store.enable_ssl_in_nifi()
-
- def publish_test_mqtt_message(self, topic: str, message: str):
- self.mqtt_helper.publish_test_mqtt_message(topic, message)
diff --git a/docker/test/integration/cluster/ImageStore.py b/docker/test/integration/cluster/ImageStore.py
index 50025e37a..a729cd030 100644
--- a/docker/test/integration/cluster/ImageStore.py
+++ b/docker/test/integration/cluster/ImageStore.py
@@ -63,8 +63,6 @@ class ImageStore:
image = self.__build_http_proxy_image()
elif container_engine == "postgresql-server":
image = self.__build_postgresql_server_image()
- elif container_engine == "mqtt-broker":
- image = self.__build_mqtt_broker_image()
else:
raise Exception("There is no associated image for " + container_engine)
@@ -277,15 +275,6 @@ class ImageStore:
""".format(base_image='postgres:17.4'))
return self.__build_image(dockerfile)
- def __build_mqtt_broker_image(self):
- dockerfile = dedent("""\
- FROM {base_image}
- RUN echo 'log_dest stderr' >> /mosquitto-no-auth.conf
- CMD ["/usr/sbin/mosquitto", "--verbose", "--config-file", "/mosquitto-no-auth.conf"]
- """.format(base_image='eclipse-mosquitto:2.0.14'))
-
- return self.__build_image(dockerfile)
-
def __build_image(self, dockerfile, context_files=[]):
conf_dockerfile_buffer = BytesIO()
docker_context_buffer = BytesIO()
diff --git a/docker/test/integration/cluster/containers/MqttBrokerContainer.py b/docker/test/integration/cluster/containers/MqttBrokerContainer.py
deleted file mode 100644
index faa168f31..000000000
--- a/docker/test/integration/cluster/containers/MqttBrokerContainer.py
+++ /dev/null
@@ -1,40 +0,0 @@
-# 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 logging
-from .Container import Container
-
-
-class MqttBrokerContainer(Container):
- def __init__(self, feature_context, name, vols, network, image_store, command=None):
- super().__init__(feature_context, name, 'mqtt-broker', vols, network, image_store, command)
-
- def get_startup_finished_log_entry(self):
- return "mosquitto version [0-9\\.]+ running"
-
- def deploy(self):
- if not self.set_deployed():
- return
-
- logging.info('Creating and running MQTT broker docker container...')
- self.client.containers.run(
- self.image_store.get_image(self.get_engine()),
- detach=True,
- name=self.name,
- ports={'1883/tcp': 1883},
- network=self.network.name,
- entrypoint=self.command)
- logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index 7a36309cd..a85c48bb7 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -389,6 +389,3 @@ class MiNiFi_integration_test:
def enable_ssl_in_nifi(self):
self.cluster.enable_ssl_in_nifi()
-
- def publish_test_mqtt_message(self, topic, message):
- self.cluster.publish_test_mqtt_message(topic, message)
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index ce222141b..521679e6f 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -121,7 +121,6 @@ def step_impl(context, processor_type, minifi_container_name):
@given("a {processor_type} processor")
@given("a {processor_type} processor set up to communicate with an Azure blob storage")
-@given("a {processor_type} processor set up to communicate with an MQTT broker instance")
def step_impl(context, processor_type):
__create_processor(context, processor_type, processor_type, None, None, "minifi-cpp-flow")
@@ -486,16 +485,6 @@ def step_impl(context, processor_name, service_property_name, property_name, pro
__set_up_the_kubernetes_controller_service(context, processor_name, service_property_name, {property_name: property_value})
-# MQTT setup
-@when("an MQTT broker is set up in correspondence with the PublishMQTT")
-@given("an MQTT broker is set up in correspondence with the PublishMQTT")
-@given("an MQTT broker is set up in correspondence with the ConsumeMQTT")
-@given("an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT")
-def step_impl(context):
- context.test.acquire_container(context=context, name="mqtt-broker", engine="mqtt-broker")
- context.test.start('mqtt-broker')
-
-
# azure storage setup
@given("an Azure storage server is set up")
def step_impl(context):
@@ -779,22 +768,6 @@ def step_impl(context, regex, duration):
context.test.check_minifi_log_matches_regex(regex, humanfriendly.parse_timespan(duration))
-# MQTT
-@then("the MQTT broker has a log line matching \"{log_pattern}\"")
-def step_impl(context, log_pattern):
- context.test.check_container_log_matches_regex('mqtt-broker', log_pattern, 60, count=1)
-
-
-@then("the MQTT broker has {log_count} log lines matching \"{log_pattern}\"")
-def step_impl(context, log_count, log_pattern):
- context.test.check_container_log_matches_regex('mqtt-broker', log_pattern, 60, count=int(log_count))
-
-
-@when("a test message \"{message}\" is published to the MQTT broker on topic \"{topic}\"")
-def step_impl(context, message, topic):
- context.test.publish_test_mqtt_message(topic, message)
-
-
@then("the \"{minifi_container_name}\" flow has a log line matching \"{log_pattern}\" in less than {duration}")
def step_impl(context, minifi_container_name, log_pattern, duration):
context.test.check_container_log_matches_regex(minifi_container_name, log_pattern, humanfriendly.parse_timespan(duration), count=1)
diff --git a/docker/test/integration/minifi/processors/ConsumeMQTT.py b/docker/test/integration/minifi/processors/ConsumeMQTT.py
deleted file mode 100644
index 8c64be6a3..000000000
--- a/docker/test/integration/minifi/processors/ConsumeMQTT.py
+++ /dev/null
@@ -1,30 +0,0 @@
-# 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.
-
-
-from ..core.Processor import Processor
-
-
-class ConsumeMQTT(Processor):
- def __init__(self, context, schedule={'scheduling strategy': 'TIMER_DRIVEN'}):
- super(ConsumeMQTT, self).__init__(
- context=context,
- clazz='ConsumeMQTT',
- properties={
- 'Broker URI': f'mqtt-broker-{context.feature_id}:1883',
- 'Topic': 'testtopic',
- 'Client ID': 'consumer-client'},
- auto_terminate=['success'],
- schedule=schedule)
diff --git a/docker/test/integration/minifi/processors/PublishMQTT.py b/docker/test/integration/minifi/processors/PublishMQTT.py
deleted file mode 100644
index 41387ce76..000000000
--- a/docker/test/integration/minifi/processors/PublishMQTT.py
+++ /dev/null
@@ -1,30 +0,0 @@
-# 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.
-
-
-from ..core.Processor import Processor
-
-
-class PublishMQTT(Processor):
- def __init__(self, context, schedule={'scheduling strategy': 'EVENT_DRIVEN'}):
- super(PublishMQTT, self).__init__(
- context=context,
- clazz='PublishMQTT',
- properties={
- 'Broker URI': f'mqtt-broker-{context.feature_id}:1883',
- 'Topic': 'testtopic',
- 'Client ID': 'publisher-client'},
- auto_terminate=['success', 'failure'],
- schedule=schedule)
diff --git a/extensions/couchbase/tests/features/steps/couchbase_server_container.py b/extensions/couchbase/tests/features/steps/couchbase_server_container.py
index 96cddf821..893ba9de9 100644
--- a/extensions/couchbase/tests/features/steps/couchbase_server_container.py
+++ b/extensions/couchbase/tests/features/steps/couchbase_server_container.py
@@ -82,8 +82,8 @@ class CouchbaseServerContainer(Container):
self.client.containers.run("minifi-couchbase-helper:latest", ["python", "-c", command], remove=True, stdout=True, stderr=True, network=self.network.name)
return True
except ContainerError as e:
- stdout = e.stdout.decode("utf-8", errors="replace") if e.stdout else ""
- stderr = e.stderr.decode("utf-8", errors="replace") if e.stderr else ""
+ stdout = e.stdout.decode("utf-8", errors="replace") if hasattr(e, "stdout") and e.stdout else ""
+ stderr = e.stderr.decode("utf-8", errors="replace") if hasattr(e, "stderr") and e.stderr else ""
logging.error(f"Python command '{command}' failed in couchbase helper docker with error: '{e}', stdout: '{stdout}', stderr: '{stderr}'")
return False
except Exception as e:
diff --git a/extensions/grafana-loki/tests/features/steps/grafana_loki_container.py b/extensions/grafana-loki/tests/features/steps/grafana_loki_container.py
index 46fc84b5a..3d7e497aa 100644
--- a/extensions/grafana-loki/tests/features/steps/grafana_loki_container.py
+++ b/extensions/grafana-loki/tests/features/steps/grafana_loki_container.py
@@ -142,8 +142,8 @@ analytics:
remove=True, stdout=True, stderr=True, network=self.network.name)
return True
except ContainerError as e:
- stdout = e.stdout.decode("utf-8", errors="replace") if e.stdout else ""
- stderr = e.stderr.decode("utf-8", errors="replace") if e.stderr else ""
+ stdout = e.stdout.decode("utf-8", errors="replace") if hasattr(e, "stdout") and e.stdout else ""
+ stderr = e.stderr.decode("utf-8", errors="replace") if hasattr(e, "stderr") and e.stderr else ""
logging.error(f"Failed to run python command in grafana loki helper docker with error: '{e}', stdout: '{stdout}', stderr: '{stderr}'")
return False
except Exception as e:
diff --git a/extensions/mqtt/tests/features/environment.py b/extensions/mqtt/tests/features/environment.py
new file mode 100644
index 000000000..b3bbe9df0
--- /dev/null
+++ b/extensions/mqtt/tests/features/environment.py
@@ -0,0 +1,44 @@
+# 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.
+
+from pathlib import Path
+from minifi_test_framework.containers.docker_image_builder import DockerImageBuilder
+from minifi_test_framework.core.hooks import common_before_scenario
+from minifi_test_framework.core.hooks import common_after_scenario
+
+
+def before_all(context):
+ check_log_lines_path = Path(__file__).resolve().parent / "resources" / "publish_mqtt_message.py"
+ check_log_lines_content = None
+ with open(check_log_lines_path, "rb") as f:
+ check_log_lines_content = f.read()
+ dockerfile = """
+FROM python:3.13-slim-bookworm
+RUN pip install paho-mqtt==2.1.0
+COPY publish_mqtt_message.py /scripts/publish_mqtt_message.py"""
+ mqtt_helper_builder = DockerImageBuilder(
+ image_tag="minifi-mqtt-helper:latest",
+ dockerfile_content=dockerfile,
+ files_on_context={"publish_mqtt_message.py": check_log_lines_content}
+ )
+ mqtt_helper_builder.build()
+
+
+def before_scenario(context, scenario):
+ common_before_scenario(context, scenario)
+
+
+def after_scenario(context, scenario):
+ common_after_scenario(context, scenario)
diff --git a/docker/test/integration/features/mqtt.feature b/extensions/mqtt/tests/features/mqtt.feature
similarity index 60%
rename from docker/test/integration/features/mqtt.feature
rename to extensions/mqtt/tests/features/mqtt.feature
index 64ef62f6b..1f4bd10d4 100644
--- a/docker/test/integration/features/mqtt.feature
+++ b/extensions/mqtt/tests/features/mqtt.feature
@@ -19,22 +19,21 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
As a user of MiNiFi
I need to have PublishMQTT and ConsumeMQTT processors
- Background:
- Given the content of "/tmp/output" is monitored
-
Scenario Outline: A MiNiFi instance transfers data to an MQTT broker
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content "test" is present in "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
And the "success" relationship of the PublishMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT
+ And an MQTT broker is started
- When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When the MiNiFi instance starts up
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
Examples: MQTT versions
@@ -48,12 +47,14 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
And a PublishMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
And the "success" relationship of the PublishMQTT processor is connected to the PutFile
And the "failure" relationship of the PublishMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
When the MiNiFi instance starts up
- Then no files are placed in the monitored directory in 30 seconds of running time
+ Then no files are placed in the "/tmp/output" directory in 30 seconds of running time
Examples: MQTT versions
| version |
@@ -66,14 +67,16 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
And a PublishMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
And the "success" relationship of the PublishMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
When the MiNiFi instance starts up
- Then no files are placed in the monitored directory in 10 seconds of running time
+ Then no files are placed in the "/tmp/output" directory in 10 seconds of running time
- When an MQTT broker is set up in correspondence with the PublishMQTT
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When an MQTT broker is started
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
Examples: MQTT versions
@@ -84,24 +87,28 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: A MiNiFi instance publishes and consumes data to/from an MQTT broker
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And a LogAttribute processor
- And "ConsumeMQTT" processor is a start node
+ And LogAttribute is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
And the "success" relationship of the PutFile processor is connected to the LogAttribute
+ And LogAttribute's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input"
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
And the Minifi logs contain the following message: "key:mqtt.broker value:mqtt-broker-" in less than 60 seconds
And the Minifi logs contain the following message: "key:mqtt.topic value:testtopic" in less than 1 seconds
@@ -120,24 +127,27 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: Subscription to topics with wildcards
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "Topic" property of the PublishMQTT processor is set to "test/my/topic"
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "Topic" property of the ConsumeMQTT processor is set to "<topic wildcard pattern>"
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And "ConsumeMQTT" processor is a start node
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input"
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*test/my/topic.*\(4 bytes\)"
Examples: Topic wildcard patterns
@@ -150,33 +160,37 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: Subscription and publishing with disconnecting clients in persistent sessions
# publishing MQTT client
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
- And the "Quality of Service" property of the PublishMQTT processor is set to "1"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "<version>" in the "publisher-client" flow
+ And the "Quality of Service" property of the PublishMQTT processor is set to "1" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
- And the "Quality of Service" property of the ConsumeMQTT processor is set to "1"
- And the "<persistent_session_property_1>" property of the ConsumeMQTT processor is set to "<persistent_session_property_1_value>"
- And the "<persistent_session_property_2>" property of the ConsumeMQTT processor is set to "<persistent_session_property_2_value>"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>" in the "consumer-client" flow
+ And the "Quality of Service" property of the ConsumeMQTT processor is set to "1" in the "consumer-client" flow
+ And the "<persistent_session_property_1>" property of the ConsumeMQTT processor is set to "<persistent_session_property_1_value>" in the "consumer-client" flow
+ And the "<persistent_session_property_2>" property of the ConsumeMQTT processor is set to "<persistent_session_property_2_value>" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
When all instances start up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And "consumer-client" flow is stopped
- And the MQTT broker has a log line matching "Received DISCONNECT from consumer-client"
+ When the "consumer-client" flow is stopped
+ Then the MQTT broker has a log line matching "Received DISCONNECT from consumer-client"
- And a file with the content "test" is placed in "/tmp/input"
- And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
+ When a file with the content "test" is placed in "/tmp/input" in the "publisher-client" flow
+ Then the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
- And "consumer-client" flow is restarted
- And the MQTT broker has 2 log lines matching "New client connected from .* as consumer-client"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When the "consumer-client" flow is restarted
+ Then the MQTT broker has 2 log lines matching "New client connected from .* as consumer-client"
+ And in the "consumer-client" container a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
Examples: MQTT versions
| version | persistent_session_property_1 | persistent_session_property_1_value | persistent_session_property_2 | persistent_session_property_2_value |
@@ -186,24 +200,27 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: UTF-8 topics and messages
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "Topic" property of the PublishMQTT processor is set to "<topic>"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And the "Topic" property of the ConsumeMQTT processor is set to "<topic>"
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And "ConsumeMQTT" processor is a start node
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "<message>" is placed in "/tmp/input"
- And a flowfile with the content "<message>" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "<message>" is placed in "/tmp/input"
+ Then a single file with the content "<message>" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*<topic>"
Examples: Topics, messages and version
@@ -214,28 +231,30 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
| Ù…Øمد بن موسی خوارزمی | ٱلْجَبْر | 3.x AUTO |
| תַּלְמוּד | תּוֹרָה | 3.x AUTO |
-
Scenario Outline: QoS 0 message flow is correct
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "Quality of Service" property of the PublishMQTT processor is set to "0"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And the "Quality of Service" property of the ConsumeMQTT processor is set to "0"
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And "ConsumeMQTT" processor is a start node
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input"
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from publisher-client \(d0, q0, r0, m0, 'testtopic'.*\(4 bytes\)"
And the MQTT broker has a log line matching "Sending PUBLISH to consumer-client \(d0, q0, r0, m0, 'testtopic',.*\(4 bytes\)\)"
@@ -247,24 +266,27 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: QoS 1 Subscriber sends PUBACK on a PUBLISH message, with correct packet ID
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "Quality of Service" property of the PublishMQTT processor is set to "1"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And the "Quality of Service" property of the ConsumeMQTT processor is set to "1"
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And "ConsumeMQTT" processor is a start node
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input"
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from publisher-client.*m1, 'testtopic'.*\(4 bytes\)"
And the MQTT broker has a log line matching "Sending PUBACK to publisher-client \(m1, rc0\)"
And the MQTT broker has a log line matching "Sending PUBLISH to consumer-client \(d0, q1, r0, m1, 'testtopic',.*\(4 bytes\)\)"
@@ -278,24 +300,27 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: QoS 2 message flow is correct
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
And the "Quality of Service" property of the PublishMQTT processor is set to "2"
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated
And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And the "Quality of Service" property of the ConsumeMQTT processor is set to "2"
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And "ConsumeMQTT" processor is a start node
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input"
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from publisher-client.*m1, 'testtopic'.*\(4 bytes\)"
And the MQTT broker has a log line matching "Sending PUBREC to publisher-client \(m1, rc0\)"
And the MQTT broker has a log line matching "Received PUBREL from publisher-client \(Mid: 1\)"
@@ -313,28 +338,33 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: Retained message
# publishing MQTT client
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a file with the content "test" is present in "/tmp/input"
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
- And the "Retain" property of the PublishMQTT processor is set to "true"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And the scheduling period of the GetFile processor is set to "120 seconds" in the "publisher-client" flow
+ And a file with the content "test" is present in "/tmp/input" in the "publisher-client" flow
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "<version>" in the "publisher-client" flow
+ And the "Retain" property of the PublishMQTT processor is set to "true" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When all instances start up
Then the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
# consumer is joining late, but it will still see the retained message
- And "consumer-client" flow is started
- And the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
+ When the "consumer-client" flow is started
+ Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ And in the "consumer-client" container a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
Examples: MQTT versions
| version |
@@ -344,27 +374,31 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: Last will
# publishing MQTT client with last will
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "<version>"
- And the "Last Will Topic" property of the PublishMQTT processor is set to "last_will_topic"
- And the "Last Will Message" property of the PublishMQTT processor is set to "last_will_message"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "<version>" in the "publisher-client" flow
+ And the "Last Will Topic" property of the PublishMQTT processor is set to "last_will_topic" in the "publisher-client" flow
+ And the "Last Will Message" property of the PublishMQTT processor is set to "last_will_message" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client set to consume last will topic
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
- And the "Topic" property of the ConsumeMQTT processor is set to "last_will_topic"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>" in the "consumer-client" flow
+ And the "Topic" property of the ConsumeMQTT processor is set to "last_will_topic" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
When all instances start up
Then the MQTT broker has a log line matching "Sending CONNACK to publisher-client"
And the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And "publisher-client" flow is killed
- And the MQTT broker has a log line matching "Sending PUBLISH to consumer-client"
- And a flowfile with the content "last_will_message" is placed in the monitored directory in less than 60 seconds
+ When the "publisher-client" flow is killed
+ Then the MQTT broker has a log line matching "Sending PUBLISH to consumer-client"
+ And in the "consumer-client" container a single file with the content "last_will_message" is placed in the "/tmp/output" directory in less than 60 seconds
Examples: MQTT versions
| version |
@@ -376,9 +410,9 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
And the "MQTT Version" property of the ConsumeMQTT processor is set to "<version>"
And the "Keep Alive Interval" property of the ConsumeMQTT processor is set to "1 sec"
- And an MQTT broker is set up in correspondence with the ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
Then the MQTT broker has a log line matching "Received PINGREQ from consumer-client"
Then the MQTT broker has a log line matching "Sending PINGRESP to consumer-client"
@@ -390,171 +424,201 @@ Feature: Sending data to MQTT streaming platform using PublishMQTT
Scenario Outline: Message Expiry Interval - MQTT 5
# publishing MQTT client
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "5.0"
- And the "Message Expiry Interval" property of the PublishMQTT processor is set to "<message_expiry_interval>"
- And the "Quality of Service" property of the PublishMQTT processor is set to "1"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "5.0" in the "publisher-client" flow
+ And the "Message Expiry Interval" property of the PublishMQTT processor is set to "<message_expiry_interval>" in the "publisher-client" flow
+ And the "Quality of Service" property of the PublishMQTT processor is set to "1" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0"
- And the "Quality of Service" property of the ConsumeMQTT processor is set to "1"
- And the "Clean Start" property of the ConsumeMQTT processor is set to "false"
- And the "Session Expiry Interval" property of the ConsumeMQTT processor is set to "1 h"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0" in the "consumer-client" flow
+ And the "Quality of Service" property of the ConsumeMQTT processor is set to "1" in the "consumer-client" flow
+ And the "Clean Start" property of the ConsumeMQTT processor is set to "false" in the "consumer-client" flow
+ And the "Session Expiry Interval" property of the ConsumeMQTT processor is set to "1 h" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
When all instances start up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And "consumer-client" flow is stopped
- And the MQTT broker has a log line matching "Received DISCONNECT from consumer-client"
+ When the "consumer-client" flow is stopped
+ Then the MQTT broker has a log line matching "Received DISCONNECT from consumer-client"
- And a file with the content "test" is placed in "/tmp/input"
- And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
+ When a file with the content "test" is placed in "/tmp/input" in the "publisher-client" flow
+ Then the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
And 2 seconds later
- And "consumer-client" flow is restarted
- And the MQTT broker has 2 log lines matching "New client connected from .* as consumer-client"
- And <expectation_num_files> placed in the monitored directory in <expectation_time_limit>
+ When the "consumer-client" flow is restarted
+ Then the MQTT broker has 2 log lines matching "New client connected from .* as consumer-client"
+ And in the "consumer-client" container <expectation_num_files> placed in the "/tmp/output" directory in <expectation_time_limit>
Examples: Message Expiry Intervals
- | message_expiry_interval | expectation_num_files | expectation_time_limit |
- | 1 h | a flowfile with the content "test" is | less than 60 seconds |
- | 1 s | no files are | 30 seconds of running time |
+ | message_expiry_interval | expectation_num_files | expectation_time_limit |
+ | 1 h | a single file with the content "test" is | less than 60 seconds |
+ | 1 s | no files are | 30 seconds of running time |
Scenario: User properties - MQTT 5
# publishing MQTT client: GetFile -> UpdateAttribute (my_attr1:true, my_attr2:true) -> PublishMQTT
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
And a UpdateAttribute processor in the "publisher-client" flow
- And the "my_attr1" property of the UpdateAttribute processor is set to "true"
- And the "my_attr2" property of the UpdateAttribute processor is set to "true"
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "5.0"
- And the "success" relationship of the GetFile processor is connected to the UpdateAttribute
- And the "success" relationship of the UpdateAttribute processor is connected to the PublishMQTT
+ And the "my_attr1" property of the UpdateAttribute processor is set to "true" in the "publisher-client" flow
+ And the "my_attr2" property of the UpdateAttribute processor is set to "true" in the "publisher-client" flow
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "5.0" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the UpdateAttribute
+ And in the "publisher-client" flow the "success" relationship of the UpdateAttribute processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client: ConsumeMQTT -> RouteOnAttribute (my_attr1) -> RouteOnAttribute (my_attr2) -> PutFile
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0" in the "consumer-client" flow
And a RouteOnAttribute processor with the name "RouteAttr1" in the "consumer-client" flow
- And the "matched_my_attr1" property of the RouteAttr1 processor is set to "${my_attr1}"
+ And RouteAttr1 is EVENT_DRIVEN in the "consumer-client" flow
+ And the "matched_my_attr1" property of the RouteAttr1 processor is set to "${my_attr1}" in the "consumer-client" flow
And a RouteOnAttribute processor with the name "RouteAttr2" in the "consumer-client" flow
- And the "matched_my_attr2" property of the RouteAttr2 processor is set to "${my_attr2}"
+ And RouteAttr2 is EVENT_DRIVEN in the "consumer-client" flow
+ And the "matched_my_attr2" property of the RouteAttr2 processor is set to "${my_attr2}" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the RouteAttr1
- And the "matched_my_attr1" relationship of the RouteAttr1 processor is connected to the RouteAttr2
- And the "matched_my_attr2" relationship of the RouteAttr2 processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the RouteAttr1
+ And in the "consumer-client" flow the "matched_my_attr1" relationship of the RouteAttr1 processor is connected to the RouteAttr2
+ And in the "consumer-client" flow the "matched_my_attr2" relationship of the RouteAttr2 processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When all instances start up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input" in the "publisher-client" flow
+ Then in the "consumer-client" container a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
Scenario: Content type - MQTT 5
# publishing MQTT client: GetFile -> PublishMQTT (Content Type: text/plain)
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "5.0"
- And the "Content Type" property of the PublishMQTT processor is set to "text/plain"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "5.0" in the "publisher-client" flow
+ And the "Content Type" property of the PublishMQTT processor is set to "text/plain" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client: ConsumeMQTT (Attribute From Content Type: content_type) -> RouteOnAttribute (content_type = text/plain) -> PutFile
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0"
- And the "Attribute From Content Type" property of the ConsumeMQTT processor is set to "content_type"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0" in the "consumer-client" flow
+ And the "Attribute From Content Type" property of the ConsumeMQTT processor is set to "content_type" in the "consumer-client" flow
And a RouteOnAttribute processor in the "consumer-client" flow
- And the "matched_content_type" property of the RouteOnAttribute processor is set to "${content_type:equals('text/plain')}"
+ And RouteOnAttribute is EVENT_DRIVEN in the "consumer-client" flow
+ And the "matched_content_type" property of the RouteOnAttribute processor is set to "${content_type:equals('text/plain')}" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the RouteOnAttribute
- And the "matched_content_type" relationship of the RouteOnAttribute processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the RouteOnAttribute
+ And in the "consumer-client" flow the "matched_content_type" relationship of the RouteOnAttribute processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
- When both instances start up
+ When all instances start up
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
And the MQTT broker has a log line matching "New client connected from .* as publisher-client"
- And a file with the content "test" is placed in "/tmp/input"
- And a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ When a file with the content "test" is placed in "/tmp/input" in the "publisher-client" flow
+ Then in the "consumer-client" container a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(4 bytes\)"
Scenario: Will content type - MQTT 5
# publishing MQTT client: GetFile -> PublishMQTT (Last Will Content Type: text/plain)
Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "publisher-client" flow
- And a PublishMQTT processor in the "publisher-client" flow
- And the "MQTT Version" property of the PublishMQTT processor is set to "5.0"
- And the "Last Will Topic" property of the PublishMQTT processor is set to "last_will_topic"
- And the "Last Will Message" property of the PublishMQTT processor is set to "last_will_message"
- And the "Last Will Content Type" property of the PublishMQTT processor is set to "text/plain"
- And the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And a PublishMQTT processor set up to communicate with an MQTT broker instance in the "publisher-client" flow
+ And PublishMQTT is EVENT_DRIVEN in the "publisher-client" flow
+ And the "MQTT Version" property of the PublishMQTT processor is set to "5.0" in the "publisher-client" flow
+ And the "Last Will Topic" property of the PublishMQTT processor is set to "last_will_topic" in the "publisher-client" flow
+ And the "Last Will Message" property of the PublishMQTT processor is set to "last_will_message" in the "publisher-client" flow
+ And the "Last Will Content Type" property of the PublishMQTT processor is set to "text/plain" in the "publisher-client" flow
+ And in the "publisher-client" flow the "success" relationship of the GetFile processor is connected to the PublishMQTT
+ And PublishMQTT's success relationship is auto-terminated in the "publisher-client" flow
# consuming MQTT client: ConsumeMQTT (Attribute From Content Type: content_type, Topic: last_will_topic) -> RouteOnAttribute (content_type = text/plain) -> PutFile
- And a ConsumeMQTT processor in the "consumer-client" flow
- And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0"
- And the "Topic" property of the ConsumeMQTT processor is set to "last_will_topic"
- And the "Attribute From Content Type" property of the ConsumeMQTT processor is set to "content_type"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance in the "consumer-client" flow
+ And the "MQTT Version" property of the ConsumeMQTT processor is set to "5.0" in the "consumer-client" flow
+ And the "Topic" property of the ConsumeMQTT processor is set to "last_will_topic" in the "consumer-client" flow
+ And the "Attribute From Content Type" property of the ConsumeMQTT processor is set to "content_type" in the "consumer-client" flow
And a RouteOnAttribute processor in the "consumer-client" flow
- And the "matched_content_type" property of the RouteOnAttribute processor is set to "${content_type:equals('text/plain')}"
+ And RouteOnAttribute is EVENT_DRIVEN in the "consumer-client" flow
+ And the "matched_content_type" property of the RouteOnAttribute processor is set to "${content_type:equals('text/plain')}" in the "consumer-client" flow
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "consumer-client" flow
- And the "success" relationship of the ConsumeMQTT processor is connected to the RouteOnAttribute
- And the "matched_content_type" relationship of the RouteOnAttribute processor is connected to the PutFile
+ And PutFile is EVENT_DRIVEN in the "consumer-client" flow
+ And in the "consumer-client" flow the "success" relationship of the ConsumeMQTT processor is connected to the RouteOnAttribute
+ And in the "consumer-client" flow the "matched_content_type" relationship of the RouteOnAttribute processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated in the "consumer-client" flow
- And an MQTT broker is set up in correspondence with the PublishMQTT and ConsumeMQTT
+ And an MQTT broker is started
When all instances start up
Then the MQTT broker has a log line matching "Sending CONNACK to publisher-client"
And the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And "publisher-client" flow is killed
- And the MQTT broker has a log line matching "Sending PUBLISH to consumer-client"
- And a flowfile with the content "last_will_message" is placed in the monitored directory in less than 60 seconds
+ When the "publisher-client" flow is killed
+ Then the MQTT broker has a log line matching "Sending PUBLISH to consumer-client"
+ And in the "consumer-client" container a single file with the content "last_will_message" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario: A MiNiFi instance uses record reader and writer to convert consumed message from an MQTT broker
Given a XMLReader controller service is set up
- And a JsonRecordSetWriter controller service is set up with "Array" output grouping
- And a ConsumeMQTT processor with the "Topic" property set to "test/my/topic"
+ And a JsonRecordSetWriter controller service is set up and the "Output Grouping" property set to "Array"
+ And a ConsumeMQTT processor set up to communicate with an MQTT broker instance
+ And the "Topic" property of the ConsumeMQTT processor is set to "test/my/topic"
And the "MQTT Version" property of the ConsumeMQTT processor is set to "3.x AUTO"
And the "Record Reader" property of the ConsumeMQTT processor is set to "XMLReader"
And the "Record Writer" property of the ConsumeMQTT processor is set to "JsonRecordSetWriter"
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And a LogAttribute processor
+ And LogAttribute is EVENT_DRIVEN
And the "success" relationship of the ConsumeMQTT processor is connected to the PutFile
And the "success" relationship of the PutFile processor is connected to the LogAttribute
- And an MQTT broker is set up in correspondence with the ConsumeMQTT
+ And PutFile's success relationship is auto-terminated
+
+ And an MQTT broker is started
- When both instances start up
+ When the MiNiFi instance starts up
And a test message "<root><element>test</element></root>" is published to the MQTT broker on topic "test/my/topic"
Then the MQTT broker has a log line matching "Received SUBSCRIBE from consumer-client"
- And a flowfile with the JSON content '[{"_isRetained": false, "_isDuplicate": false, "_qos": 0, "_topicSegments": ["test", "my", "topic"], "_topic": "test/my/topic", "element": "test"}]' is placed in the monitored directory in less than 60 seconds
+ And a file with the JSON content '[{"_isRetained": false, "_isDuplicate": false, "_qos": 0, "_topicSegments": ["test", "my", "topic"], "_topic": "test/my/topic", "element": "test"}]' is placed in the '/tmp/output' directory in less than 60 seconds
And the Minifi logs contain the following message: "key:record.count value:1" in less than 60 seconds
And the Minifi logs contain the following message: "key:mqtt.broker value:mqtt-broker-" in less than 1 seconds
Scenario: A MiNiFi instance uses record reader and writer to convert and publish records to an MQTT broker
Given a JsonTreeReader controller service is set up
And a XMLRecordSetWriter controller service is set up
- And the "Name of Record Tag" property of the XMLRecordSetWriter controller is set to "record"
- And the "Name of Root Tag" property of the XMLRecordSetWriter controller is set to "root"
+ And the "Name of Record Tag" property of the XMLRecordSetWriter controller service is set to "record"
+ And the "Name of Root Tag" property of the XMLRecordSetWriter controller service is set to "root"
And a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '[{"string": "test"}, {"int": 42}]' is present in '/tmp/input'
And a PublishMQTT processor set up to communicate with an MQTT broker instance
+ And PublishMQTT is EVENT_DRIVEN
And the "MQTT Version" property of the PublishMQTT processor is set to "3.x AUTO"
And the "Record Reader" property of the PublishMQTT processor is set to "JsonTreeReader"
And the "Record Writer" property of the PublishMQTT processor is set to "XMLRecordSetWriter"
And a UpdateAttribute processor with the "filename" property set to "${UUID()}.xml"
+ And UpdateAttribute is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishMQTT
And the "success" relationship of the PublishMQTT processor is connected to the UpdateAttribute
And the "success" relationship of the UpdateAttribute processor is connected to the PutFile
- And an MQTT broker is set up in correspondence with the PublishMQTT
+ And PutFile's success relationship is auto-terminated
- When both instances start up
+ And an MQTT broker is started
+
+ When the MiNiFi instance starts up
- Then two flowfiles with the contents '<?xml version="1.0"?><root><record><string>test</string></record></root>' and '<?xml version="1.0"?><root><record><int>42</int></record></root>' are placed in the monitored directory in less than 60 seconds
+ Then files with contents '<?xml version="1.0"?><root><record><string>test</string></record></root>' and '<?xml version="1.0"?><root><record><int>42</int></record></root>' are placed in the '/tmp/output' directory in less than 60 seconds
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(72 bytes\)"
And the MQTT broker has a log line matching "Received PUBLISH from .*testtopic.*\(64 bytes\)"
diff --git a/docker/test/integration/cluster/checkers/MqttHelper.py b/extensions/mqtt/tests/features/resources/publish_mqtt_message.py
similarity index 62%
rename from docker/test/integration/cluster/checkers/MqttHelper.py
rename to extensions/mqtt/tests/features/resources/publish_mqtt_message.py
index 719911c97..6fe709acf 100644
--- a/docker/test/integration/cluster/checkers/MqttHelper.py
+++ b/extensions/mqtt/tests/features/resources/publish_mqtt_message.py
@@ -13,11 +13,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import paho.mqtt.client as mqtt
+import sys
-class MqttHelper:
- def publish_test_mqtt_message(self, topic: str, message: str):
- client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, "docker_test_client_id")
- client.connect("localhost", 1883, 60)
- client.publish(topic, message)
- client.disconnect()
+def publish_test_mqtt_message(host: str, topic: str, message: str):
+ client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, "docker_test_client_id")
+ client.connect(host, 1883, 60)
+ client.publish(topic, message)
+ client.disconnect()
+
+
+if __name__ == "__main__":
+ if sys.argv.__len__() != 4:
+ print("Usage: publish_mqtt_message.py <host> <topic> <message>")
+ sys.exit(1)
+
+ publish_test_mqtt_message(sys.argv[1], sys.argv[2], sys.argv[3])
diff --git a/extensions/mqtt/tests/features/steps/mqtt_broker_container.py b/extensions/mqtt/tests/features/steps/mqtt_broker_container.py
new file mode 100644
index 000000000..091c12c77
--- /dev/null
+++ b/extensions/mqtt/tests/features/steps/mqtt_broker_container.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 logging
+import re
+from textwrap import dedent
+
+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.containers.docker_image_builder import DockerImageBuilder
+from docker.errors import ContainerError
+
+
+class MqttBrokerContainer(Container):
+ def __init__(self, test_context: MinifiTestContext):
+ dockerfile = dedent("""\
+ FROM {base_image}
+ RUN echo 'log_dest stderr' >> /mosquitto-no-auth.conf
+ CMD ["/usr/sbin/mosquitto", "--verbose", "--config-file", "/mosquitto-no-auth.conf"]
+ """.format(base_image='eclipse-mosquitto:2.1.1-alpine'))
+
+ builder = DockerImageBuilder(
+ image_tag="minifi-mqtt-broker:latest",
+ dockerfile_content=dockerfile
+ )
+ builder.build()
+
+ super().__init__("minifi-mqtt-broker:latest", f"mqtt-broker-{test_context.scenario_id}", test_context.network)
+
+ def deploy(self):
+ super().deploy()
+ finished_str = "mosquitto version [0-9\\.]+ running"
+ return wait_for_condition(
+ condition=lambda: re.search(finished_str, self.get_logs()),
+ timeout_seconds=60,
+ bail_condition=lambda: self.exited,
+ context=None)
+
+ def publish_mqtt_message(self, topic: str, message: str):
+ try:
+ self.client.containers.run("minifi-mqtt-helper:latest", ["python", "/scripts/publish_mqtt_message.py", self.container_name, topic, message], remove=True, stdout=True, stderr=True, network=self.network.name)
+ return True
+ except ContainerError as e:
+ stdout = e.stdout.decode("utf-8", errors="replace") if hasattr(e, "stdout") and e.stdout else ""
+ stderr = e.stderr.decode("utf-8", errors="replace") if hasattr(e, "stderr") and e.stderr else ""
+ logging.error(f"Failed to publish mqtt message in mqtt helper docker with error: '{e}', stdout: '{stdout}', stderr: '{stderr}'")
+ return False
+ except Exception as e:
+ logging.error(f"Unexpected error while publishing mqtt message in mqtt helper docker: '{e}'")
+ return False
diff --git a/extensions/mqtt/tests/features/steps/steps.py b/extensions/mqtt/tests/features/steps/steps.py
new file mode 100644
index 000000000..4bff1b2a1
--- /dev/null
+++ b/extensions/mqtt/tests/features/steps/steps.py
@@ -0,0 +1,76 @@
+# 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 re
+from behave import given, step, then, when
+
+from minifi_test_framework.steps import checking_steps # noqa: F401
+from minifi_test_framework.steps import configuration_steps # noqa: F401
+from minifi_test_framework.steps import core_steps # noqa: F401
+from minifi_test_framework.steps import flow_building_steps # noqa: F401
+from minifi_test_framework.core.minifi_test_context import DEFAULT_MINIFI_CONTAINER_NAME, MinifiTestContext
+from minifi_test_framework.minifi.processor import Processor
+from minifi_test_framework.core.helpers import wait_for_condition
+
+from mqtt_broker_container import MqttBrokerContainer
+
+
+@given("a {processor_name} processor set up to communicate with an MQTT broker instance in the \"{container_name}\" flow")
+def step_impl(context: MinifiTestContext, processor_name: str, container_name: str):
+ processor = Processor(processor_name, processor_name)
+ processor.add_property('Broker URI', f'mqtt-broker-{context.scenario_id}:1883')
+ processor.add_property('Topic', 'testtopic')
+ if processor_name == 'PublishMQTT':
+ processor.add_property('Client ID', 'publisher-client')
+ elif processor_name == 'ConsumeMQTT':
+ processor.add_property('Client ID', 'consumer-client')
+ else:
+ raise ValueError(f"Unknown processor to communicate with MQTT broker: {processor_name}")
+
+ context.get_or_create_minifi_container(container_name).flow_definition.add_processor(processor)
+
+
+@given("a {processor_name} processor set up to communicate with an MQTT broker instance")
+def step_impl(context: MinifiTestContext, processor_name: str):
+ context.execute_steps(f'given a {processor_name} processor set up to communicate with an MQTT broker instance in the "{DEFAULT_MINIFI_CONTAINER_NAME}" flow')
+
+
+@step("an MQTT broker is started")
+def step_impl(context: MinifiTestContext):
+ context.containers["mqtt-broker"] = MqttBrokerContainer(context)
+ assert context.containers["mqtt-broker"].deploy()
+
+
+@then('the MQTT broker has a log line matching "{log_line_regex}"')
+def step_impl(context: MinifiTestContext, log_line_regex: str):
+ assert wait_for_condition(
+ condition=lambda: re.search(log_line_regex, context.containers["mqtt-broker"].get_logs()),
+ timeout_seconds=60,
+ bail_condition=lambda: context.containers["mqtt-broker"].exited,
+ context=None)
+
+
+@then('the MQTT broker has {log_count:d} log lines matching "{log_line_regex}"')
+def step_impl(context: MinifiTestContext, log_count: int, log_line_regex: str):
+ assert wait_for_condition(
+ condition=lambda: len(re.findall(log_line_regex, context.containers["mqtt-broker"].get_logs())) == log_count,
+ timeout_seconds=60,
+ bail_condition=lambda: context.containers["mqtt-broker"].exited,
+ context=None)
+
+
+@when("a test message \"{message}\" is published to the MQTT broker on topic \"{topic}\"")
+def step_impl(context: MinifiTestContext, message: str, topic: str):
+ assert context.containers["mqtt-broker"].publish_mqtt_message(topic, message)
(nifi-minifi-cpp) 03/05: MINIFICPP-2256 Remove LoggerConfiguration::newInstance
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit ebc89c18690b15b5b5e61acb5be1c0389b38f6c8
Author: Ferenc Gerlits <fg...@gmail.com>
AuthorDate: Mon Jan 19 13:00:58 2026 +0100
MINIFICPP-2256 Remove LoggerConfiguration::newInstance
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #2102
---
.../include/core/logging/LoggerConfiguration.h | 9 +--
libminifi/test/libtest/unit/TestBase.cpp | 7 +-
libminifi/test/unit/LoggerTests.cpp | 84 +++++++++++-----------
3 files changed, 47 insertions(+), 53 deletions(-)
diff --git a/libminifi/include/core/logging/LoggerConfiguration.h b/libminifi/include/core/logging/LoggerConfiguration.h
index 4b4d6eb63..f9721c73f 100644
--- a/libminifi/include/core/logging/LoggerConfiguration.h
+++ b/libminifi/include/core/logging/LoggerConfiguration.h
@@ -71,15 +71,9 @@ class LoggerConfiguration {
friend class ::LogTestController;
public:
- /**
- * Gets the current log configuration
- */
+ LoggerConfiguration(); // used in some tests only
static LoggerConfiguration& getConfiguration();
- static std::unique_ptr<LoggerConfiguration> newInstance() {
- return std::unique_ptr<LoggerConfiguration>(new LoggerConfiguration()); // NOLINT(modernize-make-unique, cppcoreguidelines-owning-memory
- }
-
void disableLogging() {
controller_->setEnabled(false);
}
@@ -163,7 +157,6 @@ class LoggerConfiguration {
static std::shared_ptr<spdlog::sinks::rotating_file_sink_mt> getRotatingFileSink(const std::string& appender_key, const std::shared_ptr<LoggerProperties>& properties);
- LoggerConfiguration();
internal::CompressionManager compression_manager_;
std::shared_ptr<internal::LoggerNamespace> root_namespace_;
diff --git a/libminifi/test/libtest/unit/TestBase.cpp b/libminifi/test/libtest/unit/TestBase.cpp
index 0cf593b34..b4a58191f 100644
--- a/libminifi/test/libtest/unit/TestBase.cpp
+++ b/libminifi/test/libtest/unit/TestBase.cpp
@@ -158,8 +158,9 @@ void LogTestController::reset() {
setLevel(name, spdlog::level::err);
}
modified_loggers.clear();
- if (config)
- config = logging::LoggerConfiguration::newInstance();
+ if (config) {
+ config = std::make_unique<logging::LoggerConfiguration>();
+ }
my_properties_ = std::make_shared<logging::LoggerProperties>("");
clear();
init(nullptr);
@@ -195,7 +196,7 @@ void LogTestController::init(const std::shared_ptr<logging::LoggerProperties>& l
logging::LoggerConfiguration::getConfiguration().initialize(my_properties_);
logger_ = logging::LoggerConfiguration::getConfiguration().getLogger(minifi::core::className<LogTestController>());
} else {
- config = logging::LoggerConfiguration::newInstance();
+ config = std::make_unique<logging::LoggerConfiguration>();
// create for test purposes. most tests use the main logging factory, but this exists to test the logging
// framework itself.
config->initialize(my_properties_);
diff --git a/libminifi/test/unit/LoggerTests.cpp b/libminifi/test/unit/LoggerTests.cpp
index 1326c7ddf..b2276600d 100644
--- a/libminifi/test/unit/LoggerTests.cpp
+++ b/libminifi/test/unit/LoggerTests.cpp
@@ -228,7 +228,7 @@ class LoggerTestAccessor {
};
TEST_CASE("Test Compression", "[ttl9]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
std::string className;
SECTION("Using root logger") {
@@ -241,38 +241,38 @@ TEST_CASE("Test Compression", "[ttl9]") {
properties->set("appender.null", "null");
properties->set("logger." + className, "INFO,null");
}
- log_config->initialize(properties);
- auto logger = log_config->getLogger(className);
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger(className);
logger->log_error("Hi there");
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == 1);
auto logs = decompress(compressed_logs[0]);
REQUIRE(logs.find("Hi there") != std::string::npos);
}
TEST_CASE("Test Compression cache overflow is discarded intermittently", "[ttl10]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
properties->set(logging::internal::CompressionManager::compression_cached_log_max_size_, "10 KB");
- LoggerTestAccessor::setCompressionCacheSegmentSize(*log_config, 1_KiB);
+ LoggerTestAccessor::setCompressionCacheSegmentSize(log_config, 1_KiB);
std::string className = "CompressionTestCacheCleaned";
// by default the root logger is OFF
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger(className);
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger(className);
for (size_t idx = 0; idx < 10000; ++idx) {
logger->log_error("Hi there");
}
bool cache_shrunk = minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1}, [&] {
- return LoggerTestAccessor::getUncompressedSize(*log_config) <= 10_KiB;
+ return LoggerTestAccessor::getUncompressedSize(log_config) <= 10_KiB;
});
REQUIRE(cache_shrunk);
}
TEST_CASE("Setting either properties to 0 disables in-memory compressed logs", "[ttl11]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
bool is_empty = false;
SECTION("Cached log size is set to 0") {
@@ -289,26 +289,26 @@ TEST_CASE("Setting either properties to 0 disables in-memory compressed logs", "
}
// by default the root logger is OFF
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger("DisableCompressionTestLogger");
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger("DisableCompressionTestLogger");
logger->log_error("Hi there");
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.empty() == is_empty);
}
TEST_CASE("Setting max log entry length property trims long log entries", "[ttl12]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
properties->set("max.log.entry.length", "2");
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger("SetMaxLogEntryLengthTestLogger");
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger("SetMaxLogEntryLengthTestLogger");
logger->log_error("Hi there");
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == 1);
auto logs = decompress(compressed_logs[0]);
REQUIRE(logs.find("Hi ") == std::string::npos);
@@ -316,16 +316,16 @@ TEST_CASE("Setting max log entry length property trims long log entries", "[ttl1
}
TEST_CASE("Setting max log entry length property trims long formatted log entries", "[ttl13]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
properties->set("max.log.entry.length", "2");
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger("SetMaxLogEntryLengthInFormattedTestLogger");
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger("SetMaxLogEntryLengthInFormattedTestLogger");
logger->log_error("Hi there {}", "John");
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == 1);
auto logs = decompress(compressed_logs[0]);
REQUIRE(logs.find("Hi ") == std::string::npos);
@@ -333,18 +333,18 @@ TEST_CASE("Setting max log entry length property trims long formatted log entrie
}
TEST_CASE("Setting max log entry length to a size larger than the internal buffer size", "[ttl14]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
properties->set("max.log.entry.length", "1500");
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger("SetMaxLogEntryLengthToLargerThanBufferSizeTestLogger");
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger("SetMaxLogEntryLengthToLargerThanBufferSizeTestLogger");
std::string log(2000, 'a');
std::string expected_log(1500, 'a');
logger->log_error("{}", log);
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == 1);
auto logs = decompress(compressed_logs[0]);
REQUIRE(logs.find(log) == std::string::npos);
@@ -352,7 +352,7 @@ TEST_CASE("Setting max log entry length to a size larger than the internal buffe
}
TEST_CASE("Setting max log entry length to unlimited results in unlimited log entry size", "[ttl15]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
+ logging::LoggerConfiguration log_config;
auto properties = std::make_shared<logging::LoggerProperties>("");
std::string_view logger_name;
SECTION("Use unlimited value") {
@@ -364,13 +364,13 @@ TEST_CASE("Setting max log entry length to unlimited results in unlimited log en
logger_name = "SetMaxLogEntryLengthTo-1TestLogger";
}
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger(logger_name);
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger(logger_name);
std::string log(5000, 'a');
logger->log_error("{}", log);
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == 1);
auto logs = decompress(compressed_logs[0]);
REQUIRE(logs.find(log) != std::string::npos);
@@ -417,22 +417,22 @@ std::vector<std::string> generateRandomStrings() {
}
TEST_CASE("Test sending multiple segments at once", "[ttl16]") {
- auto log_config = logging::LoggerConfiguration::newInstance();
- LoggerTestAccessor::setCompressionCompressedSegmentSize(*log_config, 100);
- LoggerTestAccessor::setCompressionCacheSegmentSize(*log_config, 100);
+ logging::LoggerConfiguration log_config;
+ LoggerTestAccessor::setCompressionCompressedSegmentSize(log_config, 100);
+ LoggerTestAccessor::setCompressionCacheSegmentSize(log_config, 100);
auto properties = std::make_shared<logging::LoggerProperties>("");
// by default the root logger is OFF
properties->set("logger.root", "INFO");
- log_config->initialize(properties);
- auto logger = log_config->getLogger("CompressionTestMultiSegment");
+ log_config.initialize(properties);
+ auto logger = log_config.getLogger("CompressionTestMultiSegment");
const auto random_strings = generateRandomStrings();
for (const auto& random_string : random_strings) {
logger->log_error("{}", random_string);
}
- REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(*log_config));
- auto compressed_logs = LoggerTestAccessor::getCompressedLogs(*log_config);
+ REQUIRE(LoggerTestAccessor::waitForCompressionToHappen(log_config));
+ auto compressed_logs = LoggerTestAccessor::getCompressedLogs(log_config);
REQUIRE(compressed_logs.size() == random_strings.size());
for (size_t i = 0; i < compressed_logs.size(); ++i) {
auto logs = decompress(compressed_logs[i]);
(nifi-minifi-cpp) 05/05: MINIFICPP-2724 Add MinifiStatus return type to C api function that might fail
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3190ca615b172ea4f08577f99fa7b9f6130cd4be
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu Feb 19 13:00:01 2026 +0100
MINIFICPP-2724 Add MinifiStatus return type to C api function that might fail
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #2113
---
.../include/api/core/ProcessContext.h | 2 --
.../include/api/utils/ProcessorConfigUtils.h | 24 ++++++++---------
.../cpp-extension-lib/src/core/ProcessContext.cpp | 10 -------
.../cpp-extension-lib/src/core/ProcessSession.cpp | 22 ++++++++++-----
.../include/utils/ProcessorConfigUtils.h | 24 ++++++++---------
.../azure/tests/FetchAzureDataLakeStorageTests.cpp | 2 +-
.../llamacpp/tests/RunLlamaCppInferenceTests.cpp | 4 +--
extensions/mqtt/tests/ConsumeMQTTTests.cpp | 4 +--
extensions/mqtt/tests/PublishMQTTTests.cpp | 2 +-
.../tests/unit/SegmentContentTests.cpp | 4 +--
.../tests/unit/SplitContentTests.cpp | 2 +-
.../tests/unit/SplitTextTests.cpp | 2 +-
libminifi/src/minifi-c.cpp | 31 +++++++++++++---------
minifi-api/include/minifi-c/minifi-c.h | 7 +++--
minifi-api/minifi-c-api.def | 1 -
15 files changed, 71 insertions(+), 70 deletions(-)
diff --git a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
index 59eb52e02..50f5ebaa5 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -37,8 +37,6 @@ class ProcessContext {
bool hasNonEmptyProperty(std::string_view name) const;
- std::string getProcessorName() const;
-
private:
MinifiProcessContext* impl_;
};
diff --git a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
index 14644b9c7..6a14699d7 100644
--- a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
@@ -33,37 +33,37 @@ namespace org::apache::nifi::minifi::api::utils {
inline std::string parseProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
- | minifi::utils::orThrow(fmt::format("Expected valid value from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected valid value from \"{}\"", property.name));
}
inline bool parseBoolProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| minifi::utils::andThen(parsing::parseBool)
- | minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
inline uint64_t parseU64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| minifi::utils::andThen(parsing::parseIntegral<uint64_t>)
- | minifi::utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}\"", property.name));
}
inline int64_t parseI64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| minifi::utils::andThen(parsing::parseIntegral<int64_t>)
- | minifi::utils::orThrow(fmt::format("Expected parsable int64_t from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable int64_t from \"{}\"", property.name));
}
inline std::chrono::milliseconds parseDurationProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| minifi::utils::andThen(parsing::parseDuration<std::chrono::milliseconds>)
- | minifi::utils::orThrow(fmt::format("Expected parsable duration from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable duration from \"{}\"", property.name));
}
inline uint64_t parseDataSizeProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| minifi::utils::andThen(parsing::parseDataSize)
- | minifi::utils::orThrow(fmt::format("Expected parsable data size from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable data size from \"{}\"", property.name));
}
inline std::optional<std::string> parseOptionalProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
@@ -74,7 +74,7 @@ inline std::optional<std::string> parseOptionalProperty(const core::ProcessConte
inline std::optional<bool> parseOptionalBoolProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
return parsing::parseBool(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
return std::nullopt;
}
@@ -85,7 +85,7 @@ inline std::optional<uint64_t> parseOptionalU64Property(const core::ProcessConte
return std::nullopt;
}
return parsing::parseIntegral<uint64_t>(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}\"", property.name));
}
return std::nullopt;
@@ -97,7 +97,7 @@ inline std::optional<int64_t> parseOptionalI64Property(const core::ProcessContex
return std::nullopt;
}
return parsing::parseIntegral<int64_t>(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable int64_t from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable int64_t from \"{}\"", property.name));
}
return std::nullopt;
@@ -111,7 +111,7 @@ inline std::optional<std::chrono::milliseconds> parseOptionalDurationProperty(co
return std::nullopt;
}
return parsing::parseDuration(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable duration from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable duration from \"{}\"", property.name));
}
return std::nullopt;
@@ -123,7 +123,7 @@ inline std::optional<uint64_t> parseOptionalDataSizeProperty(const core::Process
return std::nullopt;
}
return parsing::parseDataSize(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable data size from \"{}::{}\"", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable data size from \"{}\"", property.name));
}
return std::nullopt;
@@ -135,7 +135,7 @@ inline std::optional<float> parseOptionalFloatProperty(const core::ProcessContex
return std::nullopt;
}
return parsing::parseFloat(*property_str)
- | minifi::utils::orThrow(fmt::format("Expected parsable float from {}::{}", ctx.getProcessorName(), property.name));
+ | minifi::utils::orThrow(fmt::format("Expected parsable float from \"{}\"", property.name));
}
return std::nullopt;
}
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
index 76ed2231b..58016d804 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -38,14 +38,4 @@ bool ProcessContext::hasNonEmptyProperty(std::string_view name) const {
return MinifiProcessContextHasNonEmptyProperty(impl_, utils::toStringView(name));
}
-std::string ProcessContext::getProcessorName() const {
- std::string result;
- MinifiProcessContextGetProcessorName(impl_, [] (void* data, MinifiStringView name) {
- *static_cast<std::string*>(data) = std::string(name.data, name.length);
- }, &result);
- return result;
-}
-
-
-
} // namespace org::apache::nifi::minifi::api::core
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
index c5c51765c..6b22c0bbc 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
@@ -79,11 +79,15 @@ FlowFile ProcessSession::create(const FlowFile* parent) {
void ProcessSession::transfer(FlowFile ff, const minifi::core::Relationship& relationship) {
const auto rel_name = relationship.getName();
- MinifiProcessSessionTransfer(impl_, ff.release(), utils::toStringView(rel_name));
+ if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionTransfer(impl_, ff.release(), utils::toStringView(rel_name))) {
+ throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to transfer flowfile");
+ }
}
void ProcessSession::remove(FlowFile ff) {
- MinifiProcessSessionRemove(impl_, ff.release());
+ if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionRemove(impl_, ff.release())) {
+ throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove flowfile");
+ }
}
void ProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) {
@@ -104,13 +108,17 @@ void ProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback& ca
}
}
-void ProcessSession::setAttribute(FlowFile& ff, std::string_view key, std::string value) { // NOLINT(performance-unnecessary-value-param)
- MinifiStringView value_ref = utils::toStringView(value);
- MinifiFlowFileSetAttribute(impl_, ff.get(), utils::toStringView(key), &value_ref);
+void ProcessSession::setAttribute(FlowFile& ff, const std::string_view key, std::string value) { // NOLINT(performance-unnecessary-value-param)
+ const MinifiStringView value_ref = utils::toStringView(value);
+ if (MINIFI_STATUS_SUCCESS != MinifiFlowFileSetAttribute(impl_, ff.get(), utils::toStringView(key), &value_ref)) {
+ throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to set attribute");
+ }
}
-void ProcessSession::removeAttribute(FlowFile& ff, std::string_view key) {
- MinifiFlowFileSetAttribute(impl_, ff.get(), utils::toStringView(key), nullptr);
+void ProcessSession::removeAttribute(FlowFile& ff, const std::string_view key) {
+ if (MINIFI_STATUS_SUCCESS != MinifiFlowFileSetAttribute(impl_, ff.get(), utils::toStringView(key), nullptr)) {
+ throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove attribute");
+ }
}
std::optional<std::string> ProcessSession::getAttribute(FlowFile& ff, std::string_view key) {
diff --git a/extension-framework/include/utils/ProcessorConfigUtils.h b/extension-framework/include/utils/ProcessorConfigUtils.h
index 148435df8..3177863df 100644
--- a/extension-framework/include/utils/ProcessorConfigUtils.h
+++ b/extension-framework/include/utils/ProcessorConfigUtils.h
@@ -34,37 +34,37 @@ namespace org::apache::nifi::minifi::utils {
inline std::string parseProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
- | orThrow(fmt::format("Expected valid value from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected valid value from \"{}\"", property.name));
}
inline bool parseBoolProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| andThen(parsing::parseBool)
- | orThrow(fmt::format("Expected parsable bool from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
inline uint64_t parseU64Property(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| andThen(parsing::parseIntegral<uint64_t>)
- | orThrow(fmt::format("Expected parsable uint64_t from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected parsable uint64_t from \"{}\"", property.name));
}
inline int64_t parseI64Property(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| andThen(parsing::parseIntegral<int64_t>)
- | orThrow(fmt::format("Expected parsable int64_t from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected parsable int64_t from \"{}\"", property.name));
}
inline std::chrono::milliseconds parseDurationProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| andThen(parsing::parseDuration<std::chrono::milliseconds>)
- | orThrow(fmt::format("Expected parsable duration from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected parsable duration from \"{}\"", property.name));
}
inline uint64_t parseDataSizeProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
return ctx.getProperty(property.name, flow_file)
| andThen(parsing::parseDataSize)
- | orThrow(fmt::format("Expected parsable data size from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | orThrow(fmt::format("Expected parsable data size from \"{}\"", property.name));
}
inline std::optional<std::string> parseOptionalProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
@@ -75,7 +75,7 @@ inline std::optional<std::string> parseOptionalProperty(const core::ProcessConte
inline std::optional<bool> parseOptionalBoolProperty(const core::ProcessContext& ctx, const core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
return parsing::parseBool(*property_str)
- | utils::orThrow(fmt::format("Expected parsable bool from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
return std::nullopt;
}
@@ -86,7 +86,7 @@ inline std::optional<uint64_t> parseOptionalU64Property(const core::ProcessConte
return std::nullopt;
}
return parsing::parseIntegral<uint64_t>(*property_str)
- | utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}\"", property.name));
}
return std::nullopt;
@@ -98,7 +98,7 @@ inline std::optional<int64_t> parseOptionalI64Property(const core::ProcessContex
return std::nullopt;
}
return parsing::parseIntegral<int64_t>(*property_str)
- | utils::orThrow(fmt::format("Expected parsable int64_t from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable int64_t from \"{}\"", property.name));
}
return std::nullopt;
@@ -110,7 +110,7 @@ inline std::optional<std::chrono::milliseconds> parseOptionalDurationProperty(co
return std::nullopt;
}
return parsing::parseDuration(*property_str)
- | utils::orThrow(fmt::format("Expected parsable duration from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable duration from \"{}\"", property.name));
}
return std::nullopt;
@@ -122,7 +122,7 @@ inline std::optional<uint64_t> parseOptionalDataSizeProperty(const core::Process
return std::nullopt;
}
return parsing::parseDataSize(*property_str)
- | utils::orThrow(fmt::format("Expected parsable data size from \"{}::{}\"", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable data size from \"{}\"", property.name));
}
return std::nullopt;
@@ -134,7 +134,7 @@ inline std::optional<float> parseOptionalFloatProperty(const core::ProcessContex
return std::nullopt;
}
return parsing::parseFloat(*property_str)
- | utils::orThrow(fmt::format("Expected parsable float from {}::{}", ctx.getProcessorInfo().getName(), property.name));
+ | utils::orThrow(fmt::format("Expected parsable float from \"{}\"", property.name));
}
return std::nullopt;
}
diff --git a/extensions/azure/tests/FetchAzureDataLakeStorageTests.cpp b/extensions/azure/tests/FetchAzureDataLakeStorageTests.cpp
index c5505f49e..af0016141 100644
--- a/extensions/azure/tests/FetchAzureDataLakeStorageTests.cpp
+++ b/extensions/azure/tests/FetchAzureDataLakeStorageTests.cpp
@@ -157,7 +157,7 @@ TEST_CASE_METHOD(FetchAzureDataLakeStorageTestsFixture, "Valid Number of Retries
TEST_CASE_METHOD(FetchAzureDataLakeStorageTestsFixture, "Invalid Number of Retries is set via EL", "[azureDataLakeStorageFetch]") {
plan_->setProperty(azure_data_lake_storage_, minifi::azure::processors::FetchAzureDataLakeStorage::NumberOfRetries, "${literal(\"asd\")}");
- REQUIRE_THROWS_WITH(test_controller_.runSession(plan_, true), "Expected parsable uint64_t from \"AzureDataLakeStorageProcessor::Number of Retries\", but got GeneralParsingError (Parsing Error:0)");
+ REQUIRE_THROWS_WITH(test_controller_.runSession(plan_, true), "Expected parsable uint64_t from \"Number of Retries\", but got GeneralParsingError (Parsing Error:0)");
CHECK_FALSE(mock_data_lake_storage_client_ptr_->getPassedFetchParams().number_of_retries);
}
diff --git a/extensions/llamacpp/tests/RunLlamaCppInferenceTests.cpp b/extensions/llamacpp/tests/RunLlamaCppInferenceTests.cpp
index ec507af33..13754f1f5 100644
--- a/extensions/llamacpp/tests/RunLlamaCppInferenceTests.cpp
+++ b/extensions/llamacpp/tests/RunLlamaCppInferenceTests.cpp
@@ -229,7 +229,7 @@ TEST_CASE("Invalid values for optional double type properties throw exception")
REQUIRE_THROWS(controller.trigger(minifi::test::InputFlowFileData{.content = "42", .attributes = {}}));
CHECK(minifi::test::utils::verifyLogLinePresenceInPollTime(1s,
- fmt::format("Expected parsable float from RunLlamaCppInference::{}, but got GeneralParsingError (Parsing Error:0)", property_name)));
+ fmt::format("Expected parsable float from \"{}\", but got GeneralParsingError (Parsing Error:0)", property_name)));
}
TEST_CASE("Top K property empty and invalid values are handled properly") {
@@ -252,7 +252,7 @@ TEST_CASE("Top K property empty and invalid values are handled properly") {
REQUIRE(controller.getProcessor()->setProperty(processors::RunLlamaCppInference::TopK.name, "invalid_value"));
REQUIRE_THROWS(controller.trigger(minifi::test::InputFlowFileData{.content = "42", .attributes = {}}));
CHECK(minifi::test::utils::verifyLogLinePresenceInPollTime(1s,
- "Expected parsable int64_t from \"RunLlamaCppInference::Top K\", but got GeneralParsingError (Parsing Error:0)"));
+ "Expected parsable int64_t from \"Top K\", but got GeneralParsingError (Parsing Error:0)"));
}
}
diff --git a/extensions/mqtt/tests/ConsumeMQTTTests.cpp b/extensions/mqtt/tests/ConsumeMQTTTests.cpp
index 8b02e2df1..8b63b1aec 100644
--- a/extensions/mqtt/tests/ConsumeMQTTTests.cpp
+++ b/extensions/mqtt/tests/ConsumeMQTTTests.cpp
@@ -120,13 +120,13 @@ using namespace std::literals::chrono_literals;
TEST_CASE_METHOD(ConsumeMqttTestFixture, "ConsumeMQTTTest_EmptyTopic", "[consumeMQTTTest]") {
REQUIRE(test_controller_.plan->setProperty(consume_mqtt_processor_, minifi::processors::AbstractMQTTProcessor::BrokerURI.name, "127.0.0.1:1883"));
REQUIRE_THROWS_WITH(test_controller_.plan->scheduleProcessor(consume_mqtt_processor_),
- Catch::Matchers::EndsWith("Expected valid value from \"TestConsumeMQTTProcessor::Topic\", but got PropertyNotSet (Property Error:2)"));
+ Catch::Matchers::EndsWith("Expected valid value from \"Topic\", but got PropertyNotSet (Property Error:2)"));
}
TEST_CASE_METHOD(ConsumeMqttTestFixture, "ConsumeMQTTTest_EmptyBrokerURI", "[consumeMQTTTest]") {
REQUIRE(test_controller_.plan->setProperty(consume_mqtt_processor_, minifi::processors::ConsumeMQTT::Topic.name, "mytopic"));
REQUIRE_THROWS_WITH(test_controller_.plan->scheduleProcessor(consume_mqtt_processor_),
- Catch::Matchers::EndsWith("Expected valid value from \"TestConsumeMQTTProcessor::Broker URI\", but got PropertyNotSet (Property Error:2)"));
+ Catch::Matchers::EndsWith("Expected valid value from \"Broker URI\", but got PropertyNotSet (Property Error:2)"));
}
TEST_CASE_METHOD(ConsumeMqttTestFixture, "ConsumeMQTTTest_DurableSessionWithID", "[consumeMQTTTest]") {
diff --git a/extensions/mqtt/tests/PublishMQTTTests.cpp b/extensions/mqtt/tests/PublishMQTTTests.cpp
index 98ff17280..62205ae84 100644
--- a/extensions/mqtt/tests/PublishMQTTTests.cpp
+++ b/extensions/mqtt/tests/PublishMQTTTests.cpp
@@ -80,7 +80,7 @@ TEST_CASE_METHOD(PublishMQTTTestFixture, "PublishMQTTTest_EmptyTopic", "[publish
TEST_CASE_METHOD(PublishMQTTTestFixture, "PublishMQTTTest_EmptyBrokerURI", "[publishMQTTTest]") {
REQUIRE(test_controller_.plan->setProperty(publish_mqtt_processor_, minifi::processors::PublishMQTT::Topic.name, "mytopic"));
REQUIRE_THROWS_WITH(test_controller_.plan->scheduleProcessor(publish_mqtt_processor_),
- Catch::Matchers::EndsWith("Expected valid value from \"TestPublishMQTTProcessor::Broker URI\", but got PropertyNotSet (Property Error:2)"));
+ Catch::Matchers::EndsWith("Expected valid value from \"Broker URI\", but got PropertyNotSet (Property Error:2)"));
}
TEST_CASE_METHOD(PublishMQTTTestFixture, "PublishMQTTTest_EmptyClientID_V_3", "[publishMQTTTest]") {
diff --git a/extensions/standard-processors/tests/unit/SegmentContentTests.cpp b/extensions/standard-processors/tests/unit/SegmentContentTests.cpp
index bb74db80d..6d86feadb 100644
--- a/extensions/standard-processors/tests/unit/SegmentContentTests.cpp
+++ b/extensions/standard-processors/tests/unit/SegmentContentTests.cpp
@@ -62,11 +62,11 @@ TEST_CASE("Invalid segmentSize tests") {
SECTION("foo") {
REQUIRE(segment_content->setProperty(SegmentContent::SegmentSize.name, "foo"));
- REQUIRE_THROWS_WITH(controller.trigger("bar"), "Expected parsable data size from \"SegmentContent::Segment Size\", but got GeneralParsingError (Parsing Error:0)");
+ REQUIRE_THROWS_WITH(controller.trigger("bar"), "Expected parsable data size from \"Segment Size\", but got GeneralParsingError (Parsing Error:0)");
}
SECTION("10 foo") {
REQUIRE(segment_content->setProperty(SegmentContent::SegmentSize.name, "10 foo"));
- REQUIRE_THROWS_WITH(controller.trigger("bar"), "Expected parsable data size from \"SegmentContent::Segment Size\", but got GeneralParsingError (Parsing Error:0)");
+ REQUIRE_THROWS_WITH(controller.trigger("bar"), "Expected parsable data size from \"Segment Size\", but got GeneralParsingError (Parsing Error:0)");
}
SECTION("0") {
REQUIRE(segment_content->setProperty(SegmentContent::SegmentSize.name, "0"));
diff --git a/extensions/standard-processors/tests/unit/SplitContentTests.cpp b/extensions/standard-processors/tests/unit/SplitContentTests.cpp
index 57a13e5c7..d24edc9b5 100644
--- a/extensions/standard-processors/tests/unit/SplitContentTests.cpp
+++ b/extensions/standard-processors/tests/unit/SplitContentTests.cpp
@@ -39,7 +39,7 @@ TEST_CASE("WithoutByteSequence") {
REQUIRE(split_content->setProperty(SplitContent::KeepByteSequence.name, "true"));
REQUIRE(split_content->setProperty(SplitContent::ByteSequenceLocationProperty.name, std::string{magic_enum::enum_name(SplitContent::ByteSequenceLocation::Leading)}));
- REQUIRE_THROWS_WITH(controller.trigger("rub-a-dub-dub"), "Expected valid value from \"SplitContent::Byte Sequence\", but got PropertyNotSet (Property Error:2)");
+ REQUIRE_THROWS_WITH(controller.trigger("rub-a-dub-dub"), "Expected valid value from \"Byte Sequence\", but got PropertyNotSet (Property Error:2)");
}
TEST_CASE("EmptyFlowFile") {
diff --git a/extensions/standard-processors/tests/unit/SplitTextTests.cpp b/extensions/standard-processors/tests/unit/SplitTextTests.cpp
index 53c437627..f3ae69673 100644
--- a/extensions/standard-processors/tests/unit/SplitTextTests.cpp
+++ b/extensions/standard-processors/tests/unit/SplitTextTests.cpp
@@ -193,7 +193,7 @@ void runSplitTextTest(const std::string& input, const std::vector<ExpectedSplitT
TEST_CASE("Line Split Count property is required") {
SingleProcessorTestController controller{minifi::test::utils::make_processor<processors::SplitText>("SplitText")};
- REQUIRE_THROWS_WITH(controller.trigger("", {}), "Expected parsable uint64_t from \"SplitText::Line Split Count\", but got PropertyNotSet (Property Error:2)");
+ REQUIRE_THROWS_WITH(controller.trigger("", {}), "Expected parsable uint64_t from \"Line Split Count\", but got PropertyNotSet (Property Error:2)");
}
TEST_CASE("Line Split Count property can only be 0 if Maximum Fragment Size is set") {
diff --git a/libminifi/src/minifi-c.cpp b/libminifi/src/minifi-c.cpp
index f14847363..ebceb9f02 100644
--- a/libminifi/src/minifi-c.cpp
+++ b/libminifi/src/minifi-c.cpp
@@ -267,12 +267,6 @@ MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, Mini
}
}
-void MinifiProcessContextGetProcessorName(MinifiProcessContext* context, void(*cb)(void* user_ctx, MinifiStringView result), void* user_ctx) {
- gsl_Assert(context != MINIFI_NULL);
- auto name = reinterpret_cast<minifi::core::ProcessContext*>(context)->getProcessorInfo().getName();
- cb(user_ctx, MinifiStringView{.data = name.data(), .length = name.length()});
-}
-
MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext* context, MinifiStringView property_name) {
gsl_Assert(context != MINIFI_NULL);
return reinterpret_cast<minifi::core::ProcessContext*>(context)->hasNonEmptyProperty(toString(property_name));
@@ -345,15 +339,27 @@ MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionCreate(MinifiProcessSession* se
return MINIFI_NULL;
}
-void MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name) {
+MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name) {
+ gsl_Assert(session != MINIFI_NULL);
gsl_Assert(flowfile != MINIFI_NULL);
- reinterpret_cast<minifi::core::ProcessSession*>(session)->transfer(
- *reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), minifi::core::Relationship{toString(relationship_name), ""});
+ try {
+ reinterpret_cast<minifi::core::ProcessSession*>(session)->transfer(
+ *reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), minifi::core::Relationship{toString(relationship_name), ""});
+ return MINIFI_STATUS_SUCCESS;
+ } catch (...) {
+ return MINIFI_STATUS_UNKNOWN_ERROR;
+ }
}
-void MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile) {
+MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile) {
+ gsl_Assert(session != MINIFI_NULL);
gsl_Assert(flowfile != MINIFI_NULL);
- reinterpret_cast<minifi::core::ProcessSession*>(session)->remove(*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile));
+ try {
+ reinterpret_cast<minifi::core::ProcessSession*>(session)->remove(*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile));
+ return MINIFI_STATUS_SUCCESS;
+ } catch (...) {
+ return MINIFI_STATUS_UNKNOWN_ERROR;
+ }
}
MinifiStatus MinifiProcessSessionRead(MinifiProcessSession* session, MinifiFlowFile* flowfile, int64_t(*cb)(void* user_ctx, MinifiInputStream*), void* user_ctx) {
@@ -412,7 +418,7 @@ void MinifiStatusToString(MinifiStatus status, void(*cb)(void* user_ctx, MinifiS
cb(user_ctx, MinifiStringView{.data = message.data(), .length = message.size()});
}
-void MinifiFlowFileSetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value) {
+MinifiStatus MinifiFlowFileSetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value) {
gsl_Assert(session != MINIFI_NULL);
gsl_Assert(flowfile != MINIFI_NULL);
if (attribute_value == nullptr) {
@@ -421,6 +427,7 @@ void MinifiFlowFileSetAttribute(MinifiProcessSession* session, MinifiFlowFile* f
reinterpret_cast<minifi::core::ProcessSession*>(session)->putAttribute(
**reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), toString(attribute_name), toString(*attribute_value));
}
+ return MINIFI_STATUS_SUCCESS;
}
MinifiBool MinifiFlowFileGetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name,
diff --git a/minifi-api/include/minifi-c/minifi-c.h b/minifi-api/include/minifi-c/minifi-c.h
index e92814b37..5be2bbb2f 100644
--- a/minifi-api/include/minifi-c/minifi-c.h
+++ b/minifi-api/include/minifi-c/minifi-c.h
@@ -190,7 +190,6 @@ MINIFI_OWNED MinifiPublishedMetrics* MinifiPublishedMetricsCreate(size_t count,
MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, MinifiStringView property_name, MinifiFlowFile* flowfile,
void(*cb)(void* user_ctx, MinifiStringView property_value), void* user_ctx);
-void MinifiProcessContextGetProcessorName(MinifiProcessContext* context, void(*cb)(void* user_ctx, MinifiStringView processor_name), void* user_ctx);
MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext* context, MinifiStringView property_name);
void MinifiLoggerSetMaxLogSize(MinifiLogger*, int32_t);
@@ -201,8 +200,8 @@ MinifiLogLevel MinifiLoggerLevel(MinifiLogger*);
MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionGet(MinifiProcessSession*);
MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionCreate(MinifiProcessSession* session, MinifiFlowFile* parent_flowfile);
-void MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name);
-void MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile);
+MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name);
+MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile);
MinifiStatus MinifiProcessSessionRead(MinifiProcessSession*, MinifiFlowFile*, int64_t(*cb)(void* user_ctx, MinifiInputStream*), void* user_ctx);
MinifiStatus MinifiProcessSessionWrite(MinifiProcessSession*, MinifiFlowFile*, int64_t(*cb)(void* user_ctx, MinifiOutputStream*), void* user_ctx);
@@ -216,7 +215,7 @@ int64_t MinifiOutputStreamWrite(MinifiOutputStream* stream, const char* data, si
void MinifiStatusToString(MinifiStatus, void(*cb)(void* user_ctx, MinifiStringView str), void* user_ctx);
-void MinifiFlowFileSetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value);
+MinifiStatus MinifiFlowFileSetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value);
MinifiBool MinifiFlowFileGetAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name,
void(*cb)(void* user_ctx, MinifiStringView attribute_value), void* user_ctx);
void MinifiFlowFileGetAttributes(MinifiProcessSession* session, MinifiFlowFile* flowfile, void(*cb)(void* user_ctx, MinifiStringView attribute_name, MinifiStringView attribute_value), void* user_ctx);
diff --git a/minifi-api/minifi-c-api.def b/minifi-api/minifi-c-api.def
index 18fad4bff..ee225e7c1 100644
--- a/minifi-api/minifi-c-api.def
+++ b/minifi-api/minifi-c-api.def
@@ -3,7 +3,6 @@ EXPORTS
MinifiCreateExtension
MinifiPublishedMetricsCreate
MinifiProcessContextGetProperty
- MinifiProcessContextGetProcessorName
MinifiProcessContextHasNonEmptyProperty
MinifiLoggerSetMaxLogSize
MinifiLoggerLogString
(nifi-minifi-cpp) 04/05: MINIFICPP-2722 C API enum explicit type
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3b133464974174e52e2fe17a5f3559008aff8dac
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Feb 18 11:25:11 2026 +0100
MINIFICPP-2722 C API enum explicit type
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #2111
---
minifi-api/include/minifi-c/minifi-c.h | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/minifi-api/include/minifi-c/minifi-c.h b/minifi-api/include/minifi-c/minifi-c.h
index 1f22c908c..e92814b37 100644
--- a/minifi-api/include/minifi-c/minifi-c.h
+++ b/minifi-api/include/minifi-c/minifi-c.h
@@ -41,7 +41,7 @@ extern "C" {
typedef bool MinifiBool;
-typedef enum MinifiInputRequirement {
+typedef enum MinifiInputRequirement : uint32_t {
MINIFI_INPUT_REQUIRED = 0,
MINIFI_INPUT_ALLOWED = 1,
MINIFI_INPUT_FORBIDDEN = 2
@@ -88,7 +88,7 @@ typedef struct MinifiConfig MinifiConfig;
typedef struct MinifiExtension MinifiExtension;
typedef struct MinifiPublishedMetrics MinifiPublishedMetrics;
-typedef enum MinifiStatus {
+typedef enum MinifiStatus : uint32_t {
MINIFI_STATUS_SUCCESS = 0,
MINIFI_STATUS_UNKNOWN_ERROR = 1,
MINIFI_STATUS_NOT_SUPPORTED_PROPERTY = 2,
@@ -98,7 +98,7 @@ typedef enum MinifiStatus {
MINIFI_STATUS_PROCESSOR_YIELD = 6
} MinifiStatus;
-typedef enum MinifiValidator {
+typedef enum MinifiValidator : uint32_t {
MINIFI_VALIDATOR_ALWAYS_VALID = 0,
MINIFI_VALIDATOR_NON_BLANK = 1,
MINIFI_VALIDATOR_TIME_PERIOD = 2,
@@ -125,7 +125,7 @@ typedef struct MinifiPropertyDefinition {
MinifiBool supports_expression_language;
} MinifiPropertyDefinition;
-typedef enum MinifiLogLevel {
+typedef enum MinifiLogLevel : uint32_t {
MINIFI_LOG_LEVEL_TRACE = 0,
MINIFI_LOG_LEVEL_DEBUG = 1,
MINIFI_LOG_LEVEL_INFO = 2,