Kinesis: Adjust code for lorrystream version 0.0.3. Add tests.

On the last occasion, dependencies to lorrystream have not been verified
correctly, so it was expectable that things go south.

This patch fixes them, and adds rudimentary software tests to avoid
tripping into the same situation as before.

CHANGES.md

@@ -15,6 +15,7 @@
 - MongoDB: Make `migr8 extract` and `migr8 export` accept the `--limit` option
 - MongoDB: Fix indentation in prettified SQL output of `migr8 translate`
 - MongoDB: Add capability to give type hints and add transformations
+- Dependencies: Adjust code for lorrystream version 0.0.3
 
 ## 2024/07/25 v0.0.16
 - `ctk load table`: Added support for MongoDB Change Streams

cratedb_toolkit/iac/aws.py

@@ -1,9 +1,10 @@
+from lorrystream.carabas.aws import DynamoDBKinesisPipe, RDSPostgreSQLDMSKinesisPipe
 from lorrystream.carabas.aws.function.model import LambdaFactory
 from lorrystream.carabas.aws.function.oci import LambdaPythonImage
-from lorrystream.carabas.aws.stack import DynamoDBKinesisPipe
 
 __all__ = [
+    "DynamoDBKinesisPipe",
     "LambdaFactory",
     "LambdaPythonImage",
-    "DynamoDBKinesisPipe",
+    "RDSPostgreSQLDMSKinesisPipe",
 ]

cratedb_toolkit/io/processor/kinesis_lambda.py

@@ -25,7 +25,7 @@
 # /// script
 # requires-python = ">=3.9"
 # dependencies = [
-#   "commons-codec==0.0.3",
+#   "commons-codec==0.0.4",
 #   "sqlalchemy-cratedb==0.38.0",
 # ]
 # ///

pyproject.toml

@@ -151,7 +151,7 @@ io = [
   "sqlalchemy>=2",
 ]
 kinesis = [
-  "lorrystream @ git+https://github.com/daq-tools/lorrystream.git@kinesis",
+  "lorrystream[carabas]==0.0.3",
 ]
 mongodb = [
   "commons-codec[mongodb,zyp]==0.0.4",

tests/io/test_iac.py

@@ -0,0 +1,10 @@
+# ruff: noqa: F401
+
+
+def test_iac_imports():
+    from cratedb_toolkit.iac.aws import (
+        DynamoDBKinesisPipe,
+        LambdaFactory,
+        LambdaPythonImage,
+        RDSPostgreSQLDMSKinesisPipe,
+    )

tests/io/test_processor.py

@@ -0,0 +1,31 @@
+import os
+import sys
+
+import pytest
+
+
+@pytest.fixture
+def reset_handler():
+    try:
+        del sys.modules["cratedb_toolkit.io.processor.kinesis_lambda"]
+    except KeyError:
+        pass
+
+
+def test_processor_invoke_no_records(reset_handler, mocker, caplog):
+    """
+    Roughly verify that the unified Lambda handler works.
+    """
+
+    # Configure environment variables.
+    handler_environment = {
+        "MESSAGE_FORMAT": "dms",
+    }
+    mocker.patch.dict(os.environ, handler_environment)
+
+    from cratedb_toolkit.io.processor.kinesis_lambda import handler
+
+    event = {"Records": []}
+    handler(event, None)
+
+    assert "Successfully processed 0 records" in caplog.messages