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
The following commit(s) were added to refs/heads/main by this push:
new 69a024228 MINIFICPP-2905 Run valgrind on all unit tests (#2282)
69a024228 is described below
commit 69a02422803151c016e55de0b792c00e55843546
Author: Ferenc Gerlits <[email protected]>
AuthorDate: Thu Oct 22 11:37:47 2026 +0200
MINIFICPP-2905 Run valgrind on all unit tests (#2282)
* MINIFICPP-2905 Run valgrind on all unit tests
Remove the "memchecked" tag from the unit tests and run valgrind on
almost all tests instead of only the tagged tests.
Also:
* increase timeouts by a factor of 50 when running under Valgrind
(this affects failures only; on success, the verifier returns early)
* skip libminifi/test/integration tests: these are too dependent on timing
and become flaky under valgrind
* skip some more tests which don't work well under valgrind
* add/expand suppressions for some known non-issues
* fix typo: 'phd' to 'pdh'
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* workaround for GCC 15/16 bug
https://gcc.gnu.org/bugzilla/show_bug.cgi?id=124584
asio::experimental::awaitable_operators::operator|| is broken due to this
bug,
so use a (more verbose) hand-rolled alternative
* narrow the valgrind suppression
* move the type from comment to code
* revert the name of the LogCompressorSink valgrind suppression rule
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.github/workflows/memcheck_ci.yml | 17 +++++----
cmake/valgrind.supp | 44 ++++++++++++++++++++++
core-framework/include/utils/net/AsioCoro.h | 18 ++++++---
extensions/azure/tests/CMakeLists.txt | 2 +-
extensions/bustache/tests/CMakeLists.txt | 2 +-
extensions/civetweb/tests/C2VerifyServeResults.cpp | 2 +-
.../civetweb/tests/HttpPostIntegrationTest.cpp | 2 +-
extensions/elasticsearch/tests/CMakeLists.txt | 2 +-
extensions/execute-process/test/CMakeLists.txt | 2 +-
extensions/grafana-loki/tests/CMakeLists.txt | 2 +-
extensions/kafka/tests/CMakeLists.txt | 2 +-
extensions/kubernetes/tests/CMakeLists.txt | 2 +-
extensions/llamacpp/tests/CMakeLists.txt | 2 +-
extensions/lua/tests/CMakeLists.txt | 2 +-
extensions/mqtt/tests/CMakeLists.txt | 2 +-
extensions/opc/tests/CMakeLists.txt | 2 +-
extensions/pdh/tests/CMakeLists.txt | 2 +-
extensions/procfs/tests/CMakeLists.txt | 2 +-
extensions/prometheus/tests/CMakeLists.txt | 2 +-
extensions/rocksdb-repos/tests/RepoTests.cpp | 5 +++
extensions/stable-api-testing/tests/CMakeLists.txt | 2 +-
.../tests/integration/ProcessGroupTest.cpp | 3 +-
.../tests/unit/AttributeRollingWindowTests.cpp | 12 ++++--
.../standard-processors/tests/unit/GetTCPTests.cpp | 6 ++-
.../tests/unit/ProcessorTests.cpp | 2 +-
.../standard-processors/tests/unit/PutTCPTests.cpp | 8 ++--
libminifi/test/flow-tests/FlowControllerTests.cpp | 17 ++-------
libminifi/test/integration/CMakeLists.txt | 2 +
.../test/libtest/integration/HTTPIntegrationBase.h | 12 +++---
.../test/libtest/integration/IntegrationBase.cpp | 1 +
.../test/libtest/integration/IntegrationBase.h | 7 ++--
.../libtest/unit/SingleProcessorTestController.cpp | 3 +-
libminifi/test/libtest/unit/TestUtils.cpp | 11 +++++-
libminifi/test/libtest/unit/TestUtils.h | 13 ++++++-
libminifi/test/unit/FlowFileQueueTests.cpp | 26 ++++++-------
libminifi/test/unit/MemoryUsageTest.cpp | 5 +++
libminifi/test/unit/MinifiConcurrentQueueTests.cpp | 23 ++++++++---
37 files changed, 183 insertions(+), 86 deletions(-)
diff --git a/.github/workflows/memcheck_ci.yml
b/.github/workflows/memcheck_ci.yml
index 6fc127a85..76e4201c2 100644
--- a/.github/workflows/memcheck_ci.yml
+++ b/.github/workflows/memcheck_ci.yml
@@ -17,8 +17,8 @@ env:
CCACHE_DIR: ${{ GITHUB.WORKSPACE }}/.ccache
jobs:
valgrind:
- name: "Valgrind on ubuntu-24.04"
- runs-on: ubuntu-24.04
+ name: "Valgrind"
+ runs-on: ubuntu-26.04
timeout-minutes: 300
steps:
- id: checkout
@@ -27,10 +27,10 @@ jobs:
uses: actions/cache/restore@v5
with:
path: ${{ env.CCACHE_DIR }}
- key: ubuntu-24.04-valgrind-ccache-${{github.ref}}-${{github.sha}}
+ key: ubuntu-26.04-valgrind-ccache-${{github.ref}}-${{github.sha}}
restore-keys: |
- ubuntu-24.04-valgrind-ccache-${{github.ref}}-
- ubuntu-24.04-valgrind-ccache-refs/heads/main-
+ ubuntu-26.04-valgrind-ccache-${{github.ref}}-
+ ubuntu-26.04-valgrind-ccache-refs/heads/main-
- id: install_deps
run: |
sudo apt update
@@ -48,16 +48,17 @@ jobs:
if: always()
with:
path: ${{ env.CCACHE_DIR }}
- key: ubuntu-24.04-valgrind-ccache-${{github.ref}}-${{github.sha}}
+ key: ubuntu-26.04-valgrind-ccache-${{github.ref}}-${{github.sha}}
- name: test
id: test
run: |
# Set core file size limit to unlimited
ulimit -c unlimited
- ctest -j$(nproc) -L memchecked -T memcheck
+ ctest -j$(nproc) -T memcheck -LE integration --output-on-failure
working-directory: build
- name: check valgrind output
id: check
+ if: always()
run: |
nonempty_valgrind_outputs=$(find Testing/Temporary -maxdepth 1 -name
'MemoryChecker.*.log' -size +0 -print)
if [ -n "$nonempty_valgrind_outputs" ]; then
@@ -76,7 +77,7 @@ jobs:
build/Testing/Temporary/LastDynamicAnalysis_*.log
if-no-files-found: ignore
address-sanitizer:
- name: "AddressSanitizer+LeakSanitizer on ubuntu 26.04"
+ name: "AddressSanitizer+LeakSanitizer"
runs-on: ubuntu-26.04
timeout-minutes: 300
steps:
diff --git a/cmake/valgrind.supp b/cmake/valgrind.supp
index 9beb95743..184303afc 100644
--- a/cmake/valgrind.supp
+++ b/cmake/valgrind.supp
@@ -45,6 +45,8 @@
match-leak-kinds: possible
fun:calloc
...
+ fun:_dl_allocate_tls
+ ...
fun:pthread_create*
...
fun:*LogCompressorSink*
@@ -91,6 +93,24 @@
...
}
+# glibc lazily allocates a module's Thread-Local-Storage block (a DTV entry)
the first time a
+# thread touches a thread_local in a dynamically-loaded library (e.g. the Lua
script extension):
+# __tls_get_addr -> tls_get_addr_tail -> allocate_and_init ->
allocate_dtv_entry. The block is
+# owned by the dynamic linker's DTV and freed by the OS at exit; valgrind only
holds an interior
+# pointer, so it is flagged "possibly lost". Sibling path to
possibly_lost_tls_dtv_growth (which
+# matches DTV *resize* instead). Never a real leak.
+{
+ possibly_lost_tls_dtv_entry_allocation
+ Memcheck:Leak
+ match-leak-kinds: possible
+ fun:malloc
+ ...
+ fun:allocate_dtv_entry
+ ...
+ fun:__tls_get_addr
+ ...
+}
+
# glibc grows the Thread-Local-Storage descriptor vector (DTV) the first time a
# thread touches a dynamically-loaded TLS variable. The DTV block is kept for
# the thread's lifetime and freed by the OS at exit; the growth happens only
@@ -183,3 +203,27 @@
fun:*CollectionScope*
...
}
+
+# Suppress a "Conditional jump or move depends on uninitialised value(s)" error
+# in zlib: this is intentional optimization, with the comment "prev[n] is
garbage
+# but its value will never be used" at deflate.c:204.
+{
+ zlib_slide_hash_uninitialised_prev
+ Memcheck:Cond
+ fun:slide_hash
+ fun:fill_window
+ ...
+}
+
+# "Conditional jump or move depends on uninitialised value(s)" inside the
libstdc++
+# std::from_chars<int> that is inlined into Cron's fromChars<>/parse<year>
+# (utils/Cron.cpp). The input std::string is fully initialised and fromChars
value-
+# initialises its output; this is a known valgrind imprecision on optimized
integer
+# parsing, not a real defect.
+{
+ cron_parse_year_fromchars_uninitialised
+ Memcheck:Cond
+ fun:*parse*year*
+ fun:*Cron*
+ ...
+}
diff --git a/core-framework/include/utils/net/AsioCoro.h
b/core-framework/include/utils/net/AsioCoro.h
index b18087268..4b38a428a 100644
--- a/core-framework/include/utils/net/AsioCoro.h
+++ b/core-framework/include/utils/net/AsioCoro.h
@@ -28,7 +28,9 @@
#include "asio/steady_timer.hpp"
#include "asio/this_coro.hpp"
#include "asio/use_awaitable.hpp"
-#include "asio/experimental/awaitable_operators.hpp"
+#include "asio/co_spawn.hpp"
+#include "asio/deferred.hpp"
+#include "asio/experimental/parallel_group.hpp"
#include "asio/as_tuple.hpp"
namespace org::apache::nifi::minifi::utils::net {
@@ -57,13 +59,19 @@ inline asio::awaitable<void>
async_wait(std::chrono::steady_clock::duration dura
template<class... Types>
asio::awaitable<std::tuple<std::error_code, Types...>>
asyncOperationWithTimeout(asio::awaitable<std::tuple<std::error_code,
Types...>>&& async_operation,
std::chrono::steady_clock::duration timeout_duration) {
- using asio::experimental::awaitable_operators::operator||;
- auto operation_result = co_await(std::move(async_operation) ||
async_wait(timeout_duration)); // NOLINT
- if (operation_result.index() == 1) {
+ auto executor = co_await asio::this_coro::executor; //
NOLINT(clang-analyzer-core.CallAndMessage)
+ std::tuple<std::array<std::size_t, 2>, std::exception_ptr,
std::tuple<std::error_code, Types...>, std::exception_ptr>
parallel_group_result{
+ co_await asio::experimental::make_parallel_group( //
NOLINT(clang-analyzer-core.CallAndMessage)
+ asio::co_spawn(executor, std::move(async_operation), asio::deferred),
+ asio::co_spawn(executor, async_wait(timeout_duration),
asio::deferred))
+ .async_wait(asio::experimental::wait_for_one_success(),
asio::use_awaitable)
+ };
+ auto index_of_operation_which_finished_first =
std::get<0>(parallel_group_result)[0]; // 0 = async_operation, 1 =
async_wait(timeout_duration)
+ if (index_of_operation_which_finished_first != 0) {
std::tuple<std::error_code, Types...> result;
std::get<0>(result) = asio::error::timed_out;
co_return result;
}
- co_return std::get<0>(operation_result);
+ co_return std::move(std::get<2>(parallel_group_result));
}
} // namespace org::apache::nifi::minifi::utils::net
diff --git a/extensions/azure/tests/CMakeLists.txt
b/extensions/azure/tests/CMakeLists.txt
index 5eac1ab38..9c26b57d2 100644
--- a/extensions/azure/tests/CMakeLists.txt
+++ b/extensions/azure/tests/CMakeLists.txt
@@ -34,6 +34,6 @@ FOREACH(testfile ${AZURE_INTEGRATION_TESTS})
target_link_libraries(${testfilename} minifi-standard-processors)
MATH(EXPR AZURE_TEST_COUNT "${AZURE_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"azure;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "azure")
ENDFOREACH()
message("-- Finished building ${AZURE_TEST_COUNT} Azure related test
file(s)...")
diff --git a/extensions/bustache/tests/CMakeLists.txt
b/extensions/bustache/tests/CMakeLists.txt
index 2a87b0ef0..8fd537e16 100644
--- a/extensions/bustache/tests/CMakeLists.txt
+++ b/extensions/bustache/tests/CMakeLists.txt
@@ -34,6 +34,6 @@ FOREACH(testfile ${BUSTACHE_INTEGRATION_TESTS})
target_link_libraries(${testfilename} Catch2WithMain)
MATH(EXPR BUSTACHE-EXTENSIONS_TEST_COUNT
"${BUSTACHE-EXTENSIONS_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"bustache;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "bustache")
ENDFOREACH()
message("-- Finished building ${BUSTACHE-EXTENSIONS_TEST_COUNT} Bustache
related test file(s)...")
diff --git a/extensions/civetweb/tests/C2VerifyServeResults.cpp
b/extensions/civetweb/tests/C2VerifyServeResults.cpp
index 9e3de470c..7c46e2c82 100644
--- a/extensions/civetweb/tests/C2VerifyServeResults.cpp
+++ b/extensions/civetweb/tests/C2VerifyServeResults.cpp
@@ -53,7 +53,7 @@ class VerifyC2Server : public HTTPIntegrationBase {
}
void runAssertions() override {
-
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_),
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(wait_time_,
"Import offset 0",
"Outputting success and response"));
}
diff --git a/extensions/civetweb/tests/HttpPostIntegrationTest.cpp
b/extensions/civetweb/tests/HttpPostIntegrationTest.cpp
index e61a4e5f9..359868532 100644
--- a/extensions/civetweb/tests/HttpPostIntegrationTest.cpp
+++ b/extensions/civetweb/tests/HttpPostIntegrationTest.cpp
@@ -66,7 +66,7 @@ class HttpTestHarness : public HTTPIntegrationBase {
}
void runAssertions() override {
-
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_),
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(wait_time_,
"curl performed",
"Size:1024 Offset:0"));
REQUIRE_FALSE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::milliseconds(200),
"Size:0 Offset:0"));
diff --git a/extensions/elasticsearch/tests/CMakeLists.txt
b/extensions/elasticsearch/tests/CMakeLists.txt
index ea63fff49..1b91fbafe 100644
--- a/extensions/elasticsearch/tests/CMakeLists.txt
+++ b/extensions/elasticsearch/tests/CMakeLists.txt
@@ -32,6 +32,6 @@ FOREACH(testfile ${ELASTICSEARCH_TESTS})
target_link_libraries(${testfilename} Catch2WithMain
civetweb::civetweb-cpp minifi-elasticsearch minifi-standard-processors)
MATH(EXPR ELASTICSEARCH_TEST_COUNT "${ELASTICSEARCH_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"elasticsearch;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "elasticsearch")
ENDFOREACH()
message("-- Finished building ${ELASTICSEARCH_TEST_COUNT} Elasticsearch
related test file(s)...")
diff --git a/extensions/execute-process/test/CMakeLists.txt
b/extensions/execute-process/test/CMakeLists.txt
index 96f4a0e51..70efe71b1 100644
--- a/extensions/execute-process/test/CMakeLists.txt
+++ b/extensions/execute-process/test/CMakeLists.txt
@@ -27,7 +27,7 @@ target_include_directories(ExecuteProcessTests PRIVATE BEFORE
"${CMAKE_SOURCE_DI
target_compile_definitions(ExecuteProcessTests PRIVATE
"MINIFI_EXTENSION_GROUP_NAME=org.apache.nifi.minifi.test")
add_test(NAME ExecuteProcessTests COMMAND ExecuteProcessTests
WORKING_DIRECTORY ${TEST_DIR})
-set_tests_properties(ExecuteProcessTests PROPERTIES LABELS
"execute-process;memchecked")
+set_tests_properties(ExecuteProcessTests PROPERTIES LABELS "execute-process")
createTests(ExecuteProcessTests)
diff --git a/extensions/grafana-loki/tests/CMakeLists.txt
b/extensions/grafana-loki/tests/CMakeLists.txt
index ee9ee6246..d048b9bf2 100644
--- a/extensions/grafana-loki/tests/CMakeLists.txt
+++ b/extensions/grafana-loki/tests/CMakeLists.txt
@@ -41,6 +41,6 @@ FOREACH(testfile ${GRAFANA_LOKI_TESTS})
target_link_libraries(${testfilename} Catch2WithMain
civetweb::civetweb-cpp minifi-grafana-loki minifi-standard-processors)
MATH(EXPR GRAFANA_LOKI_TEST_COUNT "${GRAFANA_LOKI_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"grafana-loki;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "grafana-loki")
ENDFOREACH()
message("-- Finished building ${GRAFANA_LOKI_TEST_COUNT} Grafana Loki related
test file(s)...")
diff --git a/extensions/kafka/tests/CMakeLists.txt
b/extensions/kafka/tests/CMakeLists.txt
index 01cbd1a2f..42e634d1b 100644
--- a/extensions/kafka/tests/CMakeLists.txt
+++ b/extensions/kafka/tests/CMakeLists.txt
@@ -32,6 +32,6 @@ FOREACH (testfile ${KAFKA_INTEGRATION_TESTS})
target_link_libraries(${testfilename} mock-minifi)
MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"librdkafka;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "librdkafka")
ENDFOREACH()
message("-- Finished building ${KAFKA-EXTENSIONS_TEST_COUNT} Lib Kafka related
test file(s)...")
diff --git a/extensions/kubernetes/tests/CMakeLists.txt
b/extensions/kubernetes/tests/CMakeLists.txt
index 3a037ecc0..26127ae74 100644
--- a/extensions/kubernetes/tests/CMakeLists.txt
+++ b/extensions/kubernetes/tests/CMakeLists.txt
@@ -30,6 +30,6 @@ FOREACH(TEST_FILE ${KUBERNETES_UNIT_TESTS})
createTests("${TEST_TARGET}")
add_test(NAME ${TEST_TARGET} COMMAND "${TEST_TARGET}" WORKING_DIRECTORY
"${TEST_DIR}")
MATH(EXPR KUBERNETES_UNIT_TEST_COUNT "${KUBERNETES_UNIT_TEST_COUNT}+1")
- set_tests_properties("${TEST_TARGET}" PROPERTIES LABELS
"kubernetes;memchecked")
+ set_tests_properties("${TEST_TARGET}" PROPERTIES LABELS "kubernetes")
ENDFOREACH()
message("-- Finished building ${KUBERNETES_UNIT_TEST_COUNT} Kubernetes related
test file(s)...")
diff --git a/extensions/llamacpp/tests/CMakeLists.txt
b/extensions/llamacpp/tests/CMakeLists.txt
index a9f267ab0..56f65090c 100644
--- a/extensions/llamacpp/tests/CMakeLists.txt
+++ b/extensions/llamacpp/tests/CMakeLists.txt
@@ -35,6 +35,6 @@ FOREACH(testfile ${LLAMACPP_TESTS})
MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
add_test(NAME ${testfilename} COMMAND ${testfilename} WORKING_DIRECTORY
${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"llamacpp;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "llamacpp")
ENDFOREACH()
message("-- Finished building ${EXTENSIONS_TEST_COUNT} llama.cpp related test
file(s)...")
diff --git a/extensions/lua/tests/CMakeLists.txt
b/extensions/lua/tests/CMakeLists.txt
index e959b0ace..4fd50f9bc 100644
--- a/extensions/lua/tests/CMakeLists.txt
+++ b/extensions/lua/tests/CMakeLists.txt
@@ -31,7 +31,7 @@ FOREACH(testfile ${EXECUTESCRIPT_LUA_TESTS})
createTests("${testfilename}")
MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS "lua;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "lua")
ENDFOREACH()
copyTestResources(${CMAKE_CURRENT_SOURCE_DIR}/test_lua_scripts/
${CMAKE_BINARY_DIR}/bin/resources/test_lua_scripts/)
diff --git a/extensions/mqtt/tests/CMakeLists.txt
b/extensions/mqtt/tests/CMakeLists.txt
index 9ffa1e8b8..32873027d 100644
--- a/extensions/mqtt/tests/CMakeLists.txt
+++ b/extensions/mqtt/tests/CMakeLists.txt
@@ -30,7 +30,7 @@ FOREACH(testfile ${MQTT_TESTS})
target_link_libraries(${testfilename} minifi-mqtt-extensions)
target_link_libraries(${testfilename} minifi-standard-processors)
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"mqtt;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "mqtt")
ENDFOREACH()
list(LENGTH MQTT_TESTS TEST_COUNT)
diff --git a/extensions/opc/tests/CMakeLists.txt
b/extensions/opc/tests/CMakeLists.txt
index 9d7354593..64b1bafa3 100644
--- a/extensions/opc/tests/CMakeLists.txt
+++ b/extensions/opc/tests/CMakeLists.txt
@@ -29,7 +29,7 @@ FOREACH(testfile ${OPC_UA_TESTS})
target_link_libraries(${testfilename} minifi-opc-extensions)
MATH(EXPR OPC_UA_TEST_COUNT "${OPC_UA_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}")
- set_tests_properties("${testfilename}" PROPERTIES LABELS "opc;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "opc")
target_compile_definitions(${testfilename} PRIVATE
"MINIFI_EXTENSION_GROUP_NAME=org.apache.nifi.minifi.test")
ENDFOREACH()
message("-- Finished building ${OPC_UA_TEST_COUNT} OPC UA related test
file(s)...")
diff --git a/extensions/pdh/tests/CMakeLists.txt
b/extensions/pdh/tests/CMakeLists.txt
index c1459f767..4d92d1df3 100644
--- a/extensions/pdh/tests/CMakeLists.txt
+++ b/extensions/pdh/tests/CMakeLists.txt
@@ -32,6 +32,6 @@ FOREACH(testfile ${PDH_TESTS})
target_link_libraries(${testfilename} minifi-standard-processors)
MATH(EXPR PDH_TEST_COUNT "${PDH_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}"
WORKING_DIRECTORY ${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS "phd;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "pdh")
ENDFOREACH()
message("-- Finished building ${PDH_TEST_COUNT} PDH related test file(s)...")
diff --git a/extensions/procfs/tests/CMakeLists.txt
b/extensions/procfs/tests/CMakeLists.txt
index 4c3238cb3..b61fedca0 100644
--- a/extensions/procfs/tests/CMakeLists.txt
+++ b/extensions/procfs/tests/CMakeLists.txt
@@ -33,7 +33,7 @@ FOREACH(testfile ${PROCFS_TESTS})
target_link_libraries(${testfilename} minifi-standard-processors)
MATH(EXPR PROCFS_TEST_COUNT "${PROCFS_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}")
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"procfs;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "procfs")
ENDFOREACH()
file(COPY "${CMAKE_SOURCE_DIR}/extensions/procfs/tests/mockprocfs_t0"
DESTINATION "${CMAKE_BINARY_DIR}/bin/procfs-test/")
file(COPY "${CMAKE_SOURCE_DIR}/extensions/procfs/tests/mockprocfs_t1"
DESTINATION "${CMAKE_BINARY_DIR}/bin/procfs-test/")
diff --git a/extensions/prometheus/tests/CMakeLists.txt
b/extensions/prometheus/tests/CMakeLists.txt
index aefc1f02c..aecf73427 100644
--- a/extensions/prometheus/tests/CMakeLists.txt
+++ b/extensions/prometheus/tests/CMakeLists.txt
@@ -30,6 +30,6 @@ FOREACH(testfile ${PROMETHEUS_TESTS})
target_link_libraries(${testfilename} minifi-prometheus)
MATH(EXPR PROMETHEUS_TEST_COUNT "${PROMETHEUS_TEST_COUNT}+1")
add_test(NAME "${testfilename}" COMMAND "${testfilename}")
- set_tests_properties("${testfilename}" PROPERTIES LABELS
"prometheus;memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "prometheus")
ENDFOREACH()
message("-- Finished building ${PROMETHEUS_TEST_COUNT} Prometheus related test
file(s)...")
diff --git a/extensions/rocksdb-repos/tests/RepoTests.cpp
b/extensions/rocksdb-repos/tests/RepoTests.cpp
index 4041f5771..79b887d18 100644
--- a/extensions/rocksdb-repos/tests/RepoTests.cpp
+++ b/extensions/rocksdb-repos/tests/RepoTests.cpp
@@ -363,6 +363,11 @@ TEST_CASE("Flush deleted flowfiles before shutdown",
"[TestFFR7]") {
explicit TestFlowFileRepository(const std::string& name)
: FlowFileRepository(name,
core::repository::FLOWFILE_REPOSITORY_DIRECTORY,
10min,
core::repository::MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE, 50ms) {}
+ TestFlowFileRepository(const TestFlowFileRepository&) = delete;
+ TestFlowFileRepository(TestFlowFileRepository&&) = delete;
+ TestFlowFileRepository& operator=(const TestFlowFileRepository&) = delete;
+ TestFlowFileRepository& operator=(TestFlowFileRepository&&) = delete;
+ ~TestFlowFileRepository() override { stop(); }
void flush() override {
FlowFileRepository::flush();
diff --git a/extensions/stable-api-testing/tests/CMakeLists.txt
b/extensions/stable-api-testing/tests/CMakeLists.txt
index 18e9d759e..ce22f7b76 100644
--- a/extensions/stable-api-testing/tests/CMakeLists.txt
+++ b/extensions/stable-api-testing/tests/CMakeLists.txt
@@ -32,5 +32,5 @@ FOREACH (testfile ${SOURCES})
MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
add_test(NAME ${testfilename} COMMAND ${testfilename} WORKING_DIRECTORY
${TEST_DIR})
- set_tests_properties("${testfilename}" PROPERTIES LABELS "memchecked")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS
"stable-api-testing")
ENDFOREACH ()
diff --git
a/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
b/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
index e642399b9..529956177 100644
--- a/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
+++ b/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
@@ -44,8 +44,7 @@ class ProcessGroupTestHarness : public IntegrationBase {
void runAssertions() override {
using
org::apache::nifi::minifi::test::utils::verifyLogLinePresenceInPollTime;
-
REQUIRE(verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_),
- "key:test_attribute value:success"));
+ REQUIRE(verifyLogLinePresenceInPollTime(wait_time_, "key:test_attribute
value:success"));
}
};
diff --git
a/extensions/standard-processors/tests/unit/AttributeRollingWindowTests.cpp
b/extensions/standard-processors/tests/unit/AttributeRollingWindowTests.cpp
index 13fb7f93b..c7a682036 100644
--- a/extensions/standard-processors/tests/unit/AttributeRollingWindowTests.cpp
+++ b/extensions/standard-processors/tests/unit/AttributeRollingWindowTests.cpp
@@ -15,8 +15,11 @@
* limitations under the License.
*/
+#include <algorithm>
#include <memory>
+#include <ranges>
#include <string_view>
+#include "fmt/ranges.h"
#include "unit/Catch.h"
#include "AttributeRollingWindow.h"
#include "unit/SingleProcessorTestController.h"
@@ -29,10 +32,12 @@ using AttributeRollingWindow =
processors::AttributeRollingWindow;
bool checkAttributes(const std::map<std::string, std::string>& expected, const
std::map<std::string, std::string>& actual) {
// expected may be incomplete, but if something is specified in expected,
they also need to be in the actual
// set of attributes
- return std::all_of(std::begin(expected), std::end(expected), [&actual](const
auto& kvpair) {
+ return std::ranges::all_of(expected, [&actual](const auto& kvpair) {
const auto& key = kvpair.first;
- const auto& value = kvpair.second;
- return actual.contains(key) && actual.at(key) == value;
+ if (!actual.contains(key)) { return false; }
+ double expected_value = std::stod(kvpair.second);
+ double actual_value = std::stod(actual.at(key));
+ return std::abs(expected_value - actual_value) < 0.000001;
});
}
@@ -48,6 +53,7 @@ TEST_CASE("AttributeRollingWindow properly forwards
properties to RollingWindow
const auto out_flow_files = out.at(rel("success"));
REQUIRE(out_flow_files.size() == 1);
const auto out_attrs = out_flow_files[0]->getAttributes();
+ INFO(fmt::format("expected: {}\nactual: {}", expected_out_attributes,
out_attrs));
REQUIRE(checkAttributes(expected_out_attributes, out_attrs));
};
trigger_with_value_and_check_attributes("1", {
diff --git a/extensions/standard-processors/tests/unit/GetTCPTests.cpp
b/extensions/standard-processors/tests/unit/GetTCPTests.cpp
index a071c9148..bf4080e09 100644
--- a/extensions/standard-processors/tests/unit/GetTCPTests.cpp
+++ b/extensions/standard-processors/tests/unit/GetTCPTests.cpp
@@ -25,6 +25,7 @@
#include "controllers/SSLContextService.h"
#include "range/v3/algorithm/contains.hpp"
#include "minifi-cpp/utils/gsl.h"
+#include "fmt/chrono.h"
using GetTCP = org::apache::nifi::minifi::processors::GetTCP;
@@ -155,9 +156,9 @@ TEST_CASE("GetTCP test with delimiter", "[GetTCP]") {
SingleProcessorTestController
controller{minifi::test::utils::make_processor<GetTCP>("GetTCP")};
const auto get_tcp = controller.getProcessor();
LogTestController::getInstance().setTrace<GetTCP>();
+ REQUIRE(get_tcp->setProperty(GetTCP::Timeout.name, fmt::format("{}",
minifi::test::utils::timeout(1s))));
REQUIRE(get_tcp->setProperty(GetTCP::MaxBatchSize.name, "2"));
-
TcpTestServer tcp_test_server;
SECTION("No SSL") {}
@@ -186,6 +187,7 @@ TEST_CASE("GetTCP test with too large message", "[GetTCP]")
{
SingleProcessorTestController
controller{minifi::test::utils::make_processor<GetTCP>("GetTCP")};
const auto get_tcp = controller.getProcessor();
LogTestController::getInstance().setTrace<GetTCP>();
+ REQUIRE(get_tcp->setProperty(GetTCP::Timeout.name, fmt::format("{}",
minifi::test::utils::timeout(1s))));
REQUIRE(get_tcp->setProperty(GetTCP::MaxBatchSize.name, "2"));
REQUIRE(get_tcp->setProperty(GetTCP::MaxMessageSize.name, "10"));
REQUIRE(get_tcp->setProperty(GetTCP::MessageDelimiter.name, "\r"));
@@ -227,6 +229,7 @@ TEST_CASE("GetTCP test multiple endpoints", "[GetTCP]") {
SingleProcessorTestController
controller{minifi::test::utils::make_processor<GetTCP>("GetTCP")};
const auto get_tcp = controller.getProcessor();
LogTestController::getInstance().setTrace<GetTCP>();
+ REQUIRE(get_tcp->setProperty(GetTCP::Timeout.name, fmt::format("{}",
minifi::test::utils::timeout(1s))));
REQUIRE(get_tcp->setProperty(GetTCP::MaxBatchSize.name, "2"));
TcpTestServer server_1;
@@ -271,6 +274,7 @@ TEST_CASE("GetTCP max queue and max batch size test",
"[GetTCP]") {
SingleProcessorTestController
controller{minifi::test::utils::make_processor<GetTCP>("GetTCP")};
const auto get_tcp = controller.getProcessor();
LogTestController::getInstance().setTrace<GetTCP>();
+ REQUIRE(get_tcp->setProperty(GetTCP::Timeout.name, fmt::format("{}",
minifi::test::utils::timeout(1s))));
REQUIRE(get_tcp->setProperty(GetTCP::MaxBatchSize.name, "10"));
REQUIRE(get_tcp->setProperty(GetTCP::MaxQueueSize.name, "50"));
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index 4ce6c206e..4589f1ac9 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -809,6 +809,6 @@ TEST_CASE("Test getProcessorType", "[getProcessorType]") {
TEST_CASE("IsYield and getYieldTime is consistent") {
auto processor =
minifi::test::utils::make_processor<TestProcessorNoContent>("test_processor");
- processor->yield(1ms);
+ processor->yield(minifi::test::utils::timeout(1ms));
REQUIRE(processor->isYield() == (processor->getYieldTime() != 0ms));
}
diff --git a/extensions/standard-processors/tests/unit/PutTCPTests.cpp
b/extensions/standard-processors/tests/unit/PutTCPTests.cpp
index 62b8c4e51..fdbf50bea 100644
--- a/extensions/standard-processors/tests/unit/PutTCPTests.cpp
+++ b/extensions/standard-processors/tests/unit/PutTCPTests.cpp
@@ -29,6 +29,8 @@
#include "utils/net/AsioCoro.h"
#include "utils/expected.h"
#include "unit/TestUtils.h"
+#include "asio/experimental/awaitable_operators.hpp"
+#include "fmt/chrono.h"
using namespace std::literals::chrono_literals;
using
org::apache::nifi::minifi::test::utils::verifyLogLineVariantPresenceInPollTime;
@@ -105,7 +107,7 @@ class PutTCPTestFixture {
LogTestController::getInstance().setInfo<core::ProcessSession>();
LogTestController::getInstance().setTrace<utils::net::Server>();
REQUIRE(put_tcp_->setProperty(PutTCP::Hostname.name,
"${literal('localhost')}"));
- REQUIRE(put_tcp_->setProperty(PutTCP::Timeout.name, "200 ms"));
+ REQUIRE(put_tcp_->setProperty(PutTCP::Timeout.name, fmt::format("{}",
minifi::test::utils::timeout(200ms))));
REQUIRE(put_tcp_->setProperty(PutTCP::OutgoingMessageDelimiter.name,
"\n"));
}
@@ -143,7 +145,7 @@ class PutTCPTestFixture {
cancellable_tcp_server->cancelEverything();
}
}
- std::this_thread::sleep_for(200ms);
+ std::this_thread::sleep_for(minifi::test::utils::timeout(200ms));
}
auto trigger(std::string_view message, std::unordered_map<std::string,
std::string> input_flow_file_attributes = {}) {
@@ -155,7 +157,7 @@ class PutTCPTestFixture {
}
std::optional<utils::net::Message>
tryDequeueReceivedMessage(std::optional<uint16_t> port = std::nullopt) {
- auto timeout = 200ms;
+ auto timeout = minifi::test::utils::timeout(200ms);
auto interval = 10ms;
auto start_time = std::chrono::system_clock::now();
diff --git a/libminifi/test/flow-tests/FlowControllerTests.cpp
b/libminifi/test/flow-tests/FlowControllerTests.cpp
index 9e0bb90d8..2c0a7d495 100644
--- a/libminifi/test/flow-tests/FlowControllerTests.cpp
+++ b/libminifi/test/flow-tests/FlowControllerTests.cpp
@@ -89,17 +89,6 @@ Controller Services:
- value: flowcontrollertests_state.txt
)";
-template<typename Fn>
-bool verifyWithBusyWait(std::chrono::milliseconds timeout, const Fn& fn) {
- auto start = std::chrono::steady_clock::now();
- while (std::chrono::steady_clock::now() - start < timeout) {
- if (fn()) {
- return true;
- }
- }
- return false;
-}
-
TEST_CASE("Flow shutdown drains connections", "[TestFlow1]") {
TestControllerWithFlow testController(yamlConfig);
auto controller = testController.controller_;
@@ -156,7 +145,7 @@ TEST_CASE("Flow shutdown waits for a while", "[TestFlow2]")
{
// wait for the source processor to enqueue its flowFiles
auto flowFilesEnqueued = [&] { return root->getTotalFlowFileCount() >= 3; };
- REQUIRE(verifyWithBusyWait(std::chrono::milliseconds{500},
flowFilesEnqueued));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(500ms,
flowFilesEnqueued));
REQUIRE(sourceProc.get().trigger_count.load() >= 1);
@@ -193,7 +182,7 @@ TEST_CASE("Flow stopped after grace period", "[TestFlow3]")
{
// wait for the source processor to enqueue its flowFiles
auto flowFilesEnqueued = [&] { return root->getTotalFlowFileCount() >= 3; };
- REQUIRE(verifyWithBusyWait(std::chrono::milliseconds{500},
flowFilesEnqueued));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(500ms,
flowFilesEnqueued));
REQUIRE(sourceProc.get().trigger_count.load() >= 1);
@@ -232,7 +221,7 @@ TEST_CASE("Extend the waiting period during shutdown",
"[TestFlow4]") {
// wait for the source processor to enqueue its flowFiles
auto flowFilesEnqueued = [&] { return root->getTotalFlowFileCount() >= 3; };
- REQUIRE(verifyWithBusyWait(std::chrono::milliseconds{500},
flowFilesEnqueued));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(500ms,
flowFilesEnqueued));
REQUIRE(sourceProc.get().trigger_count.load() >= 1);
diff --git a/libminifi/test/integration/CMakeLists.txt
b/libminifi/test/integration/CMakeLists.txt
index 2e0ff7b7d..23ddf1836 100644
--- a/libminifi/test/integration/CMakeLists.txt
+++ b/libminifi/test/integration/CMakeLists.txt
@@ -33,6 +33,7 @@ FOREACH(testfile ${INTEGRATION_TESTS})
target_compile_definitions(${testfilename} PRIVATE
"MINIFI_EXTENSION_GROUP_NAME=org.apache.nifi.minifi.test")
if(NOT ${testfilename} IN_LIST TESTS_WITH_LIBARCHIVE_DEPENDENCY)
add_test(NAME "${testfilename}" COMMAND "${testfilename}")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "integration")
endif()
MATH(EXPR INT_TEST_COUNT "${INT_TEST_COUNT}+1")
ENDFOREACH()
@@ -40,6 +41,7 @@ ENDFOREACH()
FOREACH(testfilename ${TESTS_WITH_LIBARCHIVE_DEPENDENCY})
if (ENABLE_LIBARCHIVE)
add_test(NAME "${testfilename}" COMMAND "${testfilename}")
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "integration")
endif()
ENDFOREACH()
diff --git a/libminifi/test/libtest/integration/HTTPIntegrationBase.h
b/libminifi/test/libtest/integration/HTTPIntegrationBase.h
index f15e39205..461adcbe5 100644
--- a/libminifi/test/libtest/integration/HTTPIntegrationBase.h
+++ b/libminifi/test/libtest/integration/HTTPIntegrationBase.h
@@ -40,7 +40,7 @@ int ssl_enable(void* /*ssl_context*/, void* /*user_data*/);
class HTTPIntegrationBase : public IntegrationBase {
public:
explicit HTTPIntegrationBase(const std::optional<std::filesystem::path>&
test_file_location = {}, const std::optional<std::filesystem::path>& home_path
= {},
- std::chrono::milliseconds waitTime =
std::chrono::milliseconds(DEFAULT_WAITTIME_MSECS))
+ std::chrono::milliseconds waitTime = DefaultWaitTime)
: IntegrationBase(test_file_location, home_path, waitTime),
server(nullptr) {
}
@@ -103,7 +103,7 @@ class VerifyC2Describe : public VerifyC2Base {
}
void runAssertions() override {
-
REQUIRE(utils::verifyEventHappenedInPollTime(std::chrono::milliseconds(wait_time_),
[&] { return verified_.load(); }));
+ REQUIRE(utils::verifyEventHappenedInPollTime(wait_time_, [&] { return
verified_.load(); }));
}
protected:
@@ -113,7 +113,7 @@ class VerifyC2Describe : public VerifyC2Base {
class VerifyC2Update : public HTTPIntegrationBase {
public:
explicit VerifyC2Update(const std::filesystem::path& test_file_location,
const std::optional<std::filesystem::path>& home_path = {},
- std::chrono::milliseconds waitTime =
std::chrono::milliseconds(DEFAULT_WAITTIME_MSECS))
+ std::chrono::milliseconds waitTime = DefaultWaitTime)
: HTTPIntegrationBase(test_file_location, home_path, waitTime) {
}
@@ -136,7 +136,7 @@ class VerifyC2Update : public HTTPIntegrationBase {
}
void runAssertions() override {
-
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::seconds(10),
"Starting to reload Flow Controller with flow control name MiNiFi Flow,
version"));
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(10s,
"Starting to reload Flow Controller with flow control name MiNiFi Flow,
version"));
}
};
@@ -167,7 +167,7 @@ class VerifyFlowFetched : public HTTPIntegrationBase {
}
void runAssertions() override {
-
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::seconds(10),
"Successfully fetched valid flow configuration"));
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(10s,
"Successfully fetched valid flow configuration"));
}
};
@@ -184,7 +184,7 @@ class VerifyC2FailedUpdate : public VerifyC2Update {
}
void runAssertions() override {
-
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::seconds(10),
"Invalid configuration payload", "update failed"));
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(10s, "Invalid
configuration payload", "update failed"));
}
void cleanup() override {
diff --git a/libminifi/test/libtest/integration/IntegrationBase.cpp
b/libminifi/test/libtest/integration/IntegrationBase.cpp
index 8a767546a..753acda82 100644
--- a/libminifi/test/libtest/integration/IntegrationBase.cpp
+++ b/libminifi/test/libtest/integration/IntegrationBase.cpp
@@ -158,6 +158,7 @@ void IntegrationBase::run() {
flowController_->stop();
}
+ flowController_.reset(); // stop the C2Agent before cleanup() resets the
logger
cleanup();
}
diff --git a/libminifi/test/libtest/integration/IntegrationBase.h
b/libminifi/test/libtest/integration/IntegrationBase.h
index 9f406d57a..1b68a2317 100644
--- a/libminifi/test/libtest/integration/IntegrationBase.h
+++ b/libminifi/test/libtest/integration/IntegrationBase.h
@@ -17,8 +17,6 @@
*/
#pragma once
-#define DEFAULT_WAITTIME_MSECS 10000
-
#include <memory>
#include <optional>
#include <string>
@@ -32,6 +30,7 @@
#include "utils/file/FileUtils.h"
#include "core/BulletinStore.h"
#include "unit/TestBase.h"
+#include "unit/TestUtils.h"
namespace minifi = org::apache::nifi::minifi;
namespace core = minifi::core;
@@ -46,8 +45,10 @@ struct FlowConfigPath {
class IntegrationBase {
public:
+ static constexpr auto DefaultWaitTime = 10s;
+
explicit IntegrationBase(const std::optional<std::filesystem::path>&
test_file_location = {}, const std::optional<std::filesystem::path>& home_path
= {},
- std::chrono::milliseconds waitTime =
std::chrono::milliseconds(DEFAULT_WAITTIME_MSECS));
+ std::chrono::milliseconds waitTime = DefaultWaitTime);
IntegrationBase(const IntegrationBase&) = delete;
IntegrationBase(IntegrationBase&& other) noexcept
: logger_properties_{std::move(other.logger_properties_)},
diff --git a/libminifi/test/libtest/unit/SingleProcessorTestController.cpp
b/libminifi/test/libtest/unit/SingleProcessorTestController.cpp
index fe2a04fc7..c7624e667 100644
--- a/libminifi/test/libtest/unit/SingleProcessorTestController.cpp
+++ b/libminifi/test/libtest/unit/SingleProcessorTestController.cpp
@@ -22,6 +22,7 @@
#include "FlowFileRecord.h"
#include "range/v3/algorithm/all_of.hpp"
#include "Connection.h"
+#include "TestUtils.h"
namespace org::apache::nifi::minifi::test {
@@ -63,7 +64,7 @@ bool SingleProcessorTestController::triggerUntil(const
std::unordered_map<core::
const std::chrono::milliseconds max_duration,
const std::chrono::milliseconds wait_time) {
auto start_time = std::chrono::steady_clock::now();
- while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ while (std::chrono::steady_clock::now() < start_time +
utils::timeout(max_duration)) {
for (auto& [relationship, flow_files] : trigger()) {
result[relationship].insert(result[relationship].end(),
flow_files.begin(), flow_files.end());
}
diff --git a/libminifi/test/libtest/unit/TestUtils.cpp
b/libminifi/test/libtest/unit/TestUtils.cpp
index d417089c0..0e1a283f8 100644
--- a/libminifi/test/libtest/unit/TestUtils.cpp
+++ b/libminifi/test/libtest/unit/TestUtils.cpp
@@ -25,6 +25,7 @@
#endif
#include "minifi-cpp/utils/gsl.h"
+#include "utils/Environment.h"
#ifdef WIN32
namespace {
@@ -63,6 +64,14 @@ void setAclOnFileOrDirectory(std::string file_name, DWORD
perms, ACCESS_MODE per
namespace org::apache::nifi::minifi::test::utils {
+bool runningUnderValgrind() {
+ static const bool result = [] {
+ const auto ld_preload =
minifi::utils::Environment::getEnvironmentVariable("LD_PRELOAD");
+ return ld_preload && (ld_preload->contains("vgpreload") ||
ld_preload->contains("valgrind"));
+ }();
+ return result;
+}
+
std::filesystem::path putFileToDir(const std::filesystem::path& dir_path,
const std::filesystem::path& file_name, const std::string& content) {
auto file_path = dir_path/file_name;
std::ofstream out_file(file_path, std::ios::binary | std::ios::out);
@@ -162,7 +171,7 @@ bool countLogOccurrencesUntil(const std::string& pattern,
const std::chrono::milliseconds max_duration,
const std::chrono::milliseconds wait_time) {
auto start_time = std::chrono::steady_clock::now();
- while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ while (std::chrono::steady_clock::now() < start_time +
timeout(max_duration)) {
if (LogTestController::getInstance().countOccurrences(pattern) ==
occurrences)
return true;
std::this_thread::sleep_for(wait_time);
diff --git a/libminifi/test/libtest/unit/TestUtils.h
b/libminifi/test/libtest/unit/TestUtils.h
index 654fdf4fc..bbbe62f91 100644
--- a/libminifi/test/libtest/unit/TestUtils.h
+++ b/libminifi/test/libtest/unit/TestUtils.h
@@ -95,12 +95,21 @@ class ManualClock : public
minifi::utils::timeutils::SteadyClock {
std::chrono::milliseconds time_{0};
};
+inline constexpr int VALGRIND_TIMEOUT_MULTIPLIER = 50;
+
+bool runningUnderValgrind();
+
+template <class Rep, class Period>
+std::chrono::duration<Rep, Period> timeout(std::chrono::duration<Rep, Period>
duration) {
+ return runningUnderValgrind() ? duration * VALGRIND_TIMEOUT_MULTIPLIER :
duration;
+}
+
template <class Rep, class Period, typename Fun>
bool verifyEventHappenedInPollTime(
const std::chrono::duration<Rep, Period>& wait_duration,
Fun&& check,
- std::chrono::microseconds check_interval = std::chrono::milliseconds(100))
{
- std::chrono::steady_clock::time_point wait_end =
std::chrono::steady_clock::now() + wait_duration;
+ std::chrono::microseconds check_interval = 100ms) {
+ std::chrono::steady_clock::time_point wait_end =
std::chrono::steady_clock::now() + timeout(wait_duration);
do {
if (std::forward<Fun>(check)()) {
return true;
diff --git a/libminifi/test/unit/FlowFileQueueTests.cpp
b/libminifi/test/unit/FlowFileQueueTests.cpp
index 73d1afe17..999cddbfa 100644
--- a/libminifi/test/unit/FlowFileQueueTests.cpp
+++ b/libminifi/test/unit/FlowFileQueueTests.cpp
@@ -49,7 +49,7 @@ TEST_CASE("A flow file can be moved into the FlowFileQueue",
"[FlowFileQueue][po
utils::FlowFileQueue queue;
auto penalized_flow_file = std::make_shared<core::FlowFileImpl>();
- penalized_flow_file->penalize(std::chrono::milliseconds{100});
+ penalized_flow_file->penalize(minifi::test::utils::timeout(100ms));
queue.push(std::move(penalized_flow_file));
queue.push(std::make_shared<core::FlowFileImpl>());
@@ -81,7 +81,7 @@ TEST_CASE("Cannot add flow files in the past preempting
others", "[FlowFileQueue
const auto flow_file_1 = std::make_shared<core::FlowFileImpl>();
queue.push(flow_file_1);
const auto flow_file_2 = std::make_shared<core::FlowFileImpl>();
- flow_file_2->penalize(std::chrono::seconds{-10});
+ flow_file_2->penalize(-10s);
queue.push(flow_file_2);
REQUIRE(queue.isWorkAvailable());
@@ -107,33 +107,33 @@ class PenaltyHasExpired {
TEST_CASE("Penalized flow files are popped from the FlowFileQueue in the order
their penalties expire", "[FlowFileQueue][pop]") {
utils::FlowFileQueue queue;
const auto flow_file_1 = std::make_shared<core::FlowFileImpl>();
- flow_file_1->penalize(std::chrono::milliseconds{70});
+ flow_file_1->penalize(minifi::test::utils::timeout(70ms));
queue.push(flow_file_1);
const auto flow_file_2 = std::make_shared<core::FlowFileImpl>();
- flow_file_2->penalize(std::chrono::milliseconds{50});
+ flow_file_2->penalize(minifi::test::utils::timeout(50ms));
queue.push(flow_file_2);
const auto flow_file_3 = std::make_shared<core::FlowFileImpl>();
- flow_file_3->penalize(std::chrono::milliseconds{80});
+ flow_file_3->penalize(minifi::test::utils::timeout(80ms));
queue.push(flow_file_3);
const auto flow_file_4 = std::make_shared<core::FlowFileImpl>();
- flow_file_4->penalize(std::chrono::milliseconds{60});
+ flow_file_4->penalize(minifi::test::utils::timeout(60ms));
queue.push(flow_file_4);
REQUIRE_FALSE(queue.isWorkAvailable());
-
REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1},
PenaltyHasExpired{flow_file_2}, std::chrono::milliseconds{10}));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(1s,
PenaltyHasExpired{flow_file_2}, minifi::test::utils::timeout(10ms)));
REQUIRE(queue.isWorkAvailable());
REQUIRE(queue.pop() == flow_file_2);
-
REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1},
PenaltyHasExpired{flow_file_4}, std::chrono::milliseconds{10}));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(1s,
PenaltyHasExpired{flow_file_4}, minifi::test::utils::timeout(10ms)));
REQUIRE(queue.isWorkAvailable());
REQUIRE(queue.pop() == flow_file_4);
-
REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1},
PenaltyHasExpired{flow_file_1}, std::chrono::milliseconds{10}));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(1s,
PenaltyHasExpired{flow_file_1}, minifi::test::utils::timeout(10ms)));
REQUIRE(queue.isWorkAvailable());
REQUIRE(queue.pop() == flow_file_1);
-
REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1},
PenaltyHasExpired{flow_file_3}, std::chrono::milliseconds{10}));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(1s,
PenaltyHasExpired{flow_file_3}, minifi::test::utils::timeout(10ms)));
REQUIRE(queue.isWorkAvailable());
REQUIRE(queue.pop() == flow_file_3);
@@ -143,7 +143,7 @@ TEST_CASE("Penalized flow files are popped from the
FlowFileQueue in the order t
TEST_CASE("If a penalized then a non-penalized flow file is added to the
FlowFileQueue, pop() returns the correct one", "[FlowFileQueue][pop]") {
utils::FlowFileQueue queue;
const auto penalized_flow_file = std::make_shared<core::FlowFileImpl>();
- penalized_flow_file->penalize(std::chrono::milliseconds{10});
+ penalized_flow_file->penalize(minifi::test::utils::timeout(10ms));
queue.push(penalized_flow_file);
const auto flow_file = std::make_shared<core::FlowFileImpl>();
queue.push(flow_file);
@@ -155,7 +155,7 @@ TEST_CASE("If a penalized then a non-penalized flow file is
added to the FlowFil
}
SECTION("Wait until the penalty expires, then pop") {
-
REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(std::chrono::seconds{1},
PenaltyHasExpired{penalized_flow_file}, std::chrono::milliseconds{10}));
+ REQUIRE(minifi::test::utils::verifyEventHappenedInPollTime(1s,
PenaltyHasExpired{penalized_flow_file}, minifi::test::utils::timeout(10ms)));
REQUIRE(queue.isWorkAvailable());
REQUIRE(queue.pop() == flow_file);
@@ -168,7 +168,7 @@ TEST_CASE("If a penalized then a non-penalized flow file is
added to the FlowFil
TEST_CASE("Force pop on FlowFileQueue returns the flow files, whether
penalized or not", "[FlowFileQueue][forcePop]") {
utils::FlowFileQueue queue;
const auto penalized_flow_file = std::make_shared<core::FlowFileImpl>();
- penalized_flow_file->penalize(std::chrono::milliseconds{10});
+ penalized_flow_file->penalize(minifi::test::utils::timeout(10ms));
queue.push(penalized_flow_file);
const auto flow_file = std::make_shared<core::FlowFileImpl>();
queue.push(flow_file);
diff --git a/libminifi/test/unit/MemoryUsageTest.cpp
b/libminifi/test/unit/MemoryUsageTest.cpp
index b818e90fa..f8bf834a7 100644
--- a/libminifi/test/unit/MemoryUsageTest.cpp
+++ b/libminifi/test/unit/MemoryUsageTest.cpp
@@ -26,9 +26,14 @@
#include "minifi-cpp/utils/gsl.h"
#include "utils/OsUtils.h"
#include "unit/TestBase.h"
+#include "unit/TestUtils.h"
#include "unit/Catch.h"
TEST_CASE("Test Physical memory usage", "[testphysicalmemoryusage]") {
+ if (minifi::test::utils::runningUnderValgrind()) {
+ SKIP("Memory usage is much higher when running under Valgrind");
+ }
+
constexpr bool cout_enabled = true;
std::vector<uint8_t> large_vector(30'000'000);
diff --git a/libminifi/test/unit/MinifiConcurrentQueueTests.cpp
b/libminifi/test/unit/MinifiConcurrentQueueTests.cpp
index dece31b50..469641b4e 100644
--- a/libminifi/test/unit/MinifiConcurrentQueueTests.cpp
+++ b/libminifi/test/unit/MinifiConcurrentQueueTests.cpp
@@ -136,7 +136,7 @@ namespace MinifiConcurrentQueueTestProducersConsumers {
constexpr std::size_t max_read_attempts = 300;
for (std::size_t attempt_num = 0; results.size() < 3 && attempt_num <
max_read_attempts; ++attempt_num) {
std::string s;
- if (queue.dequeueWaitFor(s, std::chrono::milliseconds(1))) {
+ if (queue.dequeueWaitFor(s, minifi::test::utils::timeout(1ms))) {
results.push_back(s);
}
}
@@ -240,11 +240,14 @@ TEST_CASE("TestConcurrentQueue: test simple producer and
consumer", "[ProducerCo
producer =
MinifiConcurrentQueueTestProducersConsumers::getSimpleProducerThread(queue);
consumer =
MinifiConcurrentQueueTestProducersConsumers::getSimpleConsumeConsumerThread(queue,
results);
}
- /* In this testcase the consumer thread puts back all items to the queue
to consume again
- * Even in this case the ones inserted later by the producer should be
consumed */
- SECTION("with readd") {
- producer =
MinifiConcurrentQueueTestProducersConsumers::getSimpleProducerThread(queue);
- consumer =
MinifiConcurrentQueueTestProducersConsumers::getSpinningReaddingDequeueConsumerThread(queue,
results);
+ // In this testcase the consumer thread puts back all items to the queue
to consume again
+ // Even in this case the ones inserted later by the producer should be
consumed
+ // The spinning-readding consumer doesn't work well under Valgrind, so we
skip it.
+ if (!minifi::test::utils::runningUnderValgrind()) {
+ SECTION("with readd") {
+ producer =
MinifiConcurrentQueueTestProducersConsumers::getSimpleProducerThread(queue);
+ consumer =
MinifiConcurrentQueueTestProducersConsumers::getSpinningReaddingDequeueConsumerThread(queue,
results);
+ }
}
producer.join();
consumer.join();
@@ -302,6 +305,10 @@ TEST_CASE("TestConcurrentQueue: test untimed waiting
consumers", "[ProducerConsu
}
TEST_CASE("TestConcurrentQueue: test the readding dequeue consumer",
"[ProducerConsumer]") {
+ if (minifi::test::utils::runningUnderValgrind()) {
+ SKIP("The spinning-readding consumer doesn't work well under Valgrind's
serialized scheduler");
+ }
+
utils::ConditionConcurrentQueue<std::string> queue(true);
std::vector<std::string> results;
@@ -342,6 +349,10 @@ TEST_CASE("TestConcurrentQueue: test waiting consumers
with blocked producer", "
}
TEST_CASE("TestConcurrentQueues::highLoad", "[TestConcurrentQueuesHighLoad]") {
+ if (minifi::test::utils::runningUnderValgrind()) {
+ SKIP("High-load busy-spin test is too slow under Valgrind's serialized
scheduler");
+ }
+
std::random_device dev;
std::mt19937 rng(dev());
std::uniform_int_distribution<std::mt19937::result_type> dist(1,
std::numeric_limits<int>::max());
