Add Kafka test

Signed-off-by: Nijat Khanbabayev <[email protected]>

csp/tests/adapters/conftest.py

@@ -9,9 +9,11 @@ def kafkabroker():
 
 
 @pytest.fixture(scope="module", autouse=True)
-def kafkaadapter(kafkabroker):
-    group_id = "group.id123"
-    _kafkaadapter = KafkaAdapterManager(
-        broker=kafkabroker, group_id=group_id, rd_kafka_conf_options={"allow.auto.create.topics": "true"}
-    )
+def kafkaadapterkwargs(kafkabroker):
+    return dict(broker=kafkabroker, group_id="group.id123", rd_kafka_conf_options={"allow.auto.create.topics": "true"})
+
+
+@pytest.fixture(scope="module", autouse=True)
+def kafkaadapter(kafkaadapterkwargs):
+    _kafkaadapter = KafkaAdapterManager(**kafkaadapterkwargs)
     return _kafkaadapter

csp/tests/adapters/test_kafka.py

@@ -386,3 +386,23 @@ def graph():
 
         # With bug this would deadlock
         csp.run(graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+
+    @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
+    def test_meta_field_map_extract_timestamp_from_field(self, kafkaadapterkwargs):
+        class SubData(csp.Struct):
+            msg: str
+            dt: datetime
+
+        kafkaadapter1 = KafkaAdapterManager(**kafkaadapterkwargs)
+
+        def graph_sub():
+            return kafkaadapter1.subscribe(
+                ts_type=SubData,
+                msg_mapper=RawTextMessageMapper(),
+                meta_field_map={"timestamp": "dt"},
+                topic="foobar",
+                extract_timestamp_from_field="dt",
+            )
+
+        with pytest.raises(ValueError):
+            csp.run(graph_sub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)