diff --git a/src/integrations/kafka/exceptions.hpp b/src/integrations/kafka/exceptions.hpp index 3c25d3a26..6b8ab980b 100644 --- a/src/integrations/kafka/exceptions.hpp +++ b/src/integrations/kafka/exceptions.hpp @@ -26,12 +26,13 @@ class ConsumerFailedToInitializeException : public KafkaStreamException { : KafkaStreamException("Failed to initialize Kafka consumer {} : {}", consumer_name, error) {} }; -class SettingCustomConfigFailed : public KafkaStreamException { +class SettingCustomConfigFailed : public ConsumerFailedToInitializeException { public: SettingCustomConfigFailed(const std::string &consumer_name, const std::string &error, const std::string &key, const std::string &value) - : KafkaStreamException(R"(Failed to set custom config ("{}": "{}") for Kafka consumer {} : {})", key, value, - consumer_name, error) {} + : ConsumerFailedToInitializeException( + consumer_name, + fmt::format(R"(failed to set custom config ("{}": "{}"), because of error {})", key, value, error)) {} }; class ConsumerRunningException : public KafkaStreamException { diff --git a/src/query/frontend/ast/cypher_main_visitor.cpp b/src/query/frontend/ast/cypher_main_visitor.cpp index 90d3c2af1..cf4cbbc9c 100644 --- a/src/query/frontend/ast/cypher_main_visitor.cpp +++ b/src/query/frontend/ast/cypher_main_visitor.cpp @@ -596,6 +596,8 @@ antlrcpp::Any CypherMainVisitor::visitConfigKeyValuePair(MemgraphCypher::ConfigK antlrcpp::Any CypherMainVisitor::visitConfigMap(MemgraphCypher::ConfigMapContext *ctx) { std::unordered_map map; for (auto *key_value_pair : ctx->configKeyValuePair()) { + // If the queries are cached, then only the stripped query is parsed, so the actual keys are cannot be determined + // here. That means duplicates cannot be checked. map.insert(key_value_pair->accept(this).as>()); } return map; diff --git a/src/query/stream/sources.cpp b/src/query/stream/sources.cpp index 63deb4f31..b46fbabc9 100644 --- a/src/query/stream/sources.cpp +++ b/src/query/stream/sources.cpp @@ -33,13 +33,6 @@ KafkaStream::KafkaStream(std::string stream_name, StreamInfo stream_info, KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const { const auto &info = consumer_->Info(); - using CredentialsType = decltype(StreamInfo::credentials); - CredentialsType reducted_credentials; - std::transform(info.private_configs.begin(), info.private_configs.end(), - std::inserter(reducted_credentials, reducted_credentials.end()), - [](const auto &pair) -> CredentialsType::value_type { - return {pair.first, integrations::kReducted}; - }); return {{.batch_interval = info.batch_interval, .batch_size = info.batch_size, .transformation_name = std::move(transformation_name)}, @@ -47,7 +40,7 @@ KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const .consumer_group = info.consumer_group, .bootstrap_servers = info.bootstrap_servers, .configs = info.public_configs, - .credentials = std::move(reducted_credentials)}; + .credentials = info.private_configs}; } void KafkaStream::Start() { consumer_->Start(); } diff --git a/src/query/stream/streams.cpp b/src/query/stream/streams.cpp index deb5ef33a..e772b6d9b 100644 --- a/src/query/stream/streams.cpp +++ b/src/query/stream/streams.cpp @@ -18,6 +18,7 @@ #include #include +#include "integrations/constants.hpp" #include "mg_procedure.h" #include "query/db_accessor.hpp" #include "query/discard_value_stream.hpp" @@ -316,7 +317,15 @@ void Streams::RegisterKafkaProcedures() { return; } - auto credentials_value = convert_config_map(info.credentials); + using CredentialsType = decltype(KafkaStream::StreamInfo::credentials); + CredentialsType reducted_credentials; + std::transform(info.credentials.begin(), info.credentials.end(), + std::inserter(reducted_credentials, reducted_credentials.end()), + [](const auto &pair) -> CredentialsType::value_type { + return {pair.first, integrations::kReducted}; + }); + + auto credentials_value = convert_config_map(reducted_credentials); if (credentials_value == nullptr) { return; } diff --git a/src/query/stream/streams.hpp b/src/query/stream/streams.hpp index b927c4c07..a1afd1a47 100644 --- a/src/query/stream/streams.hpp +++ b/src/query/stream/streams.hpp @@ -1,4 +1,4 @@ -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source @@ -31,6 +31,7 @@ #include "utils/rw_lock.hpp" #include "utils/synchronized.hpp" +class StreamsTest; namespace query { struct InterpreterContext; @@ -73,6 +74,8 @@ using TransformationResult = std::vector>; /// /// This class is responsible for all query supported actions to happen. class Streams final { + friend StreamsTest; + public: /// Initializes the streams. /// diff --git a/tests/unit/cypher_main_visitor.cpp b/tests/unit/cypher_main_visitor.cpp index 8eb548243..73ab6d916 100644 --- a/tests/unit/cypher_main_visitor.cpp +++ b/tests/unit/cypher_main_visitor.cpp @@ -1,4 +1,4 @@ -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source @@ -9,20 +9,10 @@ // by the Apache License, Version 2.0, included in the file // licenses/APL.txt. -// Copyright 2021 Memgraph Ltd. -// -// Use of this software is governed by the Business Source License -// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source -// License, and you may not use this file except in compliance with the Business Source License. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0, included in the file -// licenses/APL.txt. -// #include #include #include +#include #include #include #include @@ -38,6 +28,7 @@ #include ////////////////////////////////////////////////////// #include +#include #include #include @@ -87,23 +78,37 @@ class Base { } } + TypedValue GetLiteral(Expression *expression, const bool use_parameter_lookup, + const std::optional &token_position = std::nullopt) const { + if (use_parameter_lookup) { + auto *param_lookup = dynamic_cast(expression); + if (param_lookup == nullptr) { + ADD_FAILURE(); + return {}; + } + if (token_position) { + EXPECT_EQ(param_lookup->token_position_, *token_position); + } + return TypedValue(parameters_.AtTokenPosition(param_lookup->token_position_)); + } + + auto *literal = dynamic_cast(expression); + if (literal == nullptr) { + ADD_FAILURE(); + return {}; + } + if (token_position) { + EXPECT_EQ(literal->token_position_, *token_position); + } + return TypedValue(literal->value_); + } + template void CheckLiteral(Expression *expression, const TValue &expected, const std::optional &token_position = std::nullopt) const { - TypedValue value; - // NOLINTNEXTLINE(performance-unnecessary-copy-initialization) TypedValue expected_tv(expected); - if (!expected_tv.IsNull() && context_.is_query_cached) { - auto *param_lookup = dynamic_cast(expression); - ASSERT_TRUE(param_lookup); - if (token_position) EXPECT_EQ(param_lookup->token_position_, *token_position); - value = TypedValue(parameters_.AtTokenPosition(param_lookup->token_position_)); - } else { - auto *literal = dynamic_cast(expression); - ASSERT_TRUE(literal); - if (token_position) ASSERT_EQ(literal->token_position_, *token_position); - value = TypedValue(literal->value_); - } + const auto use_parameter_lookup = !expected_tv.IsNull() && context_.is_query_cached; + TypedValue value = GetLiteral(expression, use_parameter_lookup, token_position); EXPECT_TRUE(TypedValue::BoolEqual{}(value, expected_tv)); } }; @@ -3580,6 +3585,8 @@ void ValidateMostlyEmptyStreamQuery(Base &ast_generator, const std::string &quer EXPECT_EQ(parsed_query->bootstrap_servers_, nullptr); EXPECT_NO_FATAL_FAILURE(CheckOptionalExpression(ast_generator, parsed_query->batch_limit_, batch_limit)); EXPECT_NO_FATAL_FAILURE(CheckOptionalExpression(ast_generator, parsed_query->timeout_, timeout)); + EXPECT_TRUE(parsed_query->configs_.empty()); + EXPECT_TRUE(parsed_query->credentials_.empty()); } TEST_P(CypherMainVisitorTest, DropStream) { @@ -3661,7 +3668,9 @@ void ValidateCreateKafkaStreamQuery(Base &ast_generator, const std::string &quer const std::string_view transform_name, const std::string_view consumer_group, const std::optional &batch_interval, const std::optional &batch_size, - const std::string_view bootstrap_servers = "") { + const std::string_view bootstrap_servers, + const std::unordered_map &configs, + const std::unordered_map &credentials) { SCOPED_TRACE(query_string); StreamQuery *parsed_query{nullptr}; ASSERT_NO_THROW(parsed_query = dynamic_cast(ast_generator.ParseQuery(query_string))) << query_string; @@ -3675,14 +3684,31 @@ void ValidateCreateKafkaStreamQuery(Base &ast_generator, const std::string &quer EXPECT_EQ(parsed_query->batch_limit_, nullptr); if (bootstrap_servers.empty()) { EXPECT_EQ(parsed_query->bootstrap_servers_, nullptr); - return; + } else { + EXPECT_NE(parsed_query->bootstrap_servers_, nullptr); } - EXPECT_NE(parsed_query->bootstrap_servers_, nullptr); + + const auto evaluate_config_map = [&ast_generator](const std::unordered_map &config_map) { + std::unordered_map evaluated_config_map; + const auto expr_to_str = [&ast_generator](Expression *expression) { + return std::string{ast_generator.GetLiteral(expression, ast_generator.context_.is_query_cached).ValueString()}; + }; + std::transform(config_map.begin(), config_map.end(), + std::inserter(evaluated_config_map, evaluated_config_map.end()), + [&expr_to_str](const auto expr_pair) { + return std::pair{expr_to_str(expr_pair.first), expr_to_str(expr_pair.second)}; + }); + return evaluated_config_map; + }; + + using testing::UnorderedElementsAreArray; + EXPECT_THAT(evaluate_config_map(parsed_query->configs_), UnorderedElementsAreArray(configs.begin(), configs.end())); + EXPECT_THAT(evaluate_config_map(parsed_query->credentials_), + UnorderedElementsAreArray(credentials.begin(), credentials.end())); } TEST_P(CypherMainVisitorTest, CreateKafkaStream) { auto &ast_generator = *GetParam(); - TestInvalidQuery("CREATE KAFKA STREAM", ast_generator); TestInvalidQuery("CREATE KAFKA STREAM invalid stream name TOPICS topic1 TRANSFORM transform", ast_generator); TestInvalidQuery("CREATE KAFKA STREAM stream TOPICS invalid topic name TRANSFORM transform", ast_generator); @@ -3709,6 +3735,13 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { TestInvalidQuery("CREATE KAFKA STREAM stream TOPICS topic1 TRANSFORM transform BOOTSTRAP_SERVERS localhost:9092", ast_generator); TestInvalidQuery("CREATE KAFKA STREAM stream TOPICS topic1 TRANSFORM transform BOOTSTRAP_SERVERS", ast_generator); + // the keys must be string literals + TestInvalidQuery("CREATE KAFKA STREAM stream TOPICS topic1 TRANSFORM transform CONFIGS { symbolicname : 'string' }", + ast_generator); + TestInvalidQuery( + "CREATE KAFKA STREAM stream TOPICS topic1 TRANSFORM transform CREDENTIALS { symbolicname : 'string' }", + ast_generator); + TestInvalidQuery("CREATE KAFKA STREAM stream TOPICS topic1 TRANSFORM transform CREDENTIALS 2", ast_generator); const std::vector topic_names{"topic1_name.with_dot", "topic1_name.with_multiple.dots", "topic-name.with-multiple.dots-and-dashes"}; @@ -3728,34 +3761,37 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { ValidateCreateKafkaStreamQuery( ast_generator, fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {}", kStreamName, topic_names_as_str, kTransformName), - kStreamName, topic_names, kTransformName, "", std::nullopt, std::nullopt); + kStreamName, topic_names, kTransformName, "", std::nullopt, std::nullopt, {}, {}, {}); ValidateCreateKafkaStreamQuery(ast_generator, fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {} CONSUMER_GROUP {} ", kStreamName, topic_names_as_str, kTransformName, kConsumerGroup), - kStreamName, topic_names, kTransformName, kConsumerGroup, std::nullopt, - std::nullopt); + kStreamName, topic_names, kTransformName, kConsumerGroup, std::nullopt, std::nullopt, + {}, {}, {}); ValidateCreateKafkaStreamQuery(ast_generator, fmt::format("CREATE KAFKA STREAM {} TRANSFORM {} TOPICS {} BATCH_INTERVAL {}", kStreamName, kTransformName, topic_names_as_str, kBatchInterval), - kStreamName, topic_names, kTransformName, "", batch_interval_value, std::nullopt); + kStreamName, topic_names, kTransformName, "", batch_interval_value, std::nullopt, {}, + {}, {}); ValidateCreateKafkaStreamQuery(ast_generator, fmt::format("CREATE KAFKA STREAM {} BATCH_SIZE {} TOPICS {} TRANSFORM {}", kStreamName, kBatchSize, topic_names_as_str, kTransformName), - kStreamName, topic_names, kTransformName, "", std::nullopt, batch_size_value); + kStreamName, topic_names, kTransformName, "", std::nullopt, batch_size_value, {}, {}, + {}); ValidateCreateKafkaStreamQuery(ast_generator, fmt::format("CREATE KAFKA STREAM {} TOPICS '{}' BATCH_SIZE {} TRANSFORM {}", kStreamName, topic_names_as_str, kBatchSize, kTransformName), - kStreamName, topic_names, kTransformName, "", std::nullopt, batch_size_value); + kStreamName, topic_names, kTransformName, "", std::nullopt, batch_size_value, {}, {}, + {}); ValidateCreateKafkaStreamQuery( ast_generator, fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {} CONSUMER_GROUP {} BATCH_INTERVAL {} BATCH_SIZE {}", kStreamName, topic_names_as_str, kTransformName, kConsumerGroup, kBatchInterval, kBatchSize), - kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value); + kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, {}, {}, {}); using namespace std::string_literals; const auto host1 = "localhost:9094"s; ValidateCreateKafkaStreamQuery( @@ -3763,14 +3799,16 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { fmt::format("CREATE KAFKA STREAM {} TOPICS {} CONSUMER_GROUP {} BATCH_SIZE {} BATCH_INTERVAL {} TRANSFORM {} " "BOOTSTRAP_SERVERS '{}'", kStreamName, topic_names_as_str, kConsumerGroup, kBatchSize, kBatchInterval, kTransformName, host1), - kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host1); + kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host1, {}, + {}); ValidateCreateKafkaStreamQuery( ast_generator, fmt::format("CREATE KAFKA STREAM {} CONSUMER_GROUP {} TOPICS {} BATCH_INTERVAL {} TRANSFORM {} BATCH_SIZE {} " "BOOTSTRAP_SERVERS '{}'", kStreamName, kConsumerGroup, topic_names_as_str, kBatchInterval, kTransformName, kBatchSize, host1), - kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host1); + kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host1, {}, + {}); const auto host2 = "localhost:9094,localhost:1994,168.1.1.256:345"s; ValidateCreateKafkaStreamQuery( @@ -3778,7 +3816,8 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { fmt::format("CREATE KAFKA STREAM {} TOPICS {} BOOTSTRAP_SERVERS '{}' CONSUMER_GROUP {} TRANSFORM {} " "BATCH_INTERVAL {} BATCH_SIZE {}", kStreamName, topic_names_as_str, host2, kConsumerGroup, kTransformName, kBatchInterval, kBatchSize), - kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host2); + kStreamName, topic_names, kTransformName, kConsumerGroup, batch_interval_value, batch_size_value, host2, {}, + {}); }; for (const auto &topic_name : topic_names) { @@ -3793,7 +3832,7 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {} CONSUMER_GROUP {}", kStreamName, kTopicName, kTransformName, consumer_group), kStreamName, {kTopicName}, kTransformName, consumer_group, std::nullopt, - std::nullopt); + std::nullopt, {}, {}, {}); }; using namespace std::literals; @@ -3803,6 +3842,44 @@ TEST_P(CypherMainVisitorTest, CreateKafkaStream) { for (const auto consumer_group : consumer_groups) { EXPECT_NO_FATAL_FAILURE(check_consumer_group(consumer_group)); } + + auto check_config_map = [&](const std::unordered_map &config_map) { + const std::string kTopicName{"topic1"}; + + const auto map_as_str = std::invoke([&config_map] { + std::stringstream buffer; + buffer << '{'; + if (!config_map.empty()) { + const auto &first_item = *config_map.begin(); + buffer << fmt::format("'{}': '{}'", first_item.first, first_item.second); + for (auto it = ++config_map.begin(); it != config_map.end(); ++it) { + buffer << fmt::format(", '{}': '{}'", it->first, it->second); + } + } + buffer << '}'; + return std::move(buffer).str(); + }); + + ValidateCreateKafkaStreamQuery(ast_generator, + fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {} CONFIGS {}", kStreamName, + kTopicName, kTransformName, map_as_str), + kStreamName, {kTopicName}, kTransformName, "", std::nullopt, std::nullopt, {}, + config_map, {}); + + ValidateCreateKafkaStreamQuery(ast_generator, + fmt::format("CREATE KAFKA STREAM {} TOPICS {} TRANSFORM {} CREDENTIALS {}", + kStreamName, kTopicName, kTransformName, map_as_str), + kStreamName, {kTopicName}, kTransformName, "", std::nullopt, std::nullopt, {}, {}, + config_map); + }; + + const std::array, 3> config_maps = { + std::unordered_map{}, std::unordered_map{{"key", "value"}}, + std::unordered_map{{"key.with.dot", "value.with.doth"}, + {"key with space", "value with space"}}}; + for (const auto &map_to_test : config_maps) { + EXPECT_NO_FATAL_FAILURE(check_config_map(map_to_test)); + } } void ValidateCreatePulsarStreamQuery(Base &ast_generator, const std::string &query_string, diff --git a/tests/unit/integrations_kafka_consumer.cpp b/tests/unit/integrations_kafka_consumer.cpp index 502317658..d598fefaa 100644 --- a/tests/unit/integrations_kafka_consumer.cpp +++ b/tests/unit/integrations_kafka_consumer.cpp @@ -1,4 +1,4 @@ -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source @@ -59,6 +59,8 @@ struct ConsumerTest : public ::testing::Test { .bootstrap_servers = cluster.Bootstraps(), .batch_interval = kDefaultBatchInterval, .batch_size = kDefaultBatchSize, + .public_configs = {}, + .private_configs = {}, }; }; diff --git a/tests/unit/interpreter.cpp b/tests/unit/interpreter.cpp index bd884f2cf..78ca82497 100644 --- a/tests/unit/interpreter.cpp +++ b/tests/unit/interpreter.cpp @@ -1,15 +1,4 @@ -// Copyright 2021 Memgraph Ltd. -// -// Use of this software is governed by the Business Source License -// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source -// License, and you may not use this file except in compliance with the Business Source License. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0, included in the file -// licenses/APL.txt. - -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source diff --git a/tests/unit/query_streams.cpp b/tests/unit/query_streams.cpp index 018da55c3..9bd606ac1 100644 --- a/tests/unit/query_streams.cpp +++ b/tests/unit/query_streams.cpp @@ -1,4 +1,4 @@ -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source @@ -15,12 +15,15 @@ #include #include + +#include "integrations/constants.hpp" #include "integrations/kafka/exceptions.hpp" #include "kafka_mock.hpp" #include "query/config.hpp" #include "query/interpreter.hpp" #include "query/stream/streams.hpp" #include "storage/v2/storage.hpp" +#include "test_utils.hpp" using Streams = query::stream::Streams; using StreamInfo = query::stream::KafkaStream::StreamInfo; @@ -78,6 +81,17 @@ class StreamsTest : public ::testing::Test { EXPECT_EQ(check_data.is_running, status.is_running); } + void CheckConfigAndCredentials(const StreamCheckData &check_data) { + const auto locked_streams = streams_->streams_.ReadLock(); + const auto &stream = locked_streams->at(check_data.name); + const auto *stream_data = std::get_if>(&stream); + ASSERT_NE(stream_data, nullptr); + const auto stream_info = + stream_data->stream_source->ReadLock()->Info(check_data.info.common_info.transformation_name); + EXPECT_TRUE( + std::equal(check_data.info.configs.begin(), check_data.info.configs.end(), stream_info.configs.begin())); + } + void StartStream(StreamCheckData &check_data) { streams_->Start(check_data.name); check_data.is_running = true; @@ -183,6 +197,13 @@ TEST_F(StreamsTest, RestoreStreams) { stream_info.common_info.batch_interval = std::chrono::milliseconds((i + 1) * 10); stream_info.common_info.batch_size = 1000 + i; stream_check_data.owner = std::string{"owner"} + iteration_postfix; + + if (i == 1 || i == 3) { + stream_info.configs.emplace(std::string{"sasl.username"}, std::string{"username"} + iteration_postfix); + } + if (i == 2 || i == 3) { + stream_info.credentials.emplace(std::string{"sasl.password"}, std::string{"password"} + iteration_postfix); + } } mock_cluster_.CreateTopic(stream_info.topics[0]); @@ -198,6 +219,7 @@ TEST_F(StreamsTest, RestoreStreams) { EXPECT_EQ(stream_check_datas.size(), streams_->GetStreamInfo().size()); for (const auto &check_data : stream_check_datas) { ASSERT_NO_FATAL_FAILURE(CheckStreamStatus(check_data)); + ASSERT_NO_FATAL_FAILURE(CheckConfigAndCredentials(check_data)); } }; @@ -253,3 +275,32 @@ TEST_F(StreamsTest, CheckWithTimeout) { EXPECT_LE(timeout, elapsed); EXPECT_LE(elapsed, timeout * 1.2); } + +TEST_F(StreamsTest, CheckInvalidConfig) { + auto stream_info = CreateDefaultStreamInfo(); + const auto stream_name = GetDefaultStreamName(); + constexpr auto kInvalidConfigName = "doesnt.exist"; + constexpr auto kConfigValue = "myprecious"; + stream_info.configs.emplace(kInvalidConfigName, kConfigValue); + const auto checker = [](const std::string_view message) { + EXPECT_TRUE(message.find(kInvalidConfigName) != std::string::npos) << message; + EXPECT_TRUE(message.find(kConfigValue) != std::string::npos) << message; + }; + EXPECT_THROW_WITH_MSG(streams_->Create(stream_name, stream_info, std::nullopt), + integrations::kafka::SettingCustomConfigFailed, checker); +} + +TEST_F(StreamsTest, CheckInvalidCredentials) { + auto stream_info = CreateDefaultStreamInfo(); + const auto stream_name = GetDefaultStreamName(); + constexpr auto kInvalidCredentialName = "doesnt.exist"; + constexpr auto kCredentialValue = "myprecious"; + stream_info.credentials.emplace(kInvalidCredentialName, kCredentialValue); + const auto checker = [](const std::string_view message) { + EXPECT_TRUE(message.find(kInvalidCredentialName) != std::string::npos) << message; + EXPECT_TRUE(message.find(integrations::kReducted) != std::string::npos) << message; + EXPECT_TRUE(message.find(kCredentialValue) == std::string::npos) << message; + }; + EXPECT_THROW_WITH_MSG(streams_->Create(stream_name, stream_info, std::nullopt), + integrations::kafka::SettingCustomConfigFailed, checker); +} diff --git a/tests/unit/test_utils.hpp b/tests/unit/test_utils.hpp index d83254e56..c4657a36d 100644 --- a/tests/unit/test_utils.hpp +++ b/tests/unit/test_utils.hpp @@ -1,4 +1,4 @@ -// Copyright 2021 Memgraph Ltd. +// Copyright 2022 Memgraph Ltd. // // Use of this software is governed by the Business Source License // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source @@ -34,4 +34,16 @@ TResult ExpectNoError(const char *file, int line, TFunc func, TArgs &&...args) { } } // namespace test_utils -#define EXPECT_MGP_NO_ERROR(type, ...) test_utils::ExpectNoError(__FILE__, __LINE__, __VA_ARGS__) \ No newline at end of file +#define EXPECT_MGP_NO_ERROR(type, ...) test_utils::ExpectNoError(__FILE__, __LINE__, __VA_ARGS__) + +#define EXPECT_THROW_WITH_MSG(statement, expected_exception, msg_checker) \ + EXPECT_THROW( \ + { \ + try { \ + statement; \ + } catch (const expected_exception &e) { \ + EXPECT_NO_FATAL_FAILURE(msg_checker(e.what())); \ + throw; \ + } \ + }, \ + expected_exception); \ No newline at end of file