You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2024/09/26 16:56:17 UTC
(nifi-minifi-cpp) branch main updated (10911ecdc -> 0359e1be1)
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 10911ecdc MINIFICPP-2452 Modbus, Elasticsearch and OPC docker tests not running in CI
new 1eccfd8a2 MINIFICPP-2447 Include the Python processor version in the manifest
new 66e465d9b MINIFICPP-2453 Add yield function support to python API
new a6ef49d7a MINIFICPP-2456 Remove curl output from Elasticsearch responses
new 0359e1be1 MINIFICPP-2460 Add tests for S2S tests using SSL
The 4 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:
docker/test/integration/cluster/ContainerStore.py | 10 +-
.../test/integration/cluster/DockerTestCluster.py | 3 +
.../cluster/DockerTestDirectoryBindings.py | 121 ++++++++----
.../cluster/checkers/ElasticSearchChecker.py | 14 +-
.../cluster/containers/MinifiContainer.py | 5 +-
.../cluster/containers/NifiContainer.py | 61 ++++--
docker/test/integration/convert_cert_to_jks.sh | 45 +++++
.../features/MiNiFi_integration_test_driver.py | 62 +-----
docker/test/integration/features/environment.py | 7 +-
docker/test/integration/features/s2s.feature | 78 +++++++-
docker/test/integration/features/steps/steps.py | 44 ++++-
docker/test/integration/features/utils.py | 2 +-
docker/test/integration/minifi/core/InputPort.py | 3 +
.../integration/minifi/core/RemoteProcessGroup.py | 4 +
.../Minifi_flow_json_serializer.py | 3 +-
.../Minifi_flow_yaml_serializer.py | 2 +-
.../integration/resources/python/CreateNothing.py | 1 +
extensions/python/ExecutePythonProcessor.h | 9 +
extensions/python/PythonCreator.h | 2 +-
extensions/python/PythonProcessor.cpp | 4 +
extensions/python/PythonProcessor.h | 2 +
.../pythonprocessors/nifiapi/processorbase.py | 3 +
.../python/pythonprocessors/nifiapi/properties.py | 3 +
extensions/python/tests/CMakeLists.txt | 1 +
extensions/python/tests/PythonManifestTests.cpp | 212 +++++++++++++++++----
extensions/python/types/PyProcessContext.cpp | 13 ++
extensions/python/types/PyProcessContext.h | 1 +
extensions/python/types/PyProcessor.cpp | 16 ++
extensions/python/types/PyProcessor.h | 1 +
libminifi/src/RemoteProcessorGroupPort.cpp | 1 +
libminifi/src/core/flow/FlowSchema.cpp | 2 +-
libminifi/test/libtest/unit/TestUtils.h | 12 ++
32 files changed, 579 insertions(+), 168 deletions(-)
create mode 100755 docker/test/integration/convert_cert_to_jks.sh
(nifi-minifi-cpp) 04/04: MINIFICPP-2460 Add tests for S2S tests using SSL
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 0359e1be1dd5e325bf5a15570a7da2d3e5d24850
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Sep 26 17:55:59 2024 +0200
MINIFICPP-2460 Add tests for S2S tests using SSL
- Add tests with both yml and json config, and with both controller
property and minifi.properties file configuration
- Fix RPG port properties with JSON flow config
- Add verbose curl debug logs for S2S communication
- Remove redundant feature_id in test modules
- Create test certificate files on test feature level
Closes #1869
Signed-off-by: Marton Szasz <sz...@apache.org>
---
docker/test/integration/cluster/ContainerStore.py | 10 +-
.../test/integration/cluster/DockerTestCluster.py | 3 +
.../cluster/DockerTestDirectoryBindings.py | 121 +++++++++++++++------
.../cluster/containers/MinifiContainer.py | 5 +-
.../cluster/containers/NifiContainer.py | 61 +++++++----
docker/test/integration/convert_cert_to_jks.sh | 45 ++++++++
.../features/MiNiFi_integration_test_driver.py | 62 ++---------
docker/test/integration/features/environment.py | 7 +-
docker/test/integration/features/s2s.feature | 78 ++++++++++++-
docker/test/integration/features/steps/steps.py | 44 ++++++--
docker/test/integration/minifi/core/InputPort.py | 3 +
.../integration/minifi/core/RemoteProcessGroup.py | 4 +
.../Minifi_flow_json_serializer.py | 3 +-
.../Minifi_flow_yaml_serializer.py | 2 +-
libminifi/src/RemoteProcessorGroupPort.cpp | 1 +
libminifi/src/core/flow/FlowSchema.cpp | 2 +-
16 files changed, 325 insertions(+), 126 deletions(-)
diff --git a/docker/test/integration/cluster/ContainerStore.py b/docker/test/integration/cluster/ContainerStore.py
index f48646158..f1e13a894 100644
--- a/docker/test/integration/cluster/ContainerStore.py
+++ b/docker/test/integration/cluster/ContainerStore.py
@@ -17,6 +17,7 @@ import shortuuid
from .containers.MinifiContainer import MinifiOptions
from .containers.MinifiContainer import MinifiContainer
from .containers.NifiContainer import NifiContainer
+from .containers.NifiContainer import NiFiOptions
from .containers.ZookeeperContainer import ZookeeperContainer
from .containers.KafkaBrokerContainer import KafkaBrokerContainer
from .containers.S3ServerContainer import S3ServerContainer
@@ -52,6 +53,7 @@ class ContainerStore:
self.network = network
self.image_store = image_store
self.kubernetes_proxy = kubernetes_proxy
+ self.nifi_options = NiFiOptions()
def get_container_name_with_postfix(self, container_name: str):
if not container_name.endswith(self.feature_id):
@@ -83,13 +85,14 @@ class ContainerStore:
logging.info('Container name was not provided; using generated name \'%s\'', container_name)
feature_context = FeatureContext(feature_id=context.feature_id,
- root_ca_cert=context.test.root_ca_cert,
- root_ca_key=context.test.root_ca_key)
+ root_ca_cert=context.root_ca_cert,
+ root_ca_key=context.root_ca_key)
if engine == 'nifi':
return self.containers.setdefault(container_name,
NifiContainer(feature_context=feature_context,
config_dir=self.data_directories["nifi_config_dir"],
+ options=self.nifi_options,
name=container_name,
vols=self.vols,
network=self.network,
@@ -405,3 +408,6 @@ class ContainerStore:
def enable_multi_tenancy_in_grafana_loki(self):
self.grafana_loki_options.enable_multi_tenancy = True
+
+ def enable_ssl_in_nifi(self):
+ self.nifi_options.use_ssl = True
diff --git a/docker/test/integration/cluster/DockerTestCluster.py b/docker/test/integration/cluster/DockerTestCluster.py
index 430ec52f9..153ffce26 100644
--- a/docker/test/integration/cluster/DockerTestCluster.py
+++ b/docker/test/integration/cluster/DockerTestCluster.py
@@ -418,3 +418,6 @@ class DockerTestCluster:
def set_value_on_plc_with_modbus(self, container_name, modbus_cmd):
return self.modbus_checker.set_value_on_plc_with_modbus(container_name, modbus_cmd)
+
+ def enable_ssl_in_nifi(self):
+ self.container_store.enable_ssl_in_nifi()
diff --git a/docker/test/integration/cluster/DockerTestDirectoryBindings.py b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
index fec53dfb1..87cf2b4ec 100644
--- a/docker/test/integration/cluster/DockerTestDirectoryBindings.py
+++ b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
@@ -12,12 +12,13 @@
# 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 os
import shutil
import hashlib
+import subprocess
+import OpenSSL.crypto
+from ssl_utils.SSL_cert_utils import make_self_signed_cert, make_cert_without_extended_usage, make_server_cert
class DockerTestDirectoryBindings:
@@ -59,39 +60,39 @@ class DockerTestDirectoryBindings:
shutil.copytree(test_dir + "/resources/minifi", self.data_directories[self.feature_id]["minifi_config_dir"], dirs_exist_ok=True)
shutil.copytree(test_dir + "/resources/minifi-controller", self.data_directories[self.feature_id]["resources_dir"] + "/minifi-controller")
- def get_data_directories(self, feature_id):
- return self.data_directories[feature_id]
+ def get_data_directories(self):
+ return self.data_directories[self.feature_id]
- def docker_path_to_local_path(self, feature_id, docker_path):
+ def docker_path_to_local_path(self, docker_path):
# Docker paths are currently hard-coded
if docker_path == "/tmp/input":
- return self.data_directories[feature_id]["input_dir"]
+ return self.data_directories[self.feature_id]["input_dir"]
if docker_path == "/tmp/output":
- return self.data_directories[feature_id]["output_dir"]
+ return self.data_directories[self.feature_id]["output_dir"]
if docker_path == "/tmp/resources":
- return self.data_directories[feature_id]["resources_dir"]
+ return self.data_directories[self.feature_id]["resources_dir"]
# Might be worth reworking these
if docker_path == "/tmp/output/success":
- self.create_directory(self.data_directories[feature_id]["output_dir"] + "/success")
- return self.data_directories[feature_id]["output_dir"] + "/success"
+ self.create_directory(self.data_directories[self.feature_id]["output_dir"] + "/success")
+ return self.data_directories[self.feature_id]["output_dir"] + "/success"
if docker_path == "/tmp/output/failure":
- self.create_directory(self.data_directories[feature_id]["output_dir"] + "/failure")
- return self.data_directories[feature_id]["output_dir"] + "/failure"
+ self.create_directory(self.data_directories[self.feature_id]["output_dir"] + "/failure")
+ return self.data_directories[self.feature_id]["output_dir"] + "/failure"
raise Exception("Docker directory \"%s\" has no preset bindings." % docker_path)
- def get_directory_bindings(self, feature_id):
+ def get_directory_bindings(self):
"""
Performs a standard container flow deployment with the addition
of volumes supporting test input/output directories.
"""
vols = {}
- vols[self.data_directories[feature_id]["input_dir"]] = {"bind": "/tmp/input", "mode": "rw"}
- vols[self.data_directories[feature_id]["output_dir"]] = {"bind": "/tmp/output", "mode": "rw"}
- vols[self.data_directories[feature_id]["resources_dir"]] = {"bind": "/tmp/resources", "mode": "rw"}
- vols[self.data_directories[feature_id]["system_certs_dir"]] = {"bind": "/usr/local/share/certs", "mode": "rw"}
- vols[self.data_directories[feature_id]["minifi_config_dir"]] = {"bind": "/tmp/minifi_config", "mode": "rw"}
- vols[self.data_directories[feature_id]["nifi_config_dir"]] = {"bind": "/tmp/nifi_config", "mode": "rw"}
- vols[self.data_directories[feature_id]["kubernetes_config_dir"]] = {"bind": "/tmp/kubernetes_config", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["input_dir"]] = {"bind": "/tmp/input", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["output_dir"]] = {"bind": "/tmp/output", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["resources_dir"]] = {"bind": "/tmp/resources", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["system_certs_dir"]] = {"bind": "/usr/local/share/certs", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["minifi_config_dir"]] = {"bind": "/tmp/minifi_config", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["nifi_config_dir"]] = {"bind": "/tmp/nifi_config", "mode": "rw"}
+ vols[self.data_directories[self.feature_id]["kubernetes_config_dir"]] = {"bind": "/tmp/kubernetes_config", "mode": "rw"}
return vols
@staticmethod
@@ -120,21 +121,24 @@ class DockerTestDirectoryBindings:
test_input_file.write(contents)
os.chmod(file_abs_path, 0o0777)
- def put_test_resource(self, feature_id, file_name, contents):
+ def put_test_resource(self, file_name, contents):
"""
Creates a resource file in the test resource dir and writes
the given content to it.
"""
- file_abs_path = os.path.join(self.data_directories[feature_id]["resources_dir"], file_name)
+ file_abs_path = os.path.join(self.data_directories[self.feature_id]["resources_dir"], file_name)
self.put_file_contents(file_abs_path, contents)
- def put_test_input(self, feature_id, file_name, contents):
- file_abs_path = os.path.join(self.data_directories[feature_id]["input_dir"], file_name)
+ def get_test_resource_path(self, file_name):
+ return os.path.join(self.data_directories[self.feature_id]["resources_dir"], file_name)
+
+ def put_test_input(self, file_name, contents):
+ file_abs_path = os.path.join(self.data_directories[self.feature_id]["input_dir"], file_name)
self.put_file_contents(file_abs_path, contents)
- def put_file_to_docker_path(self, feature_id, path, file_name, contents):
- file_abs_path = os.path.join(self.docker_path_to_local_path(feature_id, path), file_name)
+ def put_file_to_docker_path(self, path, file_name, contents):
+ file_abs_path = os.path.join(self.docker_path_to_local_path(path), file_name)
self.put_file_contents(file_abs_path, contents)
@staticmethod
@@ -146,14 +150,67 @@ class DockerTestDirectoryBindings:
return md5_hash.hexdigest()
- def put_random_file_to_docker_path(self, test_id: str, path: str, file_name: str, file_size: int):
- file_abs_path = os.path.join(self.docker_path_to_local_path(test_id, path), file_name)
+ def put_random_file_to_docker_path(self, path: str, file_name: str, file_size: int):
+ file_abs_path = os.path.join(self.docker_path_to_local_path(path), file_name)
with open(file_abs_path, 'wb') as test_input_file:
test_input_file.write(os.urandom(file_size))
os.chmod(file_abs_path, 0o0777)
return self.generate_md5_hash(file_abs_path)
- def rm_out_child(self, feature_id, dir):
- child = os.path.join(self.data_directories[feature_id]["output_dir"], dir)
- logging.info('Removing %s from output folder', child)
- shutil.rmtree(child)
+ def create_cert_files(self):
+ self.root_ca_cert, self.root_ca_key = make_self_signed_cert("root CA")
+
+ minifi_client_cert, minifi_client_key = make_cert_without_extended_usage(common_name=f"minifi-cpp-flow-{self.feature_id}",
+ ca_cert=self.root_ca_cert,
+ ca_key=self.root_ca_key)
+ minifi_server_cert, minifi_server_key = make_server_cert(common_name=f"server-{self.feature_id}",
+ ca_cert=self.root_ca_cert,
+ ca_key=self.root_ca_key)
+ self_signed_server_cert, self_signed_server_key = make_self_signed_cert(f"server-{self.feature_id}")
+
+ self.put_test_resource('root_ca.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=self.root_ca_cert))
+ self.put_test_resource("system_certs_dir/ca-root-nss.crt",
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=self.root_ca_cert))
+ self.put_test_resource('minifi_client.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=minifi_client_cert))
+ self.put_test_resource('minifi_client.key',
+ OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=minifi_client_key))
+ self.put_test_resource('minifi_server.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=minifi_server_cert)
+ + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=minifi_server_key))
+ self.put_test_resource('self_signed_server.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=self_signed_server_cert)
+ + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=self_signed_server_key))
+ self.put_test_resource('minifi_merged_cert.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=minifi_client_cert)
+ + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=minifi_client_key))
+ nifi_client_cert, nifi_client_key = make_server_cert(common_name=f"nifi-{self.feature_id}",
+ ca_cert=self.root_ca_cert,
+ ca_key=self.root_ca_key)
+ self.put_test_resource('nifi_client.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=nifi_client_cert))
+ self.put_test_resource('nifi_client.key',
+ OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=nifi_client_key))
+ base = os.path.dirname(self.get_test_resource_path('nifi_client.key'))
+ test_dir = os.environ['TEST_DIRECTORY'] # Based on DockerVerify.sh
+ cmd = [
+ os.path.join(test_dir, "convert_cert_to_jks.sh"),
+ base,
+ os.path.join(base, "nifi_client.key"),
+ os.path.join(base, "nifi_client.crt"),
+ os.path.join(base, "root_ca.crt"),
+ ]
+ subprocess.run(cmd, check=True)
diff --git a/docker/test/integration/cluster/containers/MinifiContainer.py b/docker/test/integration/cluster/containers/MinifiContainer.py
index fee054f06..770f42df0 100644
--- a/docker/test/integration/cluster/containers/MinifiContainer.py
+++ b/docker/test/integration/cluster/containers/MinifiContainer.py
@@ -116,9 +116,8 @@ class MinifiContainer(FlowContainer):
if self.options.set_ssl_context_properties:
f.write("nifi.remote.input.secure=true\n")
- f.write("nifi.security.client.certificate=/tmp/resources/minifi-cpp-flow.crt\n")
- f.write("nifi.security.client.private.key=/tmp/resources/minifi-cpp-flow.key\n")
- f.write("nifi.security.client.pass.phrase=abcdefgh\n")
+ f.write("nifi.security.client.certificate=/tmp/resources/minifi_client.crt\n")
+ f.write("nifi.security.client.private.key=/tmp/resources/minifi_client.key\n")
f.write("nifi.security.client.ca.certificate=/tmp/resources/root_ca.crt\n")
if not self.options.enable_provenance:
diff --git a/docker/test/integration/cluster/containers/NifiContainer.py b/docker/test/integration/cluster/containers/NifiContainer.py
index 20cbd7282..c3ae965cb 100644
--- a/docker/test/integration/cluster/containers/NifiContainer.py
+++ b/docker/test/integration/cluster/containers/NifiContainer.py
@@ -22,30 +22,53 @@ import gzip
import os
+class NiFiOptions:
+ def __init__(self):
+ self.use_ssl = False
+
+
class NifiContainer(FlowContainer):
NIFI_VERSION = '2.0.0-M2'
NIFI_ROOT = '/opt/nifi/nifi-' + NIFI_VERSION
- def __init__(self, feature_context, config_dir, name, vols, network, image_store, command=None):
+ def __init__(self, feature_context, config_dir, options, name, vols, network, image_store, command=None):
if not command:
- entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.sensitive.props.key\)=.*/\1=secret_key_12345/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.remote.input.secure\)=.*/\1=false/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.web.http.port\)=.*/\1=8080/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.web.https.port\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.web.https.host\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.web.http.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.keystore\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.keystoreType\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.keystorePasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.keyPasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.truststore\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.truststoreType\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.security.truststorePasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
- r"sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\1=10000/' {nifi_root}/conf/nifi.properties && "
- r"cp /tmp/nifi_config/flow.json.gz {nifi_root}/conf && {nifi_root}/bin/nifi.sh run & "
- r"nifi_pid=$! &&"
- r"tail -F --pid=${{nifi_pid}} {nifi_root}/logs/nifi-app.log").format(name=name, nifi_root=NifiContainer.NIFI_ROOT)
+ if options.use_ssl:
+ entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.remote.input.secure\)=.*/\1=true/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.sensitive.props.key\)=.*/\1=secret_key_12345/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.https.port\)=.*/\1=8443/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.https.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystore\)=.*/\1=\/tmp\/resources\/keystore.jks/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystoreType\)=.*/\1=jks/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystorePasswd\)=.*/\1=passw0rd1!/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keyPasswd\)=.*/#\1=passw0rd1!/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststore\)=.*/\1=\/tmp\/resources\/truststore.jks/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststoreType\)=.*/\1=jks/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststorePasswd\)=.*/\1=passw0rd1!/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\1=10443/' {nifi_root}/conf/nifi.properties && "
+ r"cp /tmp/nifi_config/flow.json.gz {nifi_root}/conf && {nifi_root}/bin/nifi.sh run & "
+ r"nifi_pid=$! &&"
+ r"tail -F --pid=${{nifi_pid}} {nifi_root}/logs/nifi-app.log").format(name=name, nifi_root=NifiContainer.NIFI_ROOT)
+ else:
+ entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.sensitive.props.key\)=.*/\1=secret_key_12345/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.remote.input.secure\)=.*/\1=false/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.http.port\)=.*/\1=8080/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.https.port\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.https.host\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.web.http.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystore\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystoreType\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keystorePasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.keyPasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststore\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststoreType\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.security.truststorePasswd\)=.*/#\1=/' {nifi_root}/conf/nifi.properties && "
+ r"sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\1=10000/' {nifi_root}/conf/nifi.properties && "
+ r"cp /tmp/nifi_config/flow.json.gz {nifi_root}/conf && {nifi_root}/bin/nifi.sh run & "
+ r"nifi_pid=$! &&"
+ r"tail -F --pid=${{nifi_pid}} {nifi_root}/logs/nifi-app.log").format(name=name, nifi_root=NifiContainer.NIFI_ROOT)
command = ["/bin/sh", "-c", entry_command]
super().__init__(feature_context, config_dir, name, 'nifi', vols, network, image_store, command)
diff --git a/docker/test/integration/convert_cert_to_jks.sh b/docker/test/integration/convert_cert_to_jks.sh
new file mode 100755
index 000000000..33c37b677
--- /dev/null
+++ b/docker/test/integration/convert_cert_to_jks.sh
@@ -0,0 +1,45 @@
+#!/bin/bash
+set -euo pipefail
+
+# Usage: ./create_jks.sh <base directory> <ssl_key_path> <ssl_cert_path> <ca_cert_path>
+
+DIR=$1
+SSL_KEY_PATH=$2
+SSL_CERT_PATH=$3
+CA_CERT_PATH=$4
+
+KEYSTORE="$DIR/keystore.jks"
+TRUSTSTORE="$DIR/truststore.jks"
+PKCS12_FILE="$DIR/keystore.p12"
+PASSWORD="passw0rd1!"
+
+cat "${CA_CERT_PATH}" >> "${SSL_CERT_PATH}"
+
+if [ ! -d "$DIR" ]; then
+ mkdir -p "$DIR"
+fi
+
+openssl pkcs12 -export \
+ -inkey "$SSL_KEY_PATH" \
+ -in "$SSL_CERT_PATH" \
+ -name "nifi-key" \
+ -out "$PKCS12_FILE" \
+ -password pass:$PASSWORD
+
+keytool -importkeystore \
+ -destkeystore "$KEYSTORE" \
+ -deststoretype jks \
+ -destalias nifi-key \
+ -srckeystore "$PKCS12_FILE" \
+ -srcstoretype pkcs12 \
+ -srcalias "nifi-key" \
+ -storepass "$PASSWORD" \
+ -srcstorepass "$PASSWORD" \
+ -noprompt
+
+keytool -importcert \
+ -alias "nifi-cert" \
+ -file "$CA_CERT_PATH" \
+ -keystore "$TRUSTSTORE" \
+ -storepass "$PASSWORD" \
+ -noprompt
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index 6ad052294..2eef2f22d 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -12,22 +12,14 @@
# 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 time
import uuid
-from typing import List
-
-import OpenSSL.crypto
+from typing import List
from pydoc import locate
-
-from ssl_utils.SSL_cert_utils import make_self_signed_cert, make_cert_without_extended_usage, make_server_cert
from minifi.core.InputPort import InputPort
-
from cluster.DockerTestCluster import DockerTestCluster
-
from minifi.validators.OutputValidator import OutputValidator
from minifi.validators.EmptyFilesOutPutValidator import EmptyFilesOutPutValidator
from minifi.validators.NoFileOutPutValidator import NoFileOutPutValidator
@@ -54,44 +46,7 @@ class MiNiFi_integration_test:
self.test_file_hash = None
self.docker_directory_bindings = context.directory_bindings
- self.cluster.set_directory_bindings(self.docker_directory_bindings.get_directory_bindings(self.feature_id), self.docker_directory_bindings.get_data_directories(self.feature_id))
- self.root_ca_cert, self.root_ca_key = make_self_signed_cert("root CA")
-
- minifi_client_cert, minifi_client_key = make_cert_without_extended_usage(common_name=f"minifi-cpp-flow-{self.feature_id}",
- ca_cert=self.root_ca_cert,
- ca_key=self.root_ca_key)
- minifi_server_cert, minifi_server_key = make_server_cert(common_name=f"server-{self.feature_id}",
- ca_cert=self.root_ca_cert,
- ca_key=self.root_ca_key)
- self_signed_server_cert, self_signed_server_key = make_self_signed_cert(f"server-{self.feature_id}")
-
- self.put_test_resource('root_ca.crt',
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=self.root_ca_cert))
- self.put_test_resource("system_certs_dir/ca-root-nss.crt",
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=self.root_ca_cert))
- self.put_test_resource('minifi_client.crt',
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=minifi_client_cert))
- self.put_test_resource('minifi_client.key',
- OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
- pkey=minifi_client_key))
- self.put_test_resource('minifi_server.crt',
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=minifi_server_cert)
- + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
- pkey=minifi_server_key))
- self.put_test_resource('self_signed_server.crt',
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=self_signed_server_cert)
- + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
- pkey=self_signed_server_key))
- self.put_test_resource('minifi_merged_cert.crt',
- OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
- cert=minifi_client_cert)
- + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
- pkey=minifi_client_key))
+ self.cluster.set_directory_bindings(self.docker_directory_bindings.get_directory_bindings(), self.docker_directory_bindings.get_data_directories())
def get_container_name_with_postfix(self, container_name: str):
return self.cluster.container_store.get_container_name_with_postfix(container_name)
@@ -196,18 +151,18 @@ class MiNiFi_integration_test:
if file_name is None:
file_name = str(uuid.uuid4())
test_data = decode_escaped_str(test_data)
- self.docker_directory_bindings.put_file_to_docker_path(self.feature_id, path, file_name, test_data.encode('utf-8'))
+ self.docker_directory_bindings.put_file_to_docker_path(path, file_name, test_data.encode('utf-8'))
def add_random_test_data(self, path: str, size: int, file_name: str = None):
if file_name is None:
file_name = str(uuid.uuid4())
- self.test_file_hash = self.docker_directory_bindings.put_random_file_to_docker_path(self.feature_id, path, file_name, size)
+ self.test_file_hash = self.docker_directory_bindings.put_random_file_to_docker_path(path, file_name, size)
def put_test_resource(self, file_name, contents):
- self.docker_directory_bindings.put_test_resource(self.feature_id, file_name, contents)
+ self.docker_directory_bindings.put_test_resource(file_name, contents)
- def rm_out_child(self):
- self.docker_directory_bindings.rm_out_child(self.feature_id)
+ def get_test_resource_path(self, file_name):
+ return self.docker_directory_bindings.get_test_resource_path(file_name)
def add_file_system_observer(self, file_system_observer):
self.file_system_observer = file_system_observer
@@ -482,3 +437,6 @@ class MiNiFi_integration_test:
def set_value_on_plc_with_modbus(self, container_name, modbus_cmd):
assert self.cluster.set_value_on_plc_with_modbus(container_name, modbus_cmd)
+
+ def enable_ssl_in_nifi(self):
+ self.cluster.enable_ssl_in_nifi()
diff --git a/docker/test/integration/features/environment.py b/docker/test/integration/features/environment.py
index 5483b1138..d98d2f3f5 100644
--- a/docker/test/integration/features/environment.py
+++ b/docker/test/integration/features/environment.py
@@ -85,11 +85,14 @@ def before_feature(context, feature):
context.feature_id = feature_id
context.directory_bindings = DockerTestDirectoryBindings(feature_id)
context.directory_bindings.create_new_data_directories()
+ context.directory_bindings.create_cert_files()
+ context.root_ca_cert = context.directory_bindings.root_ca_cert
+ context.root_ca_key = context.directory_bindings.root_ca_key
if "requires.kubernetes.cluster" in feature.tags:
context.kubernetes_proxy = KubernetesProxy(
- context.directory_bindings.get_data_directories(context.feature_id)["kubernetes_temp_dir"],
+ context.directory_bindings.get_data_directories()["kubernetes_temp_dir"],
os.path.join(os.environ['TEST_DIRECTORY'], 'resources', 'kubernetes', 'pods-etc'))
- context.kubernetes_proxy.create_config(context.directory_bindings.get_directory_bindings(context.feature_id))
+ context.kubernetes_proxy.create_config(context.directory_bindings.get_directory_bindings())
context.kubernetes_proxy.start_cluster()
diff --git a/docker/test/integration/features/s2s.feature b/docker/test/integration/features/s2s.feature
index c236f2ac7..80a8eb648 100644
--- a/docker/test/integration/features/s2s.feature
+++ b/docker/test/integration/features/s2s.feature
@@ -28,7 +28,7 @@ Feature: Sending data from MiNiFi-C++ to NiFi using S2S protocol
And a RemoteProcessGroup node opened on "http://nifi-${feature_id}:8080/nifi"
And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
- And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi" on port 8080
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
And the "success" relationship of the from-minifi is connected to the PutFile
@@ -42,7 +42,7 @@ Feature: Sending data from MiNiFi-C++ to NiFi using S2S protocol
And a RemoteProcessGroup node opened on "http://nifi-${feature_id}:8080/nifi"
And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
- And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi" on port 8080
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
And the "success" relationship of the from-minifi is connected to the PutFile
@@ -56,7 +56,7 @@ Feature: Sending data from MiNiFi-C++ to NiFi using S2S protocol
And a RemoteProcessGroup node opened on "http://nifi-${feature_id}:8080/nifi"
And the "success" relationship of the GenerateFlowFile processor is connected to the input port on the RemoteProcessGroup
- And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi" on port 8080
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
And the "success" relationship of the from-minifi is connected to the PutFile
@@ -71,9 +71,79 @@ Feature: Sending data from MiNiFi-C++ to NiFi using S2S protocol
And the "success" relationship of the GenerateFlowFile processor is connected to the input port on the RemoteProcessGroup
And the connection going to the RemoteProcessGroup has "drop empty" set
- And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi" on port 8080
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
And the "success" relationship of the from-minifi is connected to the PutFile
When both instances start up
Then no files are placed in the monitored directory in 50 seconds of running time
+
+ Scenario: A MiNiFi instance produces and transfers data to a NiFi instance via s2s using SSL
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And the "Keep Source File" property of the GetFile processor is set to "true"
+ And a file with the content "test" is present in "/tmp/input"
+ And a RemoteProcessGroup node opened on "https://nifi-${feature_id}:8443/nifi"
+ And a SSL context service is set up for the following remote process group: "RemoteProcessGroup"
+ And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
+
+ And SSL is enabled in NiFi flow
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
+ And the "success" relationship of the from-minifi is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "test" is placed in the monitored directory in less than 90 seconds
+ And the Minifi logs do not contain the following message: "ProcessSession rollback" after 1 seconds
+
+ Scenario: A MiNiFi instance produces and transfers data to a NiFi instance via s2s using SSL with YAML config
+ Given a MiNiFi CPP server with yaml config
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And the "Keep Source File" property of the GetFile processor is set to "true"
+ And a file with the content "test" is present in "/tmp/input"
+ And a RemoteProcessGroup node opened on "https://nifi-${feature_id}:8443/nifi"
+ And a SSL context service is set up for the following remote process group: "RemoteProcessGroup"
+ And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
+
+ And SSL is enabled in NiFi flow
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
+ And the "success" relationship of the from-minifi is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "test" is placed in the monitored directory in less than 90 seconds
+ And the Minifi logs do not contain the following message: "ProcessSession rollback" after 1 seconds
+
+ Scenario: A MiNiFi instance produces and transfers data to a NiFi instance via s2s using SSL config defined in minifi.properties
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And the "Keep Source File" property of the GetFile processor is set to "true"
+ And a file with the content "test" is present in "/tmp/input"
+ And a RemoteProcessGroup node opened on "https://nifi-${feature_id}:8443/nifi"
+ And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
+ And SSL properties are set in MiNiFi
+
+ And SSL is enabled in NiFi flow
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
+ And the "success" relationship of the from-minifi is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "test" is placed in the monitored directory in less than 90 seconds
+ And the Minifi logs do not contain the following message: "ProcessSession rollback" after 1 seconds
+
+ Scenario: A MiNiFi instance produces and transfers data to a NiFi instance via s2s using YAML config and SSL config defined in minifi.properties
+ Given a MiNiFi CPP server with yaml config
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And the "Keep Source File" property of the GetFile processor is set to "true"
+ And a file with the content "test" is present in "/tmp/input"
+ And a RemoteProcessGroup node opened on "https://nifi-${feature_id}:8443/nifi"
+ And the "success" relationship of the GetFile processor is connected to the input port on the RemoteProcessGroup
+ And SSL properties are set in MiNiFi
+
+ And SSL is enabled in NiFi flow
+ And a NiFi flow receiving data from a RemoteProcessGroup "from-minifi"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "nifi" flow
+ And the "success" relationship of the from-minifi is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "test" is placed in the monitored directory in less than 90 seconds
+ And the Minifi logs do not contain the following message: "ProcessSession rollback" after 1 seconds
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 75cf19a34..92dd64548 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -47,7 +47,7 @@ import os
# Background
@given("the content of \"{directory}\" is monitored")
def step_impl(context, directory):
- context.test.add_file_system_observer(FileSystemObserver(context.directory_bindings.docker_path_to_local_path(context.feature_id, directory)))
+ context.test.add_file_system_observer(FileSystemObserver(context.directory_bindings.docker_path_to_local_path(directory)))
def __create_processor(context, processor_type, processor_name, property_name, property_value, container_name, engine='minifi-cpp'):
@@ -318,7 +318,7 @@ def step_impl(context, processor_name):
# NiFi setups
-@given("a NiFi flow receiving data from a RemoteProcessGroup \"{source_name}\" on port 8080")
+@given("a NiFi flow receiving data from a RemoteProcessGroup \"{source_name}\"")
def step_impl(context, source_name):
remote_process_group = context.test.get_remote_process_group_by_name("RemoteProcessGroup")
source = context.test.generate_input_port_for_remote_process_group(remote_process_group, source_name)
@@ -335,6 +335,11 @@ def step_impl(context, flow_name):
context.test.acquire_container(context=context, name=flow_name, engine='nifi')
+@given("SSL is enabled in NiFi flow")
+def step_impl(context):
+ context.test.enable_ssl_in_nifi()
+
+
@given("a transient MiNiFi flow with the name \"{flow_name}\" is set up")
def step_impl(context, flow_name):
context.test.acquire_container(context=context, name=flow_name, command=["/bin/sh", "-c", "timeout 10s ./bin/minifi.sh run && sleep 100"])
@@ -496,9 +501,20 @@ def setUpSslContextServiceForProcessor(context, processor_name: str):
minifi_key_file = '/tmp/resources/minifi_client.key'
root_ca_crt_file = '/tmp/resources/root_ca.crt'
ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
- post_elasticsearch_json = context.test.get_node_by_name(processor_name)
- post_elasticsearch_json.controller_services.append(ssl_context_service)
- post_elasticsearch_json.set_property("SSL Context Service", ssl_context_service.name)
+ processor = context.test.get_node_by_name(processor_name)
+ processor.controller_services.append(ssl_context_service)
+ processor.set_property("SSL Context Service", ssl_context_service.name)
+
+
+def setUpSslContextServiceForRPG(context, rpg_name: str):
+ minifi_crt_file = '/tmp/resources/minifi_client.crt'
+ minifi_key_file = '/tmp/resources/minifi_client.key'
+ root_ca_crt_file = '/tmp/resources/root_ca.crt'
+ ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(ssl_context_service)
+ rpg = context.test.get_remote_process_group_by_name(rpg_name)
+ rpg.add_property("SSL Context Service", ssl_context_service.name)
@given(u'a SSL context service is set up for PostElasticsearch and Elasticsearch')
@@ -551,14 +567,14 @@ def step_impl(context):
root_ca_crt_file = '/tmp/resources/root_ca.crt'
ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
- splunk_cert, splunk_key = make_server_cert(context.test.get_container_name_with_postfix("splunk"), context.test.root_ca_cert, context.test.root_ca_key)
+ splunk_cert, splunk_key = make_server_cert(context.test.get_container_name_with_postfix("splunk"), context.root_ca_cert, context.root_ca_key)
put_splunk_http = context.test.get_node_by_name("PutSplunkHTTP")
put_splunk_http.controller_services.append(ssl_context_service)
put_splunk_http.set_property("SSL Context Service", ssl_context_service.name)
query_splunk_indexing_status = context.test.get_node_by_name("QuerySplunkIndexingStatus")
query_splunk_indexing_status.controller_services.append(ssl_context_service)
query_splunk_indexing_status.set_property("SSL Context Service", ssl_context_service.name)
- context.test.enable_splunk_hec_ssl('splunk', OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, splunk_cert), OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, splunk_key), OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, context.test.root_ca_cert))
+ context.test.enable_splunk_hec_ssl('splunk', OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, splunk_cert), OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, splunk_key), OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, context.root_ca_cert))
@given(u'the {processor_one} processor is set up with a GCPCredentialsControllerService to communicate with the Google Cloud storage server')
@@ -690,11 +706,11 @@ def step_impl(context, content, topic_name):
@when("a message with content \"{content}\" is published to the \"{topic_name}\" topic using an ssl connection")
def step_impl(context, content, topic_name):
ca_cert_file = tempfile.NamedTemporaryFile(delete=False)
- ca_cert_file.write(OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM, cert=context.test.root_ca_cert))
+ ca_cert_file.write(OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM, cert=context.root_ca_cert))
ca_cert_file.close()
os.chmod(ca_cert_file.name, 0o644)
- client_cert, client_key = make_client_cert(socket.gethostname(), context.test.root_ca_cert, context.test.root_ca_key)
+ client_cert, client_key = make_client_cert(socket.gethostname(), context.root_ca_cert, context.root_ca_key)
client_cert_file = tempfile.NamedTemporaryFile(delete=False)
client_cert_file.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, client_cert))
client_cert_file.close()
@@ -1065,6 +1081,11 @@ def step_impl(context):
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")
@@ -1280,6 +1301,11 @@ def step_impl(context, processor_name: str):
setUpSslContextServiceForProcessor(context, processor_name)
+@given(u'a SSL context service is set up for the following remote process group: \"{remote_process_group}\"')
+def step_impl(context, remote_process_group: str):
+ setUpSslContextServiceForRPG(context, remote_process_group)
+
+
# Nginx reverse proxy
@given(u'a reverse proxy is set up to forward requests to the Grafana Loki server')
def step_impl(context):
diff --git a/docker/test/integration/minifi/core/InputPort.py b/docker/test/integration/minifi/core/InputPort.py
index 817f5791f..cb9d60dc6 100644
--- a/docker/test/integration/minifi/core/InputPort.py
+++ b/docker/test/integration/minifi/core/InputPort.py
@@ -22,3 +22,6 @@ class InputPort(Connectable):
super(InputPort, self).__init__(name=name)
self.remote_process_group = remote_process_group
+ self.properties = {}
+ if self.remote_process_group:
+ self.properties = self.remote_process_group.properties
diff --git a/docker/test/integration/minifi/core/RemoteProcessGroup.py b/docker/test/integration/minifi/core/RemoteProcessGroup.py
index c57b2d1a7..e3f54fa43 100644
--- a/docker/test/integration/minifi/core/RemoteProcessGroup.py
+++ b/docker/test/integration/minifi/core/RemoteProcessGroup.py
@@ -27,9 +27,13 @@ class RemoteProcessGroup(object):
self.name = name
self.url = url
+ self.properties = {}
def get_name(self):
return self.name
def get_uuid(self):
return self.uuid
+
+ def add_property(self, name, value):
+ self.properties[name] = value
diff --git a/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py b/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
index e1769f632..c1043d506 100644
--- a/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
+++ b/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
@@ -94,7 +94,8 @@ class Minifi_flow_json_serializer:
res_group['inputPorts'].append({
'identifier': str(connectable.instance_id),
- 'name': connectable.name
+ 'name': connectable.name,
+ 'properties': connectable.properties
})
if isinstance(connectable, Processor):
diff --git a/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py b/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
index ff8b37de8..dc912b122 100644
--- a/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
+++ b/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
@@ -96,7 +96,7 @@ class Minifi_flow_yaml_serializer:
'id': str(connectable.instance_id),
'name': connectable.name,
'max concurrent tasks': 1,
- 'Properties': {}
+ 'Properties': connectable.properties
})
if isinstance(connectable, Processor):
diff --git a/libminifi/src/RemoteProcessorGroupPort.cpp b/libminifi/src/RemoteProcessorGroupPort.cpp
index aee1f7c39..73e6d4123 100644
--- a/libminifi/src/RemoteProcessorGroupPort.cpp
+++ b/libminifi/src/RemoteProcessorGroupPort.cpp
@@ -292,6 +292,7 @@ std::pair<std::string, int> RemoteProcessorGroupPort::refreshRemoteSite2SiteInfo
if (!token.empty())
client->setRequestHeader("Authorization", token);
+ client->setVerbose(false);
if (client->submit() && client->getResponseCode() == 200) {
const std::vector<char> &response_body = client->getResponseBody();
diff --git a/libminifi/src/core/flow/FlowSchema.cpp b/libminifi/src/core/flow/FlowSchema.cpp
index b09b4f539..4ab3aba0b 100644
--- a/libminifi/src/core/flow/FlowSchema.cpp
+++ b/libminifi/src/core/flow/FlowSchema.cpp
@@ -141,7 +141,7 @@ FlowSchema FlowSchema::getNiFiFlowJson() {
.rpg_proxy_port = {"proxyPort"},
.rpg_input_ports = {"inputPorts"},
.rpg_output_ports = {"outputPorts"},
- .rpg_port_properties = {},
+ .rpg_port_properties = {"properties"},
.rpg_port_target_id = {"targetId"},
.parameter_contexts = {"parameterContexts"},
(nifi-minifi-cpp) 02/04: MINIFICPP-2453 Add yield function support to python API
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 66e465d9b9053cd2ad8c208611c84c329b7ac7d5
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Sep 26 17:55:20 2024 +0200
MINIFICPP-2453 Add yield function support to python API
Closes #1865
Signed-off-by: Marton Szasz <sz...@apache.org>
---
docker/test/integration/resources/python/CreateNothing.py | 1 +
extensions/python/pythonprocessors/nifiapi/properties.py | 3 +++
extensions/python/types/PyProcessContext.cpp | 13 +++++++++++++
extensions/python/types/PyProcessContext.h | 1 +
4 files changed, 18 insertions(+)
diff --git a/docker/test/integration/resources/python/CreateNothing.py b/docker/test/integration/resources/python/CreateNothing.py
index b39dbcee4..acef36dd8 100644
--- a/docker/test/integration/resources/python/CreateNothing.py
+++ b/docker/test/integration/resources/python/CreateNothing.py
@@ -29,4 +29,5 @@ class CreateNothing(FlowFileSource):
pass
def create(self, context):
+ context.yield_resources()
return None
diff --git a/extensions/python/pythonprocessors/nifiapi/properties.py b/extensions/python/pythonprocessors/nifiapi/properties.py
index 20bf34bb3..996a0bece 100644
--- a/extensions/python/pythonprocessors/nifiapi/properties.py
+++ b/extensions/python/pythonprocessors/nifiapi/properties.py
@@ -308,3 +308,6 @@ class ProcessContext:
properties[property_descriptor] = cpp_properties[property_descriptor.name]
return properties
+
+ def yield_resources(self):
+ self.cpp_context.yieldResources()
diff --git a/extensions/python/types/PyProcessContext.cpp b/extensions/python/types/PyProcessContext.cpp
index 5dd9628ff..cb98bddf0 100644
--- a/extensions/python/types/PyProcessContext.cpp
+++ b/extensions/python/types/PyProcessContext.cpp
@@ -32,6 +32,7 @@ static PyMethodDef PyProcessContext_methods[] = { // NOLINT(cppcoreguidelines-a
{"getControllerService", (PyCFunction) PyProcessContext::getControllerService, METH_VARARGS, nullptr},
{"getName", (PyCFunction) PyProcessContext::getName, METH_VARARGS, nullptr},
{"getProperties", (PyCFunction) PyProcessContext::getProperties, METH_VARARGS, nullptr},
+ {"yieldResources", (PyCFunction) PyProcessContext::getProperties, METH_VARARGS, nullptr},
{} /* Sentinel */
};
@@ -168,6 +169,18 @@ PyObject* PyProcessContext::getProperties(PyProcessContext* self, PyObject*) {
return object::returnReference(py_properties);
}
+PyObject* PyProcessContext::yieldResources(PyProcessContext* self, PyObject*) {
+ auto context = self->process_context_;
+ if (!context) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading process context outside 'on_trigger'");
+ return nullptr;
+ }
+
+ context->yield();
+
+ Py_RETURN_NONE;
+}
+
PyTypeObject* PyProcessContext::typeObject() {
static OwnedObject PyProcessContextType{PyType_FromSpec(&PyProcessContextTypeSpec)};
return reinterpret_cast<PyTypeObject*>(PyProcessContextType.get());
diff --git a/extensions/python/types/PyProcessContext.h b/extensions/python/types/PyProcessContext.h
index 00c6a04e6..8e3f515b4 100644
--- a/extensions/python/types/PyProcessContext.h
+++ b/extensions/python/types/PyProcessContext.h
@@ -39,6 +39,7 @@ struct PyProcessContext {
static PyObject* getControllerService(PyProcessContext* self, PyObject* args);
static PyObject* getName(PyProcessContext* self, PyObject* args);
static PyObject* getProperties(PyProcessContext* self, PyObject* args);
+ static PyObject* yieldResources(PyProcessContext* self, PyObject* args);
static PyTypeObject* typeObject();
};
(nifi-minifi-cpp) 01/04: MINIFICPP-2447 Include the Python processor version in the manifest
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 1eccfd8a2388807054dffe5de6cb7162bbe5b25c
Author: Ferenc Gerlits <fg...@gmail.com>
AuthorDate: Thu Sep 26 17:54:30 2024 +0200
MINIFICPP-2447 Include the Python processor version in the manifest
NiFi python processors which define a ProcessorDetails.version field should
show this version number in the manifest. For all other processors, we will
continue to use the MiNiFi version.
Closes #1861
Signed-off-by: Marton Szasz <sz...@apache.org>
---
extensions/python/ExecutePythonProcessor.h | 9 +
extensions/python/PythonCreator.h | 2 +-
extensions/python/PythonProcessor.cpp | 4 +
extensions/python/PythonProcessor.h | 2 +
.../pythonprocessors/nifiapi/processorbase.py | 3 +
extensions/python/tests/CMakeLists.txt | 1 +
extensions/python/tests/PythonManifestTests.cpp | 212 +++++++++++++++++----
extensions/python/types/PyProcessor.cpp | 16 ++
extensions/python/types/PyProcessor.h | 1 +
libminifi/test/libtest/unit/TestUtils.h | 12 ++
10 files changed, 227 insertions(+), 35 deletions(-)
diff --git a/extensions/python/ExecutePythonProcessor.h b/extensions/python/ExecutePythonProcessor.h
index f344f35af..8eec7011a 100644
--- a/extensions/python/ExecutePythonProcessor.h
+++ b/extensions/python/ExecutePythonProcessor.h
@@ -116,6 +116,14 @@ class ExecutePythonProcessor : public core::Processor {
return description_;
}
+ void setVersion(const std::string& version) {
+ version_ = version;
+ }
+
+ const std::optional<std::string>& getVersion() const {
+ return version_;
+ }
+
void setPythonClassName(const std::string& python_class_name) {
python_class_name_ = python_class_name;
}
@@ -140,6 +148,7 @@ class ExecutePythonProcessor : public core::Processor {
std::vector<core::Property> python_properties_;
std::string description_;
+ std::optional<std::string> version_;
bool processor_initialized_;
bool python_dynamic_;
diff --git a/extensions/python/PythonCreator.h b/extensions/python/PythonCreator.h
index 98717a2d5..dce65b356 100644
--- a/extensions/python/PythonCreator.h
+++ b/extensions/python/PythonCreator.h
@@ -113,7 +113,7 @@ class PythonCreator : public minifi::core::CoreComponent {
processor->initialize();
minifi::BundleDetails details;
details.artifact = path.filename().string();
- details.version = minifi::AgentBuild::VERSION;
+ details.version = processor->getVersion() && !processor->getVersion()->empty() ? *processor->getVersion() : minifi::AgentBuild::VERSION;
details.group = "python";
minifi::ClassDescription description{
diff --git a/extensions/python/PythonProcessor.cpp b/extensions/python/PythonProcessor.cpp
index 021c1e268..9fd7c26a8 100644
--- a/extensions/python/PythonProcessor.cpp
+++ b/extensions/python/PythonProcessor.cpp
@@ -37,6 +37,10 @@ void PythonProcessor::setDescription(const std::string& desc) {
processor_->setDescription(desc);
}
+void PythonProcessor::setVersion(const std::string& version) {
+ processor_->setVersion(version);
+}
+
void PythonProcessor::addProperty(const std::string& name, const std::string& description, const std::optional<std::string>& defaultvalue, bool required, bool el, bool sensitive,
const std::optional<int64_t>& property_type_code, gsl::span<const std::string_view> allowable_values, const std::optional<std::string>& controller_service_type_name) {
processor_->addProperty(name, description, defaultvalue, required, el, sensitive, property_type_code, allowable_values, controller_service_type_name);
diff --git a/extensions/python/PythonProcessor.h b/extensions/python/PythonProcessor.h
index 52a7b2a8b..4333aeaa9 100644
--- a/extensions/python/PythonProcessor.h
+++ b/extensions/python/PythonProcessor.h
@@ -40,6 +40,8 @@ class PythonProcessor {
void setDescription(const std::string& desc);
+ void setVersion(const std::string& version);
+
void addProperty(const std::string& name, const std::string& description, const std::optional<std::string>& defaultvalue, bool required, bool el, bool sensitive,
const std::optional<int64_t>& property_type_code, gsl::span<const std::string_view> allowable_values, const std::optional<std::string>& controller_service_type_name);
diff --git a/extensions/python/pythonprocessors/nifiapi/processorbase.py b/extensions/python/pythonprocessors/nifiapi/processorbase.py
index ee1034e7c..77f1b7114 100644
--- a/extensions/python/pythonprocessors/nifiapi/processorbase.py
+++ b/extensions/python/pythonprocessors/nifiapi/processorbase.py
@@ -42,6 +42,9 @@ class ProcessorBase(ABC):
else:
processor.setDescription(self.__class__.__name__)
+ if hasattr(self, 'ProcessorDetails') and hasattr(self.ProcessorDetails, 'version'):
+ processor.setVersion(self.ProcessorDetails.version)
+
def onInitialize(self, processor: Processor):
processor.setSupportsDynamicProperties()
for property in self.getPropertyDescriptors():
diff --git a/extensions/python/tests/CMakeLists.txt b/extensions/python/tests/CMakeLists.txt
index 813ecb177..06cc89e44 100644
--- a/extensions/python/tests/CMakeLists.txt
+++ b/extensions/python/tests/CMakeLists.txt
@@ -51,5 +51,6 @@ FOREACH(testfile ${EXECUTEPYTHONPROCESSOR_UNIT_TESTS})
add_test(NAME "${testfilename}" COMMAND "${testfilename}" WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR})
ENDFOREACH()
+copyTestResources(${CMAKE_SOURCE_DIR}/extensions/python/pythonprocessors/nifiapi ${CMAKE_BINARY_DIR}/bin/resources/minifi-python/nifiapi)
copyTestResources(${CMAKE_CURRENT_SOURCE_DIR}/test_python_scripts/ ${CMAKE_BINARY_DIR}/bin/resources/test_python_scripts/)
message("-- Finished building ${EXTENSIONS_TEST_COUNT} tests for minifi-python-script-extension...")
diff --git a/extensions/python/tests/PythonManifestTests.cpp b/extensions/python/tests/PythonManifestTests.cpp
index b0e58619a..c054d36f5 100644
--- a/extensions/python/tests/PythonManifestTests.cpp
+++ b/extensions/python/tests/PythonManifestTests.cpp
@@ -19,10 +19,12 @@
#include <fstream>
+#include "agent/agent_version.h"
#include "unit/TestBase.h"
#include "unit/Catch.h"
#include "unit/TestControllerWithFlow.h"
#include "unit/EmptyFlow.h"
+#include "unit/TestUtils.h"
#include "c2/C2MetricsPublisher.h"
#include "utils/gsl.h"
@@ -41,6 +43,7 @@ const SerializedResponseNode& getNode(const std::vector<SerializedResponseNode>&
for (auto& node : nodes) {
if (node.name == name) return node;
}
+ FAIL(fmt::format("Node {} was not found", name));
gsl_FailFast();
}
@@ -63,6 +66,96 @@ TEST_CASE("Python processor's description is part of the manifest") {
" proc.setSupportsDynamicProperties()\n"
" proc.addProperty('Prop1', 'A great property', 'banana', True, False, False, None, ['apple', 'orange', 'banana', 'durian'], None)\n";
+ const auto executable_dir = minifi::utils::file::FileUtils::get_executable_dir();
+#ifdef WIN32
+ std::filesystem::create_symlink(executable_dir / "minifi-python-script-extension.dll", python_dir / "minifi_native.pyd");
+#endif
+ std::filesystem::copy(executable_dir / "resources" / "minifi-python" / "nifiapi", python_dir / "nifiapi", std::filesystem::copy_options::recursive);
+ utils::file::create_dir(python_dir / "nifi_python_processors");
+ std::ofstream{python_dir / "nifi_python_processors" / "MyPyProc3.py"} << R"(
+from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult
+from nifiapi.properties import ExpressionLanguageScope, PropertyDescriptor, StandardValidators
+
+class MyPyProc3(FlowFileTransform):
+
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ class ProcessorDetails:
+ version = '1.2.3'
+ description = "Test processor number three."
+ dependencies = []
+
+ COLOR = PropertyDescriptor(
+ name="Color",
+ description="Symbolic name for the combination of frequencies of electromagnetic radiation reflected by the processor.",
+ allowable_values=['red', 'blue', 'green', 'purple'],
+ default_value='red',
+ required=True,
+ expression_language_scope=ExpressionLanguageScope.NONE
+ )
+
+ MOOD = PropertyDescriptor(
+ name="Mood",
+ description="The mental or emotional state of the processor.",
+ required=False,
+ validators=[StandardValidators.NON_EMPTY_VALIDATOR],
+ expression_language_scope=ExpressionLanguageScope.FLOWFILE_ATTRIBUTES
+ )
+
+ def __init__(self, **kwargs):
+ pass
+
+ def getPropertyDescriptors(self):
+ return [self.COLOR, self.MOOD]
+
+ def transform(self, context, flow_file):
+ color = context.getProperty(self.COLOR).getValue()
+ mood = context.getProperty(self.MOOD).evaluateAttributeExpressions(flowfile).getValue() or "OK"
+ user = flow_file.getContentsAsBytes().decode('utf-8')
+ response = f"Hello {user}! I am a {color} processor. I am feeling {mood}."
+ return FlowFileTransformResult('success', contents=response.encode('utf-8'))
+)";
+
+ std::ofstream{python_dir / "nifi_python_processors" / "MyPyProc4.py"} << R"(
+from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult
+
+class MyPyProc4(FlowFileTransform):
+
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ class ProcessorDetails:
+ description = "Test processor number four. Does not define a version."
+ dependencies = []
+
+ def __init__(self, **kwargs):
+ pass
+
+ def transform(self, context, flow_file):
+ return FlowFileTransformResult('success')
+)";
+
+ std::ofstream{python_dir / "nifi_python_processors" / "MyPyProc5.py"} << R"(
+from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult
+
+class MyPyProc5(FlowFileTransform):
+
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ class ProcessorDetails:
+ version = ''
+ description = "Test processor number five. Defines a version, but it is blank."
+ dependencies = []
+
+ def __init__(self, **kwargs):
+ pass
+
+ def transform(self, context, flow_file):
+ return FlowFileTransformResult('success')
+)";
+
controller.configuration_->set(minifi::Configuration::nifi_python_processor_dir, python_dir.string());
controller.configuration_->set(minifi::Configuration::nifi_extension_path, "*minifi-python-script*");
@@ -76,65 +169,65 @@ TEST_CASE("Python processor's description is part of the manifest") {
auto& manifest = getNode(agent_info.serialized_nodes, "agentManifest");
- auto findPythonProcessor = [&] (const std::string& name) {
+ auto findPythonBundle = [&](const std::string& name) {
// each python file gets its own bundle
- auto* python_bundle = findNode(manifest.children, [&] (auto& child) {
- return child.name == "bundles" && findNode(child.children, [&] (auto& bundle_child) {
+ auto* python_bundle = findNode(manifest.children, [&](const auto& child) {
+ return child.name == "bundles" && findNode(child.children, [&](const auto& bundle_child) {
return bundle_child.name == "artifact" && bundle_child.value == name + ".py";
});
});
REQUIRE(python_bundle);
+ return gsl::make_not_null(python_bundle);
+ };
- auto& py_processors = getNode(getNode(python_bundle->children, "componentManifest").children, "processors");
-
- // single processor in each bundle
- REQUIRE(py_processors.children.size() == 1);
-
- return findNode(py_processors.children, [&] (auto& child) {return utils::string::endsWith(child.name, name);});
+ auto getProcessorNode = [&] (gsl::not_null<const SerializedResponseNode*> bundle) {
+ auto& processors = getNode(getNode(bundle->children, "componentManifest").children, "processors");
+ REQUIRE(processors.children.size() == 1);
+ auto& only_child = processors.children[0];
+ return gsl::make_not_null(&only_child);
};
{
- auto* MyPyProc = findPythonProcessor("MyPyProc");
- REQUIRE(MyPyProc);
+ auto python_bundle = findPythonBundle("MyPyProc");
+ auto MyPyProc = getProcessorNode(python_bundle);
- REQUIRE(getNode(MyPyProc->children, "inputRequirement").value == "INPUT_ALLOWED");
- REQUIRE(getNode(MyPyProc->children, "isSingleThreaded").value == true);
- REQUIRE(getNode(MyPyProc->children, "typeDescription").value == "An amazing processor");
- REQUIRE(getNode(MyPyProc->children, "supportsDynamicRelationships").value == false);
- REQUIRE(getNode(MyPyProc->children, "supportsDynamicProperties").value == false);
- REQUIRE(getNode(MyPyProc->children, "type").value == "org.apache.nifi.minifi.processors.MyPyProc");
+ CHECK(getNode(MyPyProc->children, "inputRequirement").value == "INPUT_ALLOWED");
+ CHECK(getNode(MyPyProc->children, "isSingleThreaded").value == true);
+ CHECK(getNode(MyPyProc->children, "typeDescription").value == "An amazing processor");
+ CHECK(getNode(MyPyProc->children, "supportsDynamicRelationships").value == false);
+ CHECK(getNode(MyPyProc->children, "supportsDynamicProperties").value == false);
+ CHECK(getNode(MyPyProc->children, "type").value == "org.apache.nifi.minifi.processors.MyPyProc");
auto& rels = getNode(MyPyProc->children, "supportedRelationships").children;
REQUIRE(rels.size() == 3);
auto* success = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "success";});
REQUIRE(success);
- REQUIRE(getNode(success->children, "description").value == "Script succeeds");
+ CHECK(getNode(success->children, "description").value == "Script succeeds");
auto* failure = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "failure";});
REQUIRE(failure);
REQUIRE(getNode(failure->children, "description").value == "Script fails");
}
-
{
- auto* MyPyProc2 = findPythonProcessor("MyPyProc2");
- REQUIRE(MyPyProc2);
+ auto python_bundle = findPythonBundle("MyPyProc2");
+ auto MyPyProc2 = getProcessorNode(python_bundle);
- REQUIRE(getNode(MyPyProc2->children, "inputRequirement").value == "INPUT_ALLOWED");
- REQUIRE(getNode(MyPyProc2->children, "isSingleThreaded").value == true);
- REQUIRE(getNode(MyPyProc2->children, "typeDescription").value == "Another amazing processor");
- REQUIRE(getNode(MyPyProc2->children, "supportsDynamicRelationships").value == false);
- REQUIRE(getNode(MyPyProc2->children, "supportsDynamicProperties").value == true);
- REQUIRE(getNode(MyPyProc2->children, "type").value == "org.apache.nifi.minifi.processors.MyPyProc2");
+ CHECK(getNode(MyPyProc2->children, "inputRequirement").value == "INPUT_ALLOWED");
+ CHECK(getNode(MyPyProc2->children, "isSingleThreaded").value == true);
+ CHECK(getNode(MyPyProc2->children, "typeDescription").value == "Another amazing processor");
+ CHECK(getNode(MyPyProc2->children, "supportsDynamicRelationships").value == false);
+ CHECK(getNode(MyPyProc2->children, "supportsDynamicProperties").value == true);
+ CHECK(getNode(MyPyProc2->children, "type").value == "org.apache.nifi.minifi.processors.MyPyProc2");
auto& properties = getNode(MyPyProc2->children, "propertyDescriptors").children;
REQUIRE(properties.size() == 1);
- REQUIRE(properties[0].name == "Prop1");
- REQUIRE(getNode(properties[0].children, "name").value == "Prop1");
- REQUIRE(getNode(properties[0].children, "required").value == true);
- REQUIRE(getNode(properties[0].children, "expressionLanguageScope").value == "NONE");
- REQUIRE(getNode(properties[0].children, "defaultValue").value == "banana");
+ CHECK(properties[0].name == "Prop1");
+ CHECK(getNode(properties[0].children, "name").value == "Prop1");
+ CHECK(getNode(properties[0].children, "required").value == true);
+ CHECK(getNode(properties[0].children, "expressionLanguageScope").value == "NONE");
+ CHECK(getNode(properties[0].children, "defaultValue").value == "banana");
auto& allowable_values = getNode(properties[0].children, "allowableValues");
REQUIRE(allowable_values.children.size() == 4);
CHECK(getNthAllowableValue(allowable_values, 0) == "apple");
@@ -147,10 +240,61 @@ TEST_CASE("Python processor's description is part of the manifest") {
auto* success = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "success";});
REQUIRE(success);
- REQUIRE(getNode(success->children, "description").value == "Script succeeds");
+ CHECK(getNode(success->children, "description").value == "Script succeeds");
auto* failure = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "failure";});
REQUIRE(failure);
- REQUIRE(getNode(failure->children, "description").value == "Script fails");
+ CHECK(getNode(failure->children, "description").value == "Script fails");
+ }
+
+ {
+ auto python_bundle = findPythonBundle("MyPyProc3");
+ auto MyPyProc3 = getProcessorNode(python_bundle);
+
+ CHECK(getNode(python_bundle->children, "version").value == "1.2.3");
+
+ CHECK(getNode(MyPyProc3->children, "inputRequirement").value == "INPUT_ALLOWED");
+ CHECK(getNode(MyPyProc3->children, "isSingleThreaded").value == true);
+ CHECK(getNode(MyPyProc3->children, "typeDescription").value == "Test processor number three.");
+ CHECK(getNode(MyPyProc3->children, "supportsDynamicRelationships").value == false);
+ CHECK(getNode(MyPyProc3->children, "supportsDynamicProperties").value == true);
+ CHECK(getNode(MyPyProc3->children, "type").value == "org.apache.nifi.minifi.processors.nifi_python_processors.MyPyProc3");
+
+ auto& properties = getNode(MyPyProc3->children, "propertyDescriptors").children;
+ REQUIRE(properties.size() == 2);
+ CHECK(properties[0].name == "Color");
+ CHECK(getNode(properties[0].children, "name").value == "Color");
+ CHECK(getNode(properties[0].children, "required").value == true);
+ CHECK(getNode(properties[0].children, "expressionLanguageScope").value == "NONE");
+ CHECK(getNode(properties[0].children, "defaultValue").value == "red");
+ CHECK(properties[1].name == "Mood");
+ CHECK(getNode(properties[1].children, "name").value == "Mood");
+ CHECK(getNode(properties[1].children, "required").value == false);
+ CHECK(getNode(properties[1].children, "expressionLanguageScope").value == "FLOWFILE_ATTRIBUTES");
+
+ auto& rels = getNode(MyPyProc3->children, "supportedRelationships").children;
+ REQUIRE(rels.size() == 3);
+
+ auto* success = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "success";});
+ REQUIRE(success);
+ CHECK(getNode(success->children, "description").value == "Script succeeds");
+
+ auto* failure = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "failure";});
+ REQUIRE(failure);
+ CHECK(getNode(failure->children, "description").value == "Script fails");
+
+ auto* original = findNode(rels, [] (auto& rel) {return getNode(rel.children, "name").value == "original";});
+ REQUIRE(original);
+ CHECK(getNode(original->children, "description").value == "Original flow file");
+ }
+
+ {
+ auto python_bundle = findPythonBundle("MyPyProc4");
+ CHECK(getNode(python_bundle->children, "version").value == minifi::AgentBuild::VERSION);
+ }
+
+ {
+ auto python_bundle = findPythonBundle("MyPyProc5");
+ CHECK(getNode(python_bundle->children, "version").value == minifi::AgentBuild::VERSION);
}
}
diff --git a/extensions/python/types/PyProcessor.cpp b/extensions/python/types/PyProcessor.cpp
index dd8ad17b2..11eda4713 100644
--- a/extensions/python/types/PyProcessor.cpp
+++ b/extensions/python/types/PyProcessor.cpp
@@ -44,6 +44,7 @@ extern "C" {
static PyMethodDef PyProcessor_methods[] = { // NOLINT(cppcoreguidelines-avoid-c-arrays)
{"setSupportsDynamicProperties", (PyCFunction) PyProcessor::setSupportsDynamicProperties, METH_VARARGS, nullptr},
{"setDescription", (PyCFunction) PyProcessor::setDescription, METH_VARARGS, nullptr},
+ {"setVersion", (PyCFunction) PyProcessor::setVersion, METH_VARARGS, nullptr},
{"addProperty", (PyCFunction) PyProcessor::addProperty, METH_VARARGS, nullptr},
{} /* Sentinel */
};
@@ -103,6 +104,21 @@ PyObject* PyProcessor::setDescription(PyProcessor* self, PyObject* args) {
Py_RETURN_NONE;
}
+PyObject* PyProcessor::setVersion(PyProcessor* self, PyObject* args) {
+ auto processor = self->processor_.lock();
+ if (!processor) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading processor outside 'on_trigger'");
+ return nullptr;
+ }
+
+ const char* version = nullptr;
+ if (!PyArg_ParseTuple(args, "s", &version)) {
+ return nullptr;
+ }
+ processor->setVersion(std::string(version));
+ Py_RETURN_NONE;
+}
+
PyObject* PyProcessor::addProperty(PyProcessor* self, PyObject* args) {
auto processor = self->processor_.lock();
if (!processor) {
diff --git a/extensions/python/types/PyProcessor.h b/extensions/python/types/PyProcessor.h
index b2a1dc850..ff9e804a8 100644
--- a/extensions/python/types/PyProcessor.h
+++ b/extensions/python/types/PyProcessor.h
@@ -33,6 +33,7 @@ struct PyProcessor {
static PyObject* setSupportsDynamicProperties(PyProcessor* self, PyObject* args);
static PyObject* setDescription(PyProcessor* self, PyObject* args);
+ static PyObject* setVersion(PyProcessor* self, PyObject* args);
static PyObject* addProperty(PyProcessor* self, PyObject* args);
static PyTypeObject* typeObject();
diff --git a/libminifi/test/libtest/unit/TestUtils.h b/libminifi/test/libtest/unit/TestUtils.h
index 2cb7cb131..6303b1f5f 100644
--- a/libminifi/test/libtest/unit/TestUtils.h
+++ b/libminifi/test/libtest/unit/TestUtils.h
@@ -26,11 +26,14 @@
#include <utility>
#include <vector>
+#include "core/state/Value.h"
#include "utils/file/FileUtils.h"
#include "utils/Id.h"
#include "utils/TimeUtil.h"
#include "TestBase.h"
+#include "catch2/catch_tostring.hpp"
+#include "fmt/format.h"
#include "rapidjson/document.h"
#include "asio.hpp"
#include "asio/ssl.hpp"
@@ -229,3 +232,12 @@ inline bool runningAsUnixRoot() {
#endif
}
} // namespace org::apache::nifi::minifi::test::utils
+
+namespace Catch {
+template <>
+struct StringMaker<minifi::state::response::ValueNode> {
+ static std::string convert(const minifi::state::response::ValueNode& value_node) {
+ return fmt::format(R"("{}")", value_node.to_string());
+ }
+};
+} // namespace Catch
(nifi-minifi-cpp) 03/04: MINIFICPP-2456 Remove curl output from Elasticsearch responses
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit a6ef49d7ae296061161e30e972cccaf13080342a
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Sep 26 17:55:49 2024 +0200
MINIFICPP-2456 Remove curl output from Elasticsearch responses
Closes #1867
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.../integration/cluster/checkers/ElasticSearchChecker.py | 14 ++++++++------
docker/test/integration/features/utils.py | 2 +-
2 files changed, 9 insertions(+), 7 deletions(-)
diff --git a/docker/test/integration/cluster/checkers/ElasticSearchChecker.py b/docker/test/integration/cluster/checkers/ElasticSearchChecker.py
index 34186cc4e..e35baa25c 100644
--- a/docker/test/integration/cluster/checkers/ElasticSearchChecker.py
+++ b/docker/test/integration/cluster/checkers/ElasticSearchChecker.py
@@ -20,27 +20,29 @@ class ElasticSearchChecker:
self.container_communicator = container_communicator
def is_elasticsearch_empty(self, container_name):
- (code, output) = self.container_communicator.execute_command(container_name, ["curl", "-u", "elastic:password", "-k", "-XGET", "https://localhost:9200/_search"])
+ (code, output) = self.container_communicator.execute_command(container_name, ["curl", "-s", "-u", "elastic:password", "-k", "-XGET", "https://localhost:9200/_search"])
return code == 0 and '"hits":[]' in output
def create_doc_elasticsearch(self, container_name, index_name, doc_id):
(code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c",
- "curl -u elastic:password -k -XPUT https://localhost:9200/" + index_name + "/_doc/" + doc_id + " -H Content-Type:application/json -d'{\"field1\":\"value1\"}'"])
+ "curl -s -u elastic:password -k -XPUT https://localhost:9200/" + index_name + "/_doc/" + doc_id + " -H Content-Type:application/json -d'{\"field1\":\"value1\"}'"])
return code == 0 and ('"_id":"' + doc_id + '"') in output
def check_elastic_field_value(self, container_name, index_name, doc_id, field_name, field_value):
(code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c",
- "curl -u elastic:password -k -XGET https://localhost:9200/" + index_name + "/_doc/" + doc_id])
+ "curl -s -u elastic:password -k -XGET https://localhost:9200/" + index_name + "/_doc/" + doc_id])
return code == 0 and (field_name + '":"' + field_value) in output
def elastic_generate_apikey(self, elastic_container_name):
- (_, output) = self.container_communicator.execute_command(elastic_container_name, ["/bin/bash", "-c",
- "curl -u elastic:password -k -XPOST https://localhost:9200/_security/api_key -H Content-Type:application/json -d'{\"name\":\"my-api-key\",\"expiration\":\"1d\",\"role_descriptors\":{\"role-a\": {\"cluster\": [\"all\"],\"index\": [{\"names\": [\"my_index\"],\"privileges\": [\"all\"]}]}}}'"])
+ (code, output) = self.container_communicator.execute_command(elastic_container_name, ["/bin/bash", "-c",
+ "curl -s -u elastic:password -k -XPOST https://localhost:9200/_security/api_key -H Content-Type:application/json -d'{\"name\":\"my-api-key\",\"expiration\":\"1d\",\"role_descriptors\":{\"role-a\": {\"cluster\": [\"all\"],\"index\": [{\"names\": [\"my_index\"],\"privileges\": [\"all\"]}]}}}'"])
+ if code != 0:
+ return None
output_lines = output.splitlines()
result = json.loads(output_lines[-1])
return result["encoded"]
def add_elastic_user_to_opensearch(self, container_name):
(code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c",
- 'curl -u admin:admin -k -XPUT https://{hostname}:9200/_plugins/_security/api/internalusers/elastic -H Content-Type:application/json -d\'{{"password":"password","backend_roles":["admin"]}}\''.format(hostname=container_name)])
+ 'curl -s -u admin:admin -k -XPUT https://{hostname}:9200/_plugins/_security/api/internalusers/elastic -H Content-Type:application/json -d\'{{"password":"password","backend_roles":["admin"]}}\''.format(hostname=container_name)])
return code == 0 and '"status":"CREATED"' in output
diff --git a/docker/test/integration/features/utils.py b/docker/test/integration/features/utils.py
index 15db614d2..bd86cbab0 100644
--- a/docker/test/integration/features/utils.py
+++ b/docker/test/integration/features/utils.py
@@ -25,7 +25,7 @@ def retry_check(max_tries=5, retry_interval=1):
def retry_check_func(func):
@functools.wraps(func)
def retry_wrapper(*args, **kwargs):
- for i in range(max_tries):
+ for _ in range(max_tries):
if func(*args, **kwargs):
return True
time.sleep(retry_interval)