Schema Registry 7: Cleanup and additional tests (#476)

* Replace utcnow() with the modern now(utc) pattern

Gets rid of annoying warnings

* Silence warnings coming from confluent_kafka.schema_registry

We have no control there, so we don't need to see those.

* Rename deserializers test

"no column name" seem to be some artifact from the past.

* Catch TypeError during Avro serialization

* Test nested Avro schemas

Author: gwaramadze

pyproject.toml

@@ -63,9 +63,11 @@ line_length = 88
 
 [tool.pytest.ini_options]
 minversion = "6.0"
+filterwarnings = [
+  "ignore::Warning:confluent_kafka.schema_registry.*",
+]
 # Ignore manual tests by and some loggers by default
 addopts = "--log-disable=urllib3.connectionpool --log-disable=parso --log-disable=docker --log-disable=asyncio"
-
 # Print debug logs to the console in tests
 log_cli = true
 log_cli_level = "INFO"

quixstreams/models/serializers/avro.py

@@ -102,7 +102,7 @@ def __call__(self, value: Any, ctx: SerializationContext) -> bytes:
                     strict_allow_default=self._strict_allow_default,
                     disable_tuple_notation=self._disable_tuple_notation,
                 )
-            except ValueError as exc:
+            except (ValueError, TypeError) as exc:
                 raise SerializationError(str(exc)) from exc
 
             return data.getvalue()

tests/containerhelper.py

@@ -1,8 +1,8 @@
 import base64
-import datetime
 import random
 import time
 import uuid
+from datetime import datetime, timedelta, timezone
 from typing import Tuple
 
 from testcontainers.core.container import DockerContainer
@@ -95,9 +95,9 @@ def start_schema_registry_container(
 
 
 def wait_for_container_readiness(container: DockerContainer, text: str) -> None:
-    start = datetime.datetime.utcnow()
-    cut_off = start + datetime.timedelta(seconds=20)
-    while cut_off > datetime.datetime.utcnow():
+    start = datetime.now(timezone.utc)
+    cut_off = start + timedelta(seconds=20)
+    while cut_off > datetime.now(timezone.utc):
         time.sleep(0.5)
         logs = container.get_logs()
         for line in logs:

tests/test_quixstreams/test_models/test_serializers/constants.py

@@ -4,10 +4,21 @@
 
 AVRO_TEST_SCHEMA = {
     "type": "record",
-    "name": "testschema",
+    "name": "Root",
     "fields": [
         {"name": "name", "type": "string"},
         {"name": "id", "type": "int", "default": 0},
+        {
+            "name": "nested",
+            "type": {
+                "type": "record",
+                "name": "Nested",
+                "fields": [
+                    {"name": "id", "type": "int", "default": 0},
+                ],
+            },
+            "default": {"id": 0},
+        },
     ],
 }
 

tests/test_quixstreams/test_models/test_serializers/test_schema_registry.py

@@ -103,22 +103,29 @@ def _inject_schema_registry(
             partial(AvroSerializer, AVRO_TEST_SCHEMA),
             partial(AvroDeserializer),
             {"name": "foo", "id": 123},
-            b"\x06foo\xf6\x01",
-            {"name": "foo", "id": 123},
+            b"\x06foo\xf6\x01\x00",
+            {"name": "foo", "id": 123, "nested": {"id": 0}},
         ),
         (
             partial(AvroSerializer, AVRO_TEST_SCHEMA),
             partial(AvroDeserializer),
             {"name": "foo", "id": 0},
-            b"\x06foo\x00",
-            {"name": "foo", "id": 0},
+            b"\x06foo\x00\x00",
+            {"name": "foo", "id": 0, "nested": {"id": 0}},
+        ),
+        (
+            partial(AvroSerializer, AVRO_TEST_SCHEMA),
+            partial(AvroDeserializer),
+            {"name": "foo", "id": 123, "nested": {"id": 123}},
+            b"\x06foo\xf6\x01\xf6\x01",
+            {"name": "foo", "id": 123, "nested": {"id": 123}},
         ),
         (
             partial(AvroSerializer, AVRO_TEST_SCHEMA),
             partial(AvroDeserializer, schema=AVRO_TEST_SCHEMA),
-            {"name": "foo", "id": 123},
-            b"\x06foo\xf6\x01",
-            {"name": "foo", "id": 123},
+            {"name": "foo", "id": 123, "nested": {"id": 123}},
+            b"\x06foo\xf6\x01\xf6\x01",
+            {"name": "foo", "id": 123, "nested": {"id": 123}},
         ),
         (
             partial(JSONSerializer, schema=JSONSCHEMA_TEST_SCHEMA),

tests/test_quixstreams/test_models/test_serializers/test_serializers.py

@@ -58,9 +58,19 @@ class TestSerializers:
             (
                 AvroSerializer(AVRO_TEST_SCHEMA),
                 {"name": "foo", "id": 123},
-                b"\x06foo\xf6\x01",
+                b"\x06foo\xf6\x01\x00",
+            ),
+            (AvroSerializer(AVRO_TEST_SCHEMA), {"name": "foo"}, b"\x06foo\x00\x00"),
+            (
+                AvroSerializer(AVRO_TEST_SCHEMA),
+                {"name": "foo", "nested": {}},
+                b"\x06foo\x00\x00",
+            ),
+            (
+                AvroSerializer(AVRO_TEST_SCHEMA),
+                {"name": "foo", "nested": {"id": 123}},
+                b"\x06foo\x00\xf6\x01",
             ),
-            (AvroSerializer(AVRO_TEST_SCHEMA), {"name": "foo"}, b"\x06foo\x00"),
             (ProtobufSerializer(Root), {}, b""),
             (ProtobufSerializer(Root), {"id": 3}, b"\x10\x03"),
             (ProtobufSerializer(Root), {"name": "foo", "id": 2}, b"\n\x03foo\x10\x02"),
@@ -127,6 +137,11 @@ def test_serialize_success(self, serializer: Serializer, value, expected):
             ),
             (AvroSerializer(AVRO_TEST_SCHEMA), {"foo": "foo", "id": 123}),
             (AvroSerializer(AVRO_TEST_SCHEMA), {"id": 123}),
+            (AvroSerializer(AVRO_TEST_SCHEMA), {"name": "foo", "id": "string"}),
+            (
+                AvroSerializer(AVRO_TEST_SCHEMA),
+                {"name": "foo", "nested": {"id": "string"}},
+            ),
             (AvroSerializer(AVRO_TEST_SCHEMA, strict=True), {"name": "foo"}),
             (ProtobufSerializer(Root), {"bar": 3}),
             (ProtobufSerializer(Root), Nested()),
@@ -170,13 +185,18 @@ class TestDeserializers:
             ),
             (
                 AvroDeserializer(AVRO_TEST_SCHEMA),
-                b"\x06foo\xf6\x01",
-                {"name": "foo", "id": 123},
+                b"\x06foo\xf6\x01\x00",
+                {"name": "foo", "id": 123, "nested": {"id": 0}},
+            ),
+            (
+                AvroDeserializer(AVRO_TEST_SCHEMA),
+                b"\x06foo\x00\x00",
+                {"name": "foo", "id": 0, "nested": {"id": 0}},
             ),
             (
                 AvroDeserializer(AVRO_TEST_SCHEMA),
-                b"\x06foo\x00",
-                {"name": "foo", "id": 0},
+                b"\x06foo\x00\xf6\x01",
+                {"name": "foo", "id": 0, "nested": {"id": 123}},
             ),
             (
                 ProtobufDeserializer(Root),
@@ -234,9 +254,7 @@ class TestDeserializers:
             ),
         ],
     )
-    def test_deserialize_no_column_name_success(
-        self, deserializer: Deserializer, value, expected
-    ):
+    def test_deserialize_success(self, deserializer: Deserializer, value, expected):
         assert deserializer(value, ctx=DUMMY_CONTEXT) == expected
 
     @pytest.mark.parametrize(

tests/test_quixstreams/test_state/test_rocksdb/test_transaction.py

@@ -1,6 +1,6 @@
 import contextlib
 import secrets
-from datetime import datetime
+from datetime import datetime, timezone
 from unittest.mock import patch
 
 import pytest
@@ -182,8 +182,8 @@ def test_key_exists_deleted_in_cache(self, rocksdb_partition):
         [
             ("string", object()),
             (object(), "string"),
-            ("string", datetime.utcnow()),
-            (datetime.utcnow(), "string"),
+            ("string", datetime.now(timezone.utc)),
+            (datetime.now(timezone.utc), "string"),
         ],
     )
     def test_set_serialization_error(self, key, value, rocksdb_partition):
@@ -192,7 +192,9 @@ def test_set_serialization_error(self, key, value, rocksdb_partition):
             with pytest.raises(StateSerializationError):
                 tx.set(key, value, prefix=prefix)
 
-    @pytest.mark.parametrize("key", [object(), b"somebytes", datetime.utcnow()])
+    @pytest.mark.parametrize(
+        "key", [object(), b"somebytes", datetime.now(timezone.utc)]
+    )
     def test_delete_serialization_error(self, key, rocksdb_partition):
         prefix = b"__key__"
         with rocksdb_partition.begin() as tx:
@@ -330,7 +332,7 @@ def test_set_dict_nonstr_keys_fails(self, rocksdb_partition):
 
     def test_set_datetime_fails(self, rocksdb_partition):
         key = "key"
-        value = datetime.utcnow()
+        value = datetime.now(timezone.utc)
         prefix = b"__key__"
         with rocksdb_partition.begin() as tx:
             with pytest.raises(StateSerializationError):