Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_dumps_load_with_register_codec(client, country_schema):
payload = {"country": "Argenntina"}
country_serializer = serializer.FaustSerializer(client, "test-country", country_schema)
faust.serializers.codecs.register("country_serializer", country_serializer)
class CountryRecord(faust.Record, serializer="country_serializer"):
country: str
country_record = CountryRecord(**payload)
message_encoded = country_record.dumps()
assert message_encoded
assert len(message_encoded) > 5
assert isinstance(message_encoded, bytes)
message_decoded = CountryRecord.loads(message_encoded)
assert message_decoded == country_record
def test_nested_schema_with_register_codec(client):
nested_schema = schema.AvroSchema(data_gen.NESTED_SCHENA)
order_schema = schema.AvroSchema(data_gen.ORDER_SCHENA)
customer_serializer = serializer.FaustSerializer(client, "test-nested-schema", nested_schema)
order_serializer = serializer.FaustSerializer(client, "test-order-schema", order_schema)
faust.serializers.codecs.register("customer_serializer", customer_serializer)
faust.serializers.codecs.register("order_serializer", order_serializer)
class Order(faust.Record, serializer="order_serializer"):
uid: int
class Customer(faust.Record, serializer="customer_serializer"):
name: str
uid: int
order: Order
payload = data_gen.create_nested_schema()
customer = Customer(**payload)
message_encoded = customer.dumps()
assert message_encoded