[Schema Registry] add tests in avro serializer to check that types are not leaked

[swathipil]

background discussion: #20708.
Read through Avro API surface and add more test coverage to ensure that types are not leaked.
Check:

• Exceptions

• Serialization

• Deserialization
  If Avro specific exceptions thrown, catch and throw azure-core exceptions.

• create gist for addressing this

  • do we want to do client side validation at all, or let the service validate?

[swathipil]

• check all the following not included: "type", "name", "namespace", "fields"
• type of "name" is not "string" object
• type of "namespace" is not "string" object
• type of "fields" is not a list, not a list of dicts
  potential sources of type leak:
• https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L1202
• schema fullname:

  azure-sdk-for-python/sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_schema_registry_avro_serializer.py

  Line 150 in 2257388

  schema_id = self._get_schema_id(cached_schema.fullname, cached_schema, **kwargs)

  
  -https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L1127
• if namespace not included, avro uses default namespace: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L1128
• how to deal with non-record types? https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L1129
  __RECORD
• this? https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/name.py#L83
• InvalidName error: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/name.py#L42
• check behavior of fullname vs. fastavro: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/name.py#L58
• check for warnings? :https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L1137

____________________ FIELD ____________________

• check field is dict/mapping: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L829
• if two fields are named the same thing: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L843
• will field be added if it's in the list of reserved properties? https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L838
• set default value for field, name of val is "default": https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L834
• ALL OF FIELD IS MINEFIELD: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/schema.py#L314

[swathipil]

from fastavro import parse_schema, schemaless_writer, schemaless_reader
import avro
from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder
import json
from io import BytesIO
import time

dict_record_schema = {"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}
str_record_schema = json.dumps(dict_record_schema)
bytes_record_schema = bytes(str_record_schema, "UTF-8")

record_val = {
    "name": "hello",
    "favorite_number": 3,
    "favorite_color": "yello"
}
count = 100000
num_times = 5

avro_schema_cache = {}
avro_reader_cache = {}

def to_dict(schema):
    if isinstance(schema, str):
        return json.loads(schema)
    try:
        return json.loads(schema.decode('utf-8'))
    except json.decoder.JSONDecodeError as e:
        raise ValueError("Error parsing JSON: {}, error = {}".format(schema, e))

def serialize_fastavro(schema, value):
    # schemas passed in to fastavro must be converted to dict first
    json_schema = to_dict(schema)
    parsed_schema = parse_schema(json_schema, _write_hint=False)

    stream = BytesIO()
    with stream:
        schemaless_writer(stream, parsed_schema, value)
        encoded_data = stream.getvalue()
    return encoded_data

def serialize_avro(schema, value):
    if not isinstance(schema, avro.schema.Schema):
        schema = avro.schema.parse(schema)
    try:
        writer = avro_schema_cache[str(schema)]
    except KeyError:
        writer = DatumWriter(schema)
        avro_schema_cache[str(schema)] = writer
    
    stream = BytesIO()
    with stream:
        writer.write(value, BinaryEncoder(stream))
        encoded_data = stream.getvalue()
    return encoded_data

def deserialize_fastavro(schema, data):
    json_schema = to_dict(schema)
    parsed_schema = parse_schema(json_schema, _write_hint=False)

    stream = BytesIO(data)
    with stream:
        decoded_data = schemaless_reader(stream, parsed_schema)

def deserialize_avro(schema, data):
    if not hasattr(data, 'read'):
        data = BytesIO(data)

    if not isinstance(schema, avro.schema.Schema):
        schema = avro.schema.parse(schema)

    try:
        reader = avro_reader_cache[str(schema)]
    except KeyError:
        reader = DatumReader(writers_schema=schema)
        avro_reader_cache[str(schema)] = reader

    with data:
        bin_decoder = BinaryDecoder(data)
        decoded_data = reader.read(bin_decoder)


def serialize_100000_records_fastavro():
    # with bytes schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            serialize_fastavro(bytes_record_schema, record_val)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('serialize bytes schema fastavro: {}'.format(avg_time))

    # with str schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            serialize_fastavro(str_record_schema, record_val)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('serialize str schema fastavro: {}'.format(avg_time))

def serialize_100000_records_avro():
    # with bytes schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            serialize_avro(bytes_record_schema, record_val)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('serialize bytes schema avro: {}'.format(avg_time))

    # with str schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            serialize_avro(str_record_schema, record_val)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('serialize str schema avro: {}'.format(avg_time))

def deserialize_100000_records_fastavro(encoded_data):
    # with bytes schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            deserialize_fastavro(bytes_record_schema, encoded_data)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('deserialize bytes schema fastavro: {}'.format(avg_time))

    # with str schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            deserialize_fastavro(str_record_schema, encoded_data)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('deserialize str schema fastavro: {}'.format(total_time))

def deserialize_100000_records_avro(encoded_data):
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            deserialize_avro(bytes_record_schema, encoded_data)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('deserialize bytes schema avro: {}'.format(avg_time))

    # with str schema
    times = []
    for _ in range(num_times):
        start_time = time.time()
        for _ in range(count):
            deserialize_avro(str_record_schema, encoded_data)
        end_time = time.time()
        total_time = end_time - start_time
        times.append(total_time)
    avg_time = sum(times)/num_times
    print('deserialize str schema avro: {}'.format(avg_time))

serialize_100000_records_fastavro()
serialize_100000_records_avro()
encoded_data = serialize_avro(bytes_record_schema, record_val)
deserialize_100000_records_fastavro(encoded_data)
deserialize_100000_records_avro(encoded_data)

[swathipil]

to get same fullname in fastavro as avro: https://github.com/apache/avro/blob/c86b34d01e80e19c58f28d4104b280a3e6b891f9/lang/py/avro/name.py#L83