add tests for utils and start for sniff

Author: rerpha

src/saluki/utils.py

@@ -44,8 +44,8 @@ def deserialise_and_print_messages(
             if partition is not None and msg.partition() != partition:
                 continue
             schema, deserialised = _try_to_deserialise_message(msg.value())
-            if schemas_to_filter_to is not None and schema in schemas_to_filter_to:
-                break
+            if schemas_to_filter_to is not None and schema not in schemas_to_filter_to:
+                continue
             time = _parse_timestamp(msg)
             logger.info(f"{msg.offset()} ({time}):({schema}) {deserialised}")
         except Exception as e:

tests/test_sniff.py

@@ -0,0 +1,24 @@
+from unittest.mock import patch
+
+from confluent_kafka.admin import ClusterMetadata, BrokerMetadata, TopicMetadata
+
+from saluki.sniff import sniff
+
+def test_sniff_with_two_partitions_in_a_topic():
+    with patch("saluki.sniff.AdminClient") as a, patch("saluki.sniff.Consumer") as c, patch("saluki.sniff.logger") as logger:
+        fake_cluster_md = ClusterMetadata()
+        broker1 = BrokerMetadata()
+        broker2 = BrokerMetadata()
+        fake_cluster_md.brokers = {0: broker1, 1: broker2}
+
+        topic1 = TopicMetadata()
+        topic2 = TopicMetadata()
+
+        fake_cluster_md.topics = {
+            "topic1": topic1,
+            "topic2": topic2
+        }
+        a.list_topics.return_value = fake_cluster_md
+        sniff("whatever")
+
+        # TODO 

tests/test_utils.py

@@ -2,9 +2,11 @@
 
 import pytest
 from confluent_kafka import Message
+from streaming_data_types import serialise_f144
 from streaming_data_types.forwarder_config_update_fc00 import (
     ConfigurationUpdate,
     StreamInfo,
+    serialise_fc00
 )
 
 from saluki.utils import (
@@ -74,6 +76,21 @@ def test_deserialising_message_which_raises_does_not_stop_loop(mock_message):
         assert logger.info.call_count == 1
 
 
+def test_deserialising_with_schema_list_ignores_messages_with_schema_not_in_list(mock_message):
+    with patch("saluki.utils.logger") as logger:
+        ok_message = Mock(spec=Message)
+        ok_message.value.return_value = serialise_fc00(config_change=1, streams=[])
+        ok_message.error.return_value = False
+        ok_message.timestamp.return_value = 2, 1
+
+        mock_message.value.return_value = serialise_f144(source_name="test", value=123)
+        mock_message.error.return_value = False
+        mock_message.timestamp.return_value = 2, 1
+
+        deserialise_and_print_messages([mock_message, ok_message], None, schemas_to_filter_to=["fc00"])
+        assert logger.info.call_count == 1
+
+
 def test_message_that_has_valid_schema_but_empty_payload():
     with pytest.raises(Exception):
         # Empty fc00 message - valid schema but not valid payload