lordgamez commented on a change in pull request #1076:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1076#discussion_r648050330
##########
File path: docker/test/integration/steps/steps.py
##########
@@ -121,15 +133,71 @@ def step_impl(context):
context.test.add_node(publish_kafka)
+@given("a kafka producer workflow publishing files placed in \"{directory}\"
to a broker exactly once")
+def step_impl(context, directory):
+ context.execute_steps("""
+ given a GetFile processor with the \"Input Directory\" property set to
\"{directory}\"
+ and the \"Keep Source File\" of the GetFile processor is set to
\"false\"
+ and a PublishKafka processor set up to communicate with a kafka broker
instance
+ and the "success" relationship of the GetFile processor is connected
to the PublishKafka""".format(directory=directory))
+
+
+@given("a ConsumeKafka processor set up in a \"{cluster_name}\" flow")
+def step_impl(context, cluster_name):
+ consume_kafka = ConsumeKafka()
+ consume_kafka.set_name("ConsumeKafka")
+ context.test.add_node(consume_kafka)
+ logging.info("Acquiring " + cluster_name)
+ cluster = context.test.acquire_cluster(cluster_name)
+ # Assume that the first node declared is primary unless specified otherwise
+ if cluster.get_flow() is None:
+ cluster.set_name(cluster_name)
+ cluster.set_flow(consume_kafka)
+
+
@given("the \"{property_name}\" of the {processor_name} processor is set to
\"{property_value}\"")
def step_impl(context, property_name, processor_name, property_value):
processor = context.test.get_node_by_name(processor_name)
- processor.set_property(property_name, property_value)
+ if property_value == "(not set)":
+ processor.unset_property(property_name)
+ else:
+ processor.set_property(property_name, property_value)
+
+
+@given("the \"{property_name}\" of the {processor_name} processor is set to
match {key_attribute_encoding} encoded kafka message key \"{message_key}\"")
Review comment:
Update in cdb014f306f614bfad5e6120f089ab821c904677
##########
File path: docker/test/integration/environment.py
##########
@@ -11,20 +12,31 @@ def raise_exception(exception):
raise exception
-@fixture
-def test_driver_fixture(context):
- context.test = MiNiFi_integration_test(context)
- yield context.test
- logging.info("Integration test teardown...")
- del context.test
+def integration_test_cleanup(test):
+ logging.info("Integration test cleanup...")
+ del test
def before_scenario(context, scenario):
- use_fixture(test_driver_fixture, context)
+ if "skip" in scenario.effective_tags:
+ scenario.skip("Marked with @skip")
+ return
+
+ logging.info("Integration test setup at
{time:%H:%M:%S:%f}".format(time=datetime.datetime.now()))
Review comment:
Update in cdb014f306f614bfad5e6120f089ab821c904677
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
