yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(test): fix formatting quirks

This commit is contained in:
Andrew Mayorov 2023-11-07 23:14:33 +07:00
parent c300eb41a7
commit 18cd98def6
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
2 changed files with 192 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_producer_SUITE.erl
View File

@ -1018,112 +1018,89 @@ hocon_config(Args, ConfigTemplateFun) ->
), ),
Hocon. Hocon.
%% erlfmt-ignore
hocon_config_template() -> hocon_config_template() ->
""" "bridges.kafka.{{ bridge_name }} {"
bridges.kafka.{{ bridge_name }} { "\n bootstrap_hosts = \"{{ kafka_hosts_string }}\""
bootstrap_hosts = \"{{ kafka_hosts_string }}\" "\n enable = true"
enable = true "\n authentication = {{{ authentication }}}"
authentication = {{{ authentication }}} "\n ssl = {{{ ssl }}}"
ssl = {{{ ssl }}} "\n local_topic = \"{{ local_topic }}\""
local_topic = \"{{ local_topic }}\" "\n kafka = {"
kafka = { "\n message = {"
message = { "\n key = \"${clientid}\""
key = \"${clientid}\" "\n value = \"${.payload}\""
value = \"${.payload}\" "\n timestamp = \"${timestamp}\""
timestamp = \"${timestamp}\" "\n }"
} "\n buffer = {"
buffer = { "\n memory_overload_protection = false"
memory_overload_protection = false "\n }"
} "\n partition_strategy = {{ partition_strategy }}"
partition_strategy = {{ partition_strategy }} "\n topic = \"{{ kafka_topic }}\""
topic = \"{{ kafka_topic }}\" "\n query_mode = {{ query_mode }}"
query_mode = {{ query_mode }} "\n }"
} "\n metadata_request_timeout = 5s"
metadata_request_timeout = 5s "\n min_metadata_refresh_interval = 3s"
min_metadata_refresh_interval = 3s "\n socket_opts {"
socket_opts { "\n nodelay = true"
nodelay = true "\n }"
} "\n connect_timeout = 5s"
connect_timeout = 5s "\n }".
}
""".
%% erlfmt-ignore
hocon_config_template_with_headers() -> hocon_config_template_with_headers() ->
""" "bridges.kafka.{{ bridge_name }} {"
bridges.kafka.{{ bridge_name }} { "\n bootstrap_hosts = \"{{ kafka_hosts_string }}\""
bootstrap_hosts = \"{{ kafka_hosts_string }}\" "\n enable = true"
enable = true "\n authentication = {{{ authentication }}}"
authentication = {{{ authentication }}} "\n ssl = {{{ ssl }}}"
ssl = {{{ ssl }}} "\n local_topic = \"{{ local_topic }}\""
local_topic = \"{{ local_topic }}\" "\n kafka = {"
kafka = { "\n message = {"
message = { "\n key = \"${clientid}\""
key = \"${clientid}\" "\n value = \"${.payload}\""
value = \"${.payload}\" "\n timestamp = \"${timestamp}\""
timestamp = \"${timestamp}\" "\n }"
} "\n buffer = {"
buffer = { "\n memory_overload_protection = false"
memory_overload_protection = false "\n }"
} "\n kafka_headers = \"{{ kafka_headers }}\""
kafka_headers = \"{{ kafka_headers }}\" "\n kafka_header_value_encode_mode: json"
kafka_header_value_encode_mode: json "\n kafka_ext_headers: {{{ kafka_ext_headers }}}"
kafka_ext_headers: {{{ kafka_ext_headers }}} "\n partition_strategy = {{ partition_strategy }}"
partition_strategy = {{ partition_strategy }} "\n topic = \"{{ kafka_topic }}\""
topic = \"{{ kafka_topic }}\" "\n query_mode = {{ query_mode }}"
query_mode = {{ query_mode }} "\n }"
} "\n metadata_request_timeout = 5s"
metadata_request_timeout = 5s "\n min_metadata_refresh_interval = 3s"
min_metadata_refresh_interval = 3s "\n socket_opts {"
socket_opts { "\n nodelay = true"
nodelay = true "\n }"
} "\n connect_timeout = 5s"
connect_timeout = 5s "\n }".
}
""".
%% erlfmt-ignore
hocon_config_template_authentication("none") -> hocon_config_template_authentication("none") ->
"none"; "none";
hocon_config_template_authentication(#{"mechanism" := _}) -> hocon_config_template_authentication(#{"mechanism" := _}) ->
""" "{"
{ "\n mechanism = {{ mechanism }}"
mechanism = {{ mechanism }} "\n password = {{ password }}"
password = {{ password }} "\n username = {{ username }}"
username = {{ username }} "\n }";
}
""";
hocon_config_template_authentication(#{"kerberos_principal" := _}) -> hocon_config_template_authentication(#{"kerberos_principal" := _}) ->
""" "{"
{ "\n kerberos_principal = \"{{ kerberos_principal }}\""
kerberos_principal = \"{{ kerberos_principal }}\" "\n kerberos_keytab_file = \"{{ kerberos_keytab_file }}\""
kerberos_keytab_file = \"{{ kerberos_keytab_file }}\" "\n }".
}
""".
%% erlfmt-ignore
hocon_config_template_ssl(Map) when map_size(Map) =:= 0 -> hocon_config_template_ssl(Map) when map_size(Map) =:= 0 ->
""" "{ enable = false }";
{
enable = false
}
""";
hocon_config_template_ssl(#{"enable" := "false"}) -> hocon_config_template_ssl(#{"enable" := "false"}) ->
""" "{ enable = false }";
{
enable = false
}
""";
hocon_config_template_ssl(#{"enable" := "true"}) -> hocon_config_template_ssl(#{"enable" := "true"}) ->
""" "{ enable = true"
{ "\n cacertfile = \"{{{cacertfile}}}\""
enable = true "\n certfile = \"{{{certfile}}}\""
cacertfile = \"{{{cacertfile}}}\" "\n keyfile = \"{{{keyfile}}}\""
certfile = \"{{{certfile}}}\" "\n }".
keyfile = \"{{{keyfile}}}\"
}
""".
kafka_hosts_string(tcp, none) -> kafka_hosts_string(tcp, none) ->
kafka_hosts_string(); kafka_hosts_string();

250
apps/emqx_bridge_kafka/test/emqx_bridge_kafka_tests.erl
View File

@ -223,144 +223,136 @@ check_atom_key(Conf) when is_map(Conf) ->
%% Data section %% Data section
%%=========================================================================== %%===========================================================================
%% erlfmt-ignore
kafka_producer_old_hocon(_WithLocalTopic = true) -> kafka_producer_old_hocon(_WithLocalTopic = true) ->
kafka_producer_old_hocon("mqtt {topic = \"mqtt/local\"}\n"); kafka_producer_old_hocon("mqtt {topic = \"mqtt/local\"}\n");
kafka_producer_old_hocon(_WithLocalTopic = false) -> kafka_producer_old_hocon(_WithLocalTopic = false) ->
kafka_producer_old_hocon("mqtt {}\n"); kafka_producer_old_hocon("mqtt {}\n");
kafka_producer_old_hocon(MQTTConfig) when is_list(MQTTConfig) -> kafka_producer_old_hocon(MQTTConfig) when is_list(MQTTConfig) ->
""" [
bridges.kafka { "bridges.kafka {"
myproducer { "\n myproducer {"
authentication = \"none\" "\n authentication = \"none\""
bootstrap_hosts = \"toxiproxy:9292\" "\n bootstrap_hosts = \"toxiproxy:9292\""
connect_timeout = \"5s\" "\n connect_timeout = \"5s\""
metadata_request_timeout = \"5s\" "\n metadata_request_timeout = \"5s\""
min_metadata_refresh_interval = \"3s\" "\n min_metadata_refresh_interval = \"3s\""
producer { "\n producer {"
kafka { "\n kafka {"
buffer { "\n buffer {"
memory_overload_protection = false "\n memory_overload_protection = false"
mode = \"memory\" "\n mode = \"memory\""
per_partition_limit = \"2GB\" "\n per_partition_limit = \"2GB\""
segment_bytes = \"100MB\" "\n segment_bytes = \"100MB\""
} "\n }"
compression = \"no_compression\" "\n compression = \"no_compression\""
max_batch_bytes = \"896KB\" "\n max_batch_bytes = \"896KB\""
max_inflight = 10 "\n max_inflight = 10"
message { "\n message {"
key = \"${.clientid}\" "\n key = \"${.clientid}\""
timestamp = \"${.timestamp}\" "\n timestamp = \"${.timestamp}\""
value = \"${.}\" "\n value = \"${.}\""
} "\n }"
partition_count_refresh_interval = \"60s\" "\n partition_count_refresh_interval = \"60s\""
partition_strategy = \"random\" "\n partition_strategy = \"random\""
required_acks = \"all_isr\" "\n required_acks = \"all_isr\""
topic = \"test-topic-two-partitions\" "\n topic = \"test-topic-two-partitions\""
} "\n }",
""" ++ MQTTConfig ++ MQTTConfig,
""" "\n }"
} "\n socket_opts {"
socket_opts { "\n nodelay = true"
nodelay = true "\n recbuf = \"1024KB\""
recbuf = \"1024KB\" "\n sndbuf = \"1024KB\""
sndbuf = \"1024KB\" "\n }"
} "\n ssl {enable = false, verify = \"verify_peer\"}"
ssl {enable = false, verify = \"verify_peer\"} "\n }"
} "\n}"
} ].
""".
kafka_producer_new_hocon() -> kafka_producer_new_hocon() ->
"" "bridges.kafka {"
"\n" "\n myproducer {"
"bridges.kafka {\n" "\n authentication = \"none\""
" myproducer {\n" "\n bootstrap_hosts = \"toxiproxy:9292\""
" authentication = \"none\"\n" "\n connect_timeout = \"5s\""
" bootstrap_hosts = \"toxiproxy:9292\"\n" "\n metadata_request_timeout = \"5s\""
" connect_timeout = \"5s\"\n" "\n min_metadata_refresh_interval = \"3s\""
" metadata_request_timeout = \"5s\"\n" "\n kafka {"
" min_metadata_refresh_interval = \"3s\"\n" "\n buffer {"
" kafka {\n" "\n memory_overload_protection = false"
" buffer {\n" "\n mode = \"memory\""
" memory_overload_protection = false\n" "\n per_partition_limit = \"2GB\""
" mode = \"memory\"\n" "\n segment_bytes = \"100MB\""
" per_partition_limit = \"2GB\"\n" "\n }"
" segment_bytes = \"100MB\"\n" "\n compression = \"no_compression\""
" }\n" "\n max_batch_bytes = \"896KB\""
" compression = \"no_compression\"\n" "\n max_inflight = 10"
" max_batch_bytes = \"896KB\"\n" "\n message {"
" max_inflight = 10\n" "\n key = \"${.clientid}\""
" message {\n" "\n timestamp = \"${.timestamp}\""
" key = \"${.clientid}\"\n" "\n value = \"${.}\""
" timestamp = \"${.timestamp}\"\n" "\n }"
" value = \"${.}\"\n" "\n partition_count_refresh_interval = \"60s\""
" }\n" "\n partition_strategy = \"random\""
" partition_count_refresh_interval = \"60s\"\n" "\n required_acks = \"all_isr\""
" partition_strategy = \"random\"\n" "\n topic = \"test-topic-two-partitions\""
" required_acks = \"all_isr\"\n" "\n }"
" topic = \"test-topic-two-partitions\"\n" "\n local_topic = \"mqtt/local\""
" }\n" "\n socket_opts {"
" local_topic = \"mqtt/local\"\n" "\n nodelay = true"
" socket_opts {\n" "\n recbuf = \"1024KB\""
" nodelay = true\n" "\n sndbuf = \"1024KB\""
" recbuf = \"1024KB\"\n" "\n }"
" sndbuf = \"1024KB\"\n" "\n ssl {enable = false, verify = \"verify_peer\"}"
" }\n" "\n resource_opts {"
" ssl {enable = false, verify = \"verify_peer\"}\n" "\n health_check_interval = 10s"
" resource_opts {\n" "\n }"
" health_check_interval = 10s\n" "\n }"
" }\n" "\n}".
" }\n"
"}\n"
"".
%% erlfmt-ignore
kafka_consumer_hocon() -> kafka_consumer_hocon() ->
""" "bridges.kafka_consumer.my_consumer {"
bridges.kafka_consumer.my_consumer { "\n enable = true"
enable = true "\n bootstrap_hosts = \"kafka-1.emqx.net:9292\""
bootstrap_hosts = \"kafka-1.emqx.net:9292\" "\n connect_timeout = 5s"
connect_timeout = 5s "\n min_metadata_refresh_interval = 3s"
min_metadata_refresh_interval = 3s "\n metadata_request_timeout = 5s"
metadata_request_timeout = 5s "\n authentication = {"
authentication = { "\n mechanism = plain"
mechanism = plain "\n username = emqxuser"
username = emqxuser "\n password = password"
password = password "\n }"
} "\n kafka {"
kafka { "\n max_batch_bytes = 896KB"
max_batch_bytes = 896KB "\n max_rejoin_attempts = 5"
max_rejoin_attempts = 5 "\n offset_commit_interval_seconds = 3s"
offset_commit_interval_seconds = 3s "\n offset_reset_policy = latest"
offset_reset_policy = latest "\n }"
} "\n topic_mapping = ["
topic_mapping = [ "\n {"
{ "\n kafka_topic = \"kafka-topic-1\""
kafka_topic = \"kafka-topic-1\" "\n mqtt_topic = \"mqtt/topic/1\""
mqtt_topic = \"mqtt/topic/1\" "\n qos = 1"
qos = 1 "\n payload_template = \"${.}\""
payload_template = \"${.}\" "\n },"
}, "\n {"
{ "\n kafka_topic = \"kafka-topic-2\""
kafka_topic = \"kafka-topic-2\" "\n mqtt_topic = \"mqtt/topic/2\""
mqtt_topic = \"mqtt/topic/2\" "\n qos = 2"
qos = 2 "\n payload_template = \"v = ${.value}\""
payload_template = \"v = ${.value}\" "\n }"
} "\n ]"
] "\n key_encoding_mode = none"
key_encoding_mode = none "\n value_encoding_mode = none"
value_encoding_mode = none "\n ssl {"
ssl { "\n enable = false"
enable = false "\n verify = verify_none"
verify = verify_none "\n server_name_indication = \"auto\""
server_name_indication = \"auto\" "\n }"
} "\n resource_opts {"
resource_opts { "\n health_check_interval = 10s"
health_check_interval = 10s "\n }"
} "\n }".
}
""".
%% assert compatibility %% assert compatibility
bridge_schema_json_test() -> bridge_schema_json_test() ->