Dev basic msg relay test

src/cgw_connection_server.rs

@@ -733,7 +733,10 @@ impl CGWConnectionServer {
             debug!("Received {num_of_msg_read} messages from NB API, processing...");
 
             let partition_mapping = self.nb_api_client.get_partition_to_local_shard_mapping();
-            debug!("Kafka partitions idx:key mapping info: {:?}", partition_mapping);
+            debug!(
+                "Kafka partitions idx:key mapping info: {:?}",
+                partition_mapping
+            );
             if !partition_mapping.is_empty() {
                 partition_array_idx += 1;
                 if partition_array_idx >= partition_mapping.len() {

tests/conftest.py

@@ -5,6 +5,7 @@
 from client_simulator.src.simulation_runner import Device as DeviceSimulator
 from kafka_producer.src.producer import Producer as KafkaProducer
 from kafka_producer.src.consumer import Consumer as KafkaConsumer
+from kafka_producer.src.admin import Admin as KafkaAdmin
 from psql_client.psql_client import PostgreSQLClient as PSQLClient
 from redis_client.redis_client import RedisClient as RedisClient
 import requests
@@ -26,6 +27,10 @@ def default_kafka_group() -> str:
     def default_shard_id() -> int:
         return 0
 
+    @staticmethod
+    def default_producer_topic() -> str:
+        return 'CnC'
+
     def __init__(self):
         device = DeviceSimulator(
                 mac=self.default_dev_sim_mac(),
@@ -53,9 +58,11 @@ def __init__(self):
 
         producer = KafkaProducer(db='localhost:9092', topic='CnC')
         consumer = KafkaConsumer(db='localhost:9092', topic='CnC_Res', consumer_timeout=12000)
+        admin = KafkaAdmin(host='localhost', port=9092)
 
         self.kafka_producer = producer
         self.kafka_consumer = consumer
+        self.kafka_admin = admin
 
         psql_client = PSQLClient(host="localhost", port=5432, database="cgw", user="cgw", password="123")
         self.psql_client = psql_client
@@ -90,6 +97,7 @@ def test_context():
 
     ctx.kafka_producer.disconnect()
     ctx.kafka_consumer.disconnect()
+    ctx.kafka_admin.disconnect()
 
     ctx.psql_client.disconnect()
     ctx.redis_client.disconnect()
@@ -118,6 +126,13 @@ def kafka_probe(test_context):
     # We have to clear any messages before we can work with kafka
     test_context.kafka_consumer.flush()
 
+@pytest.fixture(scope='function')
+def kafka_admin_probe(test_context):
+    try:
+        test_context.kafka_admin.connect()
+    except:
+        raise Exception('Failed to connect to Kafka broker!')
+
 @pytest.fixture(scope='function')
 def device_sim_connect(test_context):
     # Make sure we initiate connect;

tests/run.sh

@@ -9,6 +9,8 @@ ln -sf ../utils/client_simulator/sim_data sim_data
 ln -sf ../utils/kafka_producer/kafka_data kafka_data
 ln -sf ../utils/cert_generator/certs/client/ certs
 ln -sf ../utils/cert_generator/certs/ca/ ca-certs
+ln -sf ../utils/client_simulator/ client_simulator
+ln -sf ../utils/kafka_producer/ kafka_producer
 ln -sf ../utils/psql_client/ psql_client
 ln -sf ../utils/redis_client/ redis_client