Add JSON Schema validation test; fix typo (#2003)

* Add JSON Schema validation test; fix typo

* Fix test

* Fix flake8

Author: rayokota

src/confluent_kafka/schema_registry/_async/json_schema.py

@@ -272,7 +272,7 @@ async def __init_impl(
             raise ValueError("schema.id.serializer must be callable")
 
         self._validate = conf_copy.pop('validate')
-        if not isinstance(self._normalize_schemas, bool):
+        if not isinstance(self._validate, bool):
             raise ValueError("validate must be a boolean value")
 
         if len(conf_copy) > 0:

src/confluent_kafka/schema_registry/_sync/json_schema.py

@@ -272,7 +272,7 @@ def __init_impl(
             raise ValueError("schema.id.serializer must be callable")
 
         self._validate = conf_copy.pop('validate')
-        if not isinstance(self._normalize_schemas, bool):
+        if not isinstance(self._validate, bool):
             raise ValueError("validate must be a boolean value")
 
         if len(conf_copy) > 0:

tests/schema_registry/_async/test_json_serdes.py

@@ -87,7 +87,7 @@ async def run_before_and_after_tests(tmpdir):
 async def test_json_basic_serialization():
     conf = {'url': _BASE_URL}
     client = AsyncSchemaRegistryClient.new_client(conf)
-    ser_conf = {'auto.register.schemas': True}
+    ser_conf = {'auto.register.schemas': True, 'validate': True}
     obj = {
         'intField': 123,
         'doubleField': 45.67,
@@ -121,6 +121,40 @@ async def test_json_basic_serialization():
     assert obj == obj2
 
 
+async def test_json_basic_failing_validation():
+    conf = {'url': _BASE_URL}
+    client = AsyncSchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': True, 'validate': True}
+    obj = {
+        'intField': '123',
+        'doubleField': 45.67,
+        'stringField': 'hi',
+        'booleanField': True,
+        'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+    }
+    schema = {
+        "type": "object",
+        "properties": {
+            "intField": {"type": "integer"},
+            "doubleField": {"type": "number"},
+            "stringField": {
+                "type": "string",
+                "confluent:tags": ["PII"]
+            },
+            "booleanField": {"type": "boolean"},
+            "bytesField": {
+                "type": "string",
+                "contentEncoding": "base64",
+                "confluent:tags": ["PII"]
+            }
+        }
+    }
+    ser = await AsyncJSONSerializer(json.dumps(schema), client, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    with pytest.raises(SerializationError):
+        await ser(obj, ser_ctx)
+
+
 async def test_json_guid_in_header():
     conf = {'url': _BASE_URL}
     client = AsyncSchemaRegistryClient.new_client(conf)

tests/schema_registry/_sync/test_json_serdes.py

@@ -87,7 +87,7 @@ def run_before_and_after_tests(tmpdir):
 def test_json_basic_serialization():
     conf = {'url': _BASE_URL}
     client = SchemaRegistryClient.new_client(conf)
-    ser_conf = {'auto.register.schemas': True}
+    ser_conf = {'auto.register.schemas': True, 'validate': True}
     obj = {
         'intField': 123,
         'doubleField': 45.67,
@@ -121,6 +121,40 @@ def test_json_basic_serialization():
     assert obj == obj2
 
 
+def test_json_basic_failing_validation():
+    conf = {'url': _BASE_URL}
+    client = SchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': True, 'validate': True}
+    obj = {
+        'intField': '123',
+        'doubleField': 45.67,
+        'stringField': 'hi',
+        'booleanField': True,
+        'bytesField': base64.b64encode(b'foobar').decode('utf-8'),
+    }
+    schema = {
+        "type": "object",
+        "properties": {
+            "intField": {"type": "integer"},
+            "doubleField": {"type": "number"},
+            "stringField": {
+                "type": "string",
+                "confluent:tags": ["PII"]
+            },
+            "booleanField": {"type": "boolean"},
+            "bytesField": {
+                "type": "string",
+                "contentEncoding": "base64",
+                "confluent:tags": ["PII"]
+            }
+        }
+    }
+    ser = JSONSerializer(json.dumps(schema), client, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    with pytest.raises(SerializationError):
+        ser(obj, ser_ctx)
+
+
 def test_json_guid_in_header():
     conf = {'url': _BASE_URL}
     client = SchemaRegistryClient.new_client(conf)