more tests

Author: rerpha

pyproject.toml

@@ -40,7 +40,7 @@ exclude_lines = [
   "if typing.TYPE_CHECKING:",
   "@abstractmethod",
 ]
-omit = ["main.py"]
+omit = ["main.py"]  # No logic here besides argparse
 
 
 [tool.coverage.html]

tests/test_consume.py

@@ -1,16 +1,84 @@
+from unittest import mock
 from unittest.mock import patch
 
 import pytest
+from confluent_kafka import TopicPartition
 
 from saluki.consume import consume
 
+
 @patch("saluki.consume.Consumer")
 def test_go_forwards_with_no_offset_raises(_):
     with pytest.raises(ValueError):
         consume("broker", "topic", go_forwards=True, offset=None)
 
-# test that tries going forwards that consumes from offset
 
-# test that checks start offset
+@patch("saluki.consume.Consumer")
+def test_go_forwards_with_offset_assigns_at_offset(mock_consumer):
+    expected_topic = "topic"
+    expected_offset = 1234
+    expected_partition = 1
+    consume(
+        "broker",
+        expected_topic,
+        go_forwards=True,
+        offset=expected_offset,
+        partition=expected_partition,
+    )
+    mock_assign = mock_consumer.return_value.assign
+
+    mock_assign.assert_called_with(
+        [TopicPartition(expected_topic, expected_partition, expected_offset)]
+    )
+
+
+@patch("saluki.consume.Consumer")
+def test_consume_with_offset_and_num_of_messages_goes_back_offset_minus_messages(
+    mock_consumer,
+):
+    expected_offset = 1234
+    expected_topic = "sometopic"
+    num_messages = 3
+    expected_start_offset = expected_offset - num_messages + 1
+
+    consume("broker", expected_topic, offset=expected_offset, num_messages=num_messages)
+
+    mock_assign = mock_consumer.return_value.assign
+    mock_assign.assert_called_once()
+
+    mock_assign_call = mock_assign.call_args.args[0][0]
+    assert mock_assign_call.topic == expected_topic
+    assert mock_assign_call.offset == expected_start_offset
+
+
+@patch("saluki.consume.Consumer")
+def test_consume_with_no_offset_and_num_of_messages_goes_back_high_watermark_minus_messages(
+    mock_consumer,
+):
+    expected_topic = "sometopic"
+    num_messages = 3
+    high_watermark_offset = 2345
+    expected_start_offset = high_watermark_offset - num_messages
+
+    mock_consumer.return_value.get_watermark_offsets.return_value = (
+        None,
+        high_watermark_offset,
+    )
+
+    consume("broker", topic=expected_topic, num_messages=num_messages)
+    mock_assign = mock_consumer.return_value.assign
+    mock_assign.assert_called_once()
+
+    mock_assign_call = mock_assign.call_args.args[0][0]
+
+    assert mock_assign_call.topic == expected_topic
+    assert mock_assign_call.offset == expected_start_offset
+
 
-# test that catches exception
+def test_consume_but_exception_thrown_consumer_is_closed():
+    with (
+        mock.patch("saluki.consume.Consumer") as c,
+    ):
+        c.return_value.consume.side_effect = Exception
+        consume("somebroker", "sometopic", num_messages=1)
+        c.return_value.close.assert_called_once()

tests/test_listen.py

@@ -5,6 +5,7 @@
 from saluki.listen import listen
 from unittest import mock
 
+
 def test_listen_with_partition_assigns_to_partition():
     expected_partition = 123
     topic = "sometopic"

tests/test_utils.py

@@ -1,6 +1,23 @@
+from unittest.mock import Mock
+
 import pytest
 
-from saluki.utils import parse_kafka_uri
+from saluki.utils import parse_kafka_uri, _parse_timestamp, _deserialise_and_print_messages
+from confluent_kafka import Message
+
+@pytest.fixture
+def mock_message():
+    return Mock(spec=Message)
+
+def test_deserialising_message_with_no_message_continues(mock_message):
+    _deserialise_and_print_messages([mock_message])
+
+def test_deserialising_message_with_error_continues(mock_message):
+    pass
+
+def test_deserialising_message_with_wrong_partition_continues(mock_message):
+    pass
+
 
 
 # test with normal payload
@@ -11,8 +28,14 @@
 
 # test exception while deserialising
 
-# test _parse_timestamp
 
+def test_parse_timestamp_with_valid_timestamp(mock_message):
+    mock_message.timestamp.return_value = (1, 1753434939336)
+    assert _parse_timestamp(mock_message) == '2025-07-25 10:15:39.336000'
+
+def test_parse_timestamp_with_timestamp_not_available(mock_message):
+    mock_message.timestamp.return_value = (2, "blah")
+    assert _parse_timestamp(mock_message) == "Unknown"
 
 def test_uri_with_broker_name_and_topic_successfully_split():
     test_broker = "localhost"