-
Notifications
You must be signed in to change notification settings - Fork 3
/
test_avro_serde_base.py
41 lines (36 loc) · 957 Bytes
/
test_avro_serde_base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from confluent_kafka import avro
from kafkian.serde.avroserdebase import HasSchemaMixin, _wrap
SCHEMA = avro.loads("""
{
"name": "basic",
"type": "record",
"doc": "basic schema for tests",
"namespace": "python.test.basic",
"fields": [
{
"name": "number",
"doc": "age",
"type": [
"long",
"null"
]
},
{
"name": "name",
"doc": "a name",
"type": [
"string"
]
}
]
}
""")
def test_schema_mixin_wrapper():
for base_class in (int, float, dict, list):
val = base_class()
wrapped = _wrap(val, SCHEMA)
assert val == wrapped
assert isinstance(wrapped, base_class)
assert isinstance(wrapped, HasSchemaMixin)
assert wrapped.schema() is SCHEMA
assert wrapped.__class__.__name__ == 'python.test.basic.basic'