test: integration test for dynamic metrics

Author: d-perl

src/tests/metrics_integration_test.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use redis::Commands;
 use testcontainers_modules::{
     redis::Redis,
     testcontainers::{ContainerAsync, runners::AsyncRunner},
@@ -18,9 +19,14 @@ async fn create_redis() -> (ContainerAsync<Redis>, String, u16) {
     (r, h, p)
 }
 
-fn config(redis_url: String, redis_port: u16, metrics_url: String) -> IngestorConfig {
+fn config(
+    redis_url: String,
+    redis_port: u16,
+    metrics_url: String,
+    additional: &str,
+) -> IngestorConfig {
     toml::from_str(
-        format!(
+        (format!(
             r#"enable_logging=false
 
 [redis]
@@ -46,8 +52,8 @@ watchdog_interval = {{ Secondly = 20 }}
 publish_interval = {{ Millis = 5 }}
 
 "#
-        )
-        .as_str(),
+        ) + additional)
+            .as_str(),
     )
     .unwrap()
 }
@@ -67,7 +73,7 @@ async fn test_system_metrics_propagated() {
         let (redis_container, redis_url, redis_port) = create_redis().await;
         let _ = redis_container.start().await;
         let config: &'static mut IngestorConfig =
-            Box::leak(Box::new(config(redis_url, redis_port, server.url())));
+            Box::leak(Box::new(config(redis_url, redis_port, server.url(), "")));
 
         let handle = spawn(metrics_loop(config));
         while !mock.matched_async().await {
@@ -81,3 +87,43 @@ async fn test_system_metrics_propagated() {
     .await
     .unwrap()
 }
+
+const ENCODED_METRIC: &[u8; 539] = b"\x81\xad__bec_codec__\x83\xacencoder_name\xaaBECMessage\xa9type_name\xb4DynamicMetricMessage\xa4data\x83\xa8metadata\x80\xa7metrics\x84\xa1a\x81\xad__bec_codec__\x83\xacencoder_name\xa9BaseModel\xa9type_name\xb6_StrDynamicMetricValue\xa4data\x82\xa5value\xa6test_a\xa9type_name\xa3str\xa1b\x81\xad__bec_codec__\x83\xacencoder_name\xa9BaseModel\xa9type_name\xb6_IntDynamicMetricValue\xa4data\x82\xa5value\x05\xa9type_name\xa3int\xa1c\x81\xad__bec_codec__\x83\xacencoder_name\xa9BaseModel\xa9type_name\xb8_FloatDynamicMetricValue\xa4data\x82\xa5value\xcb@#\xcc\xcc\xcc\xcc\xcc\xcd\xa9type_name\xa5float\xa1d\x81\xad__bec_codec__\x83\xacencoder_name\xa9BaseModel\xa9type_name\xb7_BoolDynamicMetricValue\xa4data\x82\xa5value\xc2\xa9type_name\xa4bool\xa9timestamp\xcbA\xdal\xa4\x90\x93\x8f\xca";
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_dynamic_metrics() {
+    let mut server = mockito::Server::new_async().await;
+    let mock = server.mock("POST", "/").expect(4).create();
+
+    let test = tokio::time::timeout(Duration::from_secs(10), async {
+        let (redis_container, redis_url, redis_port) = create_redis().await;
+        let _ = redis_container.start().await;
+        let config: &'static mut IngestorConfig = Box::leak(Box::new(config(
+            redis_url,
+            redis_port,
+            server.url(),
+            "
+[metrics.dynamic]
+dynamic_metric = { read_type = \"PubSub\", key = \"info/dynamic_metrics/test\" }
+",
+        )));
+
+        let handle = spawn(metrics_loop(config));
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        let mut conn = redis::Client::open(config.redis.url.full_url())
+            .unwrap()
+            .get_connection()
+            .unwrap();
+        let _: Result<String, _> = dbg!(conn.publish("info/dynamic_metrics/test", ENCODED_METRIC));
+        while !mock.matched_async().await {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        handle.abort();
+        let _ = redis_container.stop_with_timeout(Some(0)).await;
+        let _ = handle.await;
+    })
+    .await;
+
+    mock.assert(); // check no more came after finishing
+    test.unwrap();
+}