Add tests for JSON with def and avro union with ref (#185)

Author: rayokota

schemaregistry/serde/avro.ts

@@ -83,13 +83,13 @@ export class AvroSerializer extends Serializer implements AvroSerde {
       }
     }
     const [id, info] = await this.getId(topic, msg, schema)
-    let avroSchema: avro.Type
+    let avroType: avro.Type
     let deps: Map<string, string>
-    [avroSchema, deps] = await this.toType(info)
+    [avroType, deps] = await this.toType(info)
     const subject = this.subjectName(topic, info)
     msg = await this.executeRules(
       subject, topic, RuleMode.WRITE, null, info, msg, getInlineTags(info, deps))
-    const msgBytes = avroSchema.toBuffer(msg)
+    const msgBytes = avroType.toBuffer(msg)
     return this.writeBytes(id, msgBytes)
   }
 

schemaregistry/test/serde/avro.spec.ts

@@ -366,7 +366,7 @@ describe('AvroSerializer', () => {
       schemaType: 'AVRO',
       schema: demoSchema,
     }
-    await client.register('demo-value', info , false)
+    await client.register('demo-value', info, false)
 
     info = {
       schemaType: 'AVRO',
@@ -377,7 +377,7 @@ describe('AvroSerializer', () => {
         version: 1
       }]
     }
-    await client.register(subject, info , false)
+    await client.register(subject, info, false)
 
     let nested = {
       intField: 123,
@@ -454,6 +454,111 @@ describe('AvroSerializer', () => {
     const obj2 = await deser.deserialize(topic, bytes);
     expect(obj2.color).toEqual(obj.color);
   })
+  it('serialize reference', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let ser = new AvroSerializer(client, SerdeType.VALUE, {useLatestVersion: true})
+
+    let info: SchemaInfo = {
+      schemaType: 'AVRO',
+      schema: demoSchema,
+    }
+    await client.register('demo-value', info, false)
+
+    info = {
+      schemaType: 'AVRO',
+      schema: rootPointerSchema,
+      references: [{
+        name: 'DemoSchema',
+        subject: 'demo-value',
+        version: 1
+      }]
+    }
+    await client.register(subject, info, false)
+
+    let nested = {
+      intField: 123,
+      doubleField: 45.67,
+      stringField: 'hi',
+      boolField: true,
+      bytesField: Buffer.from([1, 2]),
+    }
+    let obj = {
+      otherField: nested
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    let deser = new AvroDeserializer(client, SerdeType.VALUE, {})
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.otherField.intField).toEqual(nested.intField);
+    expect(obj2.otherField.doubleField).toBeCloseTo(nested.doubleField, 0.001);
+    expect(obj2.otherField.stringField).toEqual(nested.stringField);
+    expect(obj2.otherField.boolField).toEqual(nested.boolField);
+    expect(obj2.otherField.bytesField).toEqual(nested.bytesField);
+  })
+  it('serialize union with references', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let ser = new AvroSerializer(client, SerdeType.VALUE, {useLatestVersion: true})
+
+    let info: SchemaInfo = {
+      schemaType: 'AVRO',
+      schema: demoSchema,
+    }
+    await client.register('demo-value', info, false)
+
+    info = {
+      schemaType: 'AVRO',
+      schema: complexSchema,
+    }
+    await client.register('complex-value', info, false)
+
+    info = {
+      schemaType: 'AVRO',
+      schema: '[ "DemoSchema", "ComplexSchema" ]',
+      references: [
+        {
+          name: 'DemoSchema',
+          subject: 'demo-value',
+          version: 1
+        },
+        {
+          name: 'ComplexSchema',
+          subject: 'complex-value',
+          version: 1
+        }]
+    }
+    await client.register(subject, info, false)
+
+    let obj = {
+      intField: 123,
+      doubleField: 45.67,
+      stringField: 'hi',
+      boolField: true,
+      bytesField: Buffer.from([1, 2]),
+    }
+    // need to wrap union
+    let union = { DemoSchema: obj }
+    let bytes = await ser.serialize(topic, union)
+
+    let deser = new AvroDeserializer(client, SerdeType.VALUE, {})
+    let union2 = await deser.deserialize(topic, bytes)
+
+    // need to unwrap union
+    let obj2 = union2.DemoSchema
+
+    expect(obj2.intField).toEqual(obj.intField);
+    expect(obj2.doubleField).toBeCloseTo(obj.doubleField, 0.001);
+    expect(obj2.stringField).toEqual(obj.stringField);
+    expect(obj2.boolField).toEqual(obj.boolField);
+    expect(obj2.bytesField).toEqual(obj.bytesField);
+  })
   it('schema evolution', async () => {
     let conf: ClientConfig = {
       baseURLs: [baseURL],

schemaregistry/test/serde/json.spec.ts

@@ -145,6 +145,38 @@ const complexSchema = `
   }
 }
 `
+const defSchema = `
+{
+	"$schema" : "http://json-schema.org/draft-07/schema#",
+	"additionalProperties" : false,
+	"definitions" : {
+		"Address" : {
+			"additionalProperties" : false,
+			"properties" : {
+				"doornumber" : {
+					"type" : "integer"
+				},
+				"doorpin" : {
+					"confluent:tags" : [ "PII" ],
+					"type" : "string"
+				}
+			},
+			"type" : "object"
+		}
+	},
+	"properties" : {
+		"address" : {
+			"$ref" : "#/definitions/Address"
+		},
+		"name" : {
+			"confluent:tags" : [ "PII" ],
+			"type" : "string"
+		}
+	},
+	"title" : "Sample Event",
+	"type" : "object"
+}
+`
 
 describe('JsonSerializer', () => {
   afterEach(async () => {
@@ -501,6 +533,77 @@ describe('JsonSerializer', () => {
     obj2 = await deser.deserialize(topic, bytes)
     expect(obj2).not.toEqual(obj);
   })
+  it('encryption with def', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let serConfig: JsonSerializerConfig = {
+      useLatestVersion: true,
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let ser = new JsonSerializer(client, SerdeType.VALUE, serConfig)
+    let dekClient = fieldEncryptionExecutor.client!
+
+    let encRule: Rule = {
+      name: 'test-encrypt',
+      kind: 'TRANSFORM',
+      mode: RuleMode.WRITEREAD,
+      type: 'ENCRYPT',
+      tags: ['PII'],
+      params: {
+        'encrypt.kek.name': 'kek1',
+        'encrypt.kms.type': 'local-kms',
+        'encrypt.kms.key.id': 'mykey',
+      },
+      onFailure: 'ERROR,NONE'
+    }
+    let ruleSet: RuleSet = {
+      domainRules: [encRule]
+    }
+
+    let info: SchemaInfo = {
+      schemaType: 'JSON',
+      schema: defSchema,
+      ruleSet
+    }
+
+    await client.register(subject, info, false)
+
+    let addr = {
+      doornumber: 123,
+      doorpin: 'hi'
+    }
+    let obj = {
+      address: addr,
+      name: 'bob'
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    // reset encrypted field
+    obj.name = 'bob'
+    obj.address.doorpin = 'hi'
+
+    let deserConfig: JsonDeserializerConfig = {
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let deser = new JsonDeserializer(client, SerdeType.VALUE, deserConfig)
+    fieldEncryptionExecutor.client = dekClient
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2).toEqual(obj)
+
+    clearKmsClients()
+    let registry = new RuleRegistry()
+    registry.registerExecutor(new FieldEncryptionExecutor())
+    deser = new JsonDeserializer(client, SerdeType.VALUE, {}, registry)
+    obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2).not.toEqual(obj);
+  })
   it('encryption with union', async () => {
     let conf: ClientConfig = {
       baseURLs: [baseURL],