Added integration test for plain text message processing with Kafka

Author: starichkov

tests/test_cli_integration.py

@@ -2,6 +2,7 @@
 import subprocess
 import time
 import os
+from kafka import KafkaProducer
 
 
 def test_producer_and_consumer_via_scripts(tmp_path):
@@ -24,6 +25,8 @@ def test_producer_and_consumer_via_scripts(tmp_path):
         time.sleep(1)
 
         # 2. Capture consumer output
+        env["COVERAGE_PROCESS_START"] = ".coveragerc"  # Enable coverage for subprocess
+
         result = subprocess.run(
             ["python", "consumer.py", "--test-mode"],
             env=env,
@@ -34,3 +37,43 @@ def test_producer_and_consumer_via_scripts(tmp_path):
         )
 
         assert "note event" in result.stdout.lower()
+
+
+def test_plain_text_consumer(tmp_path):
+    topic = "test-topic"
+
+    with KafkaContainer(image="confluentinc/cp-kafka:7.9.2") as kafka:
+        bootstrap_servers = kafka.get_bootstrap_server()
+
+        # Create a producer that sends plain text messages
+        producer = KafkaProducer(
+            bootstrap_servers=bootstrap_servers,
+            key_serializer=lambda k: k.encode('utf-8') if k else None,
+            value_serializer=lambda v: v.encode('utf-8')  # Plain text serializer
+        )
+
+        # Send a plain text message
+        plain_text_message = "This is a plain text message"
+        producer.send(topic, key="plain-text-key", value=plain_text_message)
+        producer.flush()
+
+        time.sleep(1)
+
+        # Run the consumer and capture its output
+        env = os.environ.copy()
+        env["KAFKA_BOOTSTRAP_SERVERS"] = bootstrap_servers
+        env["KAFKA_TOPIC"] = topic
+        env["COVERAGE_PROCESS_START"] = ".coveragerc"  # Enable coverage for subprocess
+
+        result = subprocess.run(
+            ["python", "consumer.py", "--test-mode"],
+            env=env,
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+
+        # Verify that the plain text message was processed correctly
+        assert "📦 Plain" in result.stdout
+        assert plain_text_message in result.stdout