Increase encoder test coverage

singer_sdk/_singerlib/encoding/base.py

@@ -58,11 +58,11 @@ def _process_lines(self, file_input: t.IO[T]) -> t.Counter[str]:
             elif record_type == SingerMessageType.STATE:
                 self._process_state_message(line_dict)
 
-            elif record_type == SingerMessageType.BATCH:  # pragma: no cover
+            elif record_type == SingerMessageType.BATCH:
                 self._process_batch_message(line_dict)
 
             else:
-                self._process_unknown_message(line_dict)  # pragma: no cover
+                self._process_unknown_message(line_dict)
 
             stats[record_type] += 1
 
@@ -106,7 +106,7 @@ def _process_activate_version_message(self, message_dict: dict) -> None: ...
     @abc.abstractmethod
     def _process_batch_message(self, message_dict: dict) -> None: ...
 
-    def _process_unknown_message(self, message_dict: dict) -> None:  # noqa: PLR6301 # pragma: no cover
+    def _process_unknown_message(self, message_dict: dict) -> None:  # noqa: PLR6301
         """Internal method to process unknown message types from a Singer tap.
 
         Args:

tests/core/test_io.py

@@ -7,6 +7,7 @@
 import itertools
 import json
 from contextlib import nullcontext, redirect_stdout
+from textwrap import dedent
 
 import pytest
 
@@ -59,6 +60,29 @@ def test_deserialize(line, expected, exception):
         assert reader.deserialize_json(line) == expected
 
 
+def test_listen():
+    reader = DummyReader()
+    input_lines = io.StringIO(
+        dedent("""\
+        {"type": "SCHEMA", "stream": "users", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 1, "value": 1.23}}
+        {"type": "RECORD", "stream": "users", "record": {"id": 2, "value": 2.34}}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}}}}
+        {"type": "SCHEMA", "stream": "batches", "schema": {"type": "object", "properties": {"id": {"type": "integer"}, "value": {"type": "number"}}}}
+        {"type": "BATCH", "stream": "batches", "encoding": {"format": "jsonl", "compression": "gzip"}, "manifest": ["file1.jsonl.gz", "file2.jsonl.gz"]}
+        {"type": "STATE", "value": {"bookmarks": {"users": {"id": 2}, "batches": {"id": 1000000}}}}
+    """)  # noqa: E501
+    )
+    reader.listen(input_lines)
+
+
+def test_listen_unknown_message():
+    reader = DummyReader()
+    input_lines = io.StringIO('{"type": "UNKNOWN"}\n')
+    with pytest.raises(ValueError, match="Unknown message type"):
+        reader.listen(input_lines)
+
+
 def test_write_message():
     writer = SingerWriter()
     message = RecordMessage(