DynamoDB: Add integration test for full-load operation

.github/workflows/tests.yml

@@ -52,7 +52,9 @@ jobs:
           pip install --use-pep517 --prefer-binary --editable=.[develop,test]
 
       - name: Run linters and software tests
-        run: poe check
+        run: |
+          poe lint
+          poe -- -m 'not (dynamodb or mongodb)'
 
       # https://github.com/codecov/codecov-action
       - name: Upload coverage results to Codecov
@@ -67,6 +69,61 @@ jobs:
           fail_ci_if_error: true
 
 
+  test-dynamodb:
+    name: "
+    DynamoDB: Python ${{ matrix.python-version }}
+    "
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: ['ubuntu-latest']
+        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
+
+    env:
+      OS: ${{ matrix.os }}
+      PYTHON: ${{ matrix.python-version }}
+
+    steps:
+
+      - name: Acquire sources
+        uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          architecture: x64
+          cache: 'pip'
+          cache-dependency-path:
+            pyproject.toml
+
+      - name: Set up project
+        run: |
+
+          # `setuptools 0.64.0` adds support for editable install hooks (PEP 660).
+          # https://github.com/pypa/setuptools/blob/main/CHANGES.rst#v6400
+          pip install "setuptools>=64" --upgrade
+
+          # Install package in editable mode.
+          pip install --use-pep517 --prefer-binary --editable=.[mongodb,develop,test]
+
+      - name: Run linters and software tests
+        run: poe -- -m 'dynamodb'
+
+      # https://github.com/codecov/codecov-action
+      - name: Upload coverage results to Codecov
+        uses: codecov/codecov-action@v4
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+        with:
+          files: ./coverage.xml
+          flags: dynamodb
+          env_vars: OS,PYTHON
+          name: codecov-umbrella
+          fail_ci_if_error: true
+
+
   test-mongodb:
     name: "
     MongoDB: Python ${{ matrix.python-version }}
@@ -107,7 +164,7 @@ jobs:
           pip install --use-pep517 --prefer-binary --editable=.[mongodb,develop,test]
 
       - name: Run linters and software tests
-        run: poe check
+        run: poe -- -m 'mongodb'
 
       # https://github.com/codecov/codecov-action
       - name: Upload coverage results to Codecov

doc/development.md

@@ -1,12 +1,17 @@
 # Development Sandbox
 
-Acquire source code, install development sandbox, and invoke software tests.
+Acquire source code, and install development sandbox.
 ```shell
 git clone https://github.com/daq-tools/commons-codec
 cd commons-codec
 python3 -m venv .venv
 source .venv/bin/activate
 pip install --editable='.[all,develop,doc,test]'
+```
+
+Invoke software tests.
+```
+export TC_KEEPALIVE=true
 poe check
 ```
 

pyproject.toml

@@ -138,6 +138,7 @@ optional-dependencies.release = [
   "twine<6",
 ]
 optional-dependencies.test = [
+  "cratedb-toolkit[testing]",
   "pytest<9",
   "pytest-cov<6",
   "pytest-mock<4",
@@ -245,6 +246,8 @@ python_files = [
 ]
 xfail_strict = true
 markers = [
+  "dynamodb",
+  "mongodb",
   "tasmota",
   "wemos",
 ]
@@ -329,6 +332,12 @@ release = [
   { cmd = "twine upload --skip-existing dist/*" },
 ]
 
-test = [
-  { cmd = "pytest" },
-]
+[tool.poe.tasks.test]
+cmd = "pytest"
+help = "Invoke software tests"
+
+[tool.poe.tasks.test.args.expression]
+options = [ "-k" ]
+
+[tool.poe.tasks.test.args.marker]
+options = [ "-m" ]

tests/transform/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+
+RESET_TABLES = [
+    "from.dynamodb",
+]
+
+
+@pytest.fixture(scope="function")
+def cratedb(cratedb_service):
+    """
+    Provide a fresh canvas to each test case invocation, by resetting database content.
+    """
+    cratedb_service.reset(tables=RESET_TABLES)
+    yield cratedb_service

tests/transform/test_dynamodb_cdc.py

@@ -6,6 +6,9 @@
 from commons_codec.model import DualRecord, SQLOperation
 from commons_codec.transform.dynamodb import CrateDBTypeDeserializer, DynamoDBCDCTranslator
 
+pytestmark = pytest.mark.dynamodb
+
+
 READING_BASIC = {"device": "foo", "temperature": 42.42, "humidity": 84.84}
 
 MSG_UNKNOWN_SOURCE = {

tests/transform/test_dynamodb_full.py

@@ -1,7 +1,11 @@
+import pytest
+
 from commons_codec.model import SQLOperation
 from commons_codec.transform.dynamodb import DynamoDBFullLoadTranslator
 
-RECORD_ALL_TYPES = {
+pytestmark = pytest.mark.dynamodb
+
+RECORD_IN = {
     "id": {"S": "5F9E-Fsadd41C-4C92-A8C1-70BF3FFB9266"},
     "data": {"M": {"temperature": {"N": "42.42"}, "humidity": {"N": "84.84"}}},
     "meta": {"M": {"timestamp": {"S": "2024-07-12T01:17:42"}, "device": {"S": "foo"}}},
@@ -37,6 +41,40 @@
     "set_of_strings": {"SS": ["location_1"]},
 }
 
+RECORD_OUT_DATA = {
+    "id": "5F9E-Fsadd41C-4C92-A8C1-70BF3FFB9266",
+    "data": {"temperature": 42.42, "humidity": 84.84},
+    "meta": {"timestamp": "2024-07-12T01:17:42", "device": "foo"},
+    "location": {
+        "address": "Berchtesgaden Salt Mine",
+        "coordinates": [
+            "",
+        ],
+        "meetingPoint": "At the end of the tunnel",
+    },
+    "list_of_objects": [
+        {
+            "date": "2024-08-28T20:05:42.603Z",
+            "utm_adgroup": ["", ""],
+            "utm_campaign": "34374686341",
+            "utm_medium": "foobar",
+            "utm_source": "google",
+        }
+    ],
+    "map_of_numbers": {"test": 1.0, "test2": 2.0},
+    "set_of_binaries": ["U3Vubnk="],
+    "set_of_numbers": [0.34, 1.0, 2.0, 3.0],
+    "set_of_strings": ["location_1"],
+}
+
+RECORD_OUT_AUX = {
+    "list_of_varied": [
+        {"a": 1.0},
+        2.0,
+        "Three",
+    ],
+}
+
 
 def test_sql_ddl():
     assert (
@@ -45,41 +83,37 @@ def test_sql_ddl():
     )
 
 
-def test_to_sql_all_types():
-    assert DynamoDBFullLoadTranslator(table_name="foo").to_sql(RECORD_ALL_TYPES) == SQLOperation(
+def test_to_sql_operation():
+    """
+    Verify outcome of `DynamoDBFullLoadTranslator.to_sql` operation.
+    """
+    assert DynamoDBFullLoadTranslator(table_name="foo").to_sql(RECORD_IN) == SQLOperation(
         statement="INSERT INTO foo (data, aux) VALUES (:typed, :untyped);",
         parameters={
-            "typed": {
-                "id": "5F9E-Fsadd41C-4C92-A8C1-70BF3FFB9266",
-                "data": {"temperature": 42.42, "humidity": 84.84},
-                "meta": {"timestamp": "2024-07-12T01:17:42", "device": "foo"},
-                "location": {
-                    "address": "Berchtesgaden Salt Mine",
-                    "coordinates": [
-                        "",
-                    ],
-                    "meetingPoint": "At the end of the tunnel",
-                },
-                "list_of_objects": [
-                    {
-                        "date": "2024-08-28T20:05:42.603Z",
-                        "utm_adgroup": ["", ""],
-                        "utm_campaign": "34374686341",
-                        "utm_medium": "foobar",
-                        "utm_source": "google",
-                    }
-                ],
-                "map_of_numbers": {"test": 1.0, "test2": 2.0},
-                "set_of_binaries": ["U3Vubnk="],
-                "set_of_numbers": [0.34, 1.0, 2.0, 3.0],
-                "set_of_strings": ["location_1"],
-            },
-            "untyped": {
-                "list_of_varied": [
-                    {"a": 1.0},
-                    2.0,
-                    "Three",
-                ],
-            },
+            "typed": RECORD_OUT_DATA,
+            "untyped": RECORD_OUT_AUX,
         },
     )
+
+
+def test_to_sql_cratedb(caplog, cratedb):
+    """
+    Verify writing converted DynamoDB record to CrateDB.
+    """
+
+    # Compute CrateDB operation (SQL+parameters) from DynamoDB record.
+    translator = DynamoDBFullLoadTranslator(table_name="from.dynamodb")
+    operation = translator.to_sql(record=RECORD_IN)
+
+    # Insert into CrateDB.
+    cratedb.database.run_sql(translator.sql_ddl)
+    cratedb.database.run_sql(operation.statement, operation.parameters)
+
+    # Verify data in target database.
+    assert cratedb.database.table_exists("from.dynamodb") is True
+    assert cratedb.database.refresh_table("from.dynamodb") is True
+    assert cratedb.database.count_records("from.dynamodb") == 1
+
+    results = cratedb.database.run_sql('SELECT * FROM "from".dynamodb;', records=True)  # noqa: S608
+    assert results[0]["data"] == RECORD_OUT_DATA
+    assert results[0]["aux"] == RECORD_OUT_AUX

tests/transform/test_dynamodb_types_cratedb.py

@@ -1,9 +1,13 @@
 import unittest
 from decimal import Decimal
 
+import pytest
+
 from commons_codec.model import DualRecord
 from commons_codec.transform.dynamodb import CrateDBTypeDeserializer, DynamoDBCDCTranslator
 
+pytestmark = pytest.mark.dynamodb
+
 
 class TestDeserializer(unittest.TestCase):
     def setUp(self):

tests/transform/test_dynamodb_types_vanilla.py

@@ -17,6 +17,8 @@
 
 from commons_codec.vendor.boto3.dynamodb.types import Binary, TypeDeserializer
 
+pytestmark = pytest.mark.dynamodb
+
 
 class TestBinary(unittest.TestCase):
     def test_bytes_input(self):

tests/transform/test_mongodb.py

@@ -1,8 +1,10 @@
 # ruff: noqa: E402, E501
-import datetime
-
 import pytest
 
+pytestmark = pytest.mark.mongodb
+
+import datetime
+
 from commons_codec.model import SQLOperation
 
 pytest.importorskip("pymongo")