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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
import pytest
from multidict import MultiDict
from grpclib.metadata import Deadline
from grpclib.metadata import encode_timeout, decode_timeout
from grpclib.metadata import encode_metadata, decode_metadata
from grpclib.metadata import encode_grpc_message, decode_grpc_message
@pytest.mark.parametrize('value, expected', [
(100, '100S'),
(15, '15S'),
(7, '7000m'),
(0.02, '20m'),
(0.001, '1000u'),
(0.00002, '20u'),
(0.000001, '1000n'),
(0.00000002, '20n'),
])
def test_encode_timeout(value, expected):
assert encode_timeout(value) == expected
@pytest.mark.parametrize('value, expected', [
('5H', 5 * 3600),
('4M', 4 * 60),
('3S', 3),
('200m', pytest.approx(0.2)),
('100u', pytest.approx(0.0001)),
('50n', pytest.approx(0.00000005)),
])
def test_decode_timeout(value, expected):
assert decode_timeout(value) == expected
def test_deadline():
assert Deadline.from_timeout(1) < Deadline.from_timeout(2)
with pytest.raises(TypeError) as err:
Deadline.from_timeout(1) < 'invalid'
err.match('comparison is not supported between instances '
'of \'Deadline\' and \'str\'')
assert Deadline(_timestamp=1) == Deadline(_timestamp=1)
assert Deadline.from_timeout(1) != 'invalid'
@pytest.mark.parametrize('value, output', [
('а^2+б^2=ц^2', '%D0%B0^2+%D0%B1^2=%D1%86^2'),
])
def test_grpc_message_encoding(value, output):
assert encode_grpc_message(value) == output
encode_grpc_message(value).encode('ascii')
assert decode_grpc_message(output) == value
def test_grpc_message_decode_safe():
# 0xFF is invalid byte in utf-8:
# https://www.cl.cam.ac.uk/~mgk25/ucs/examples/UTF-8-test.txt
assert (decode_grpc_message('%FF^2+%FF^2=%FF^2')
== '\ufffd^2+\ufffd^2=\ufffd^2')
@pytest.mark.parametrize('value, output', [
({'regular': 'value'}, [('regular', 'value')]), # dict-like
([('regular', 'value')], [('regular', 'value')]), # list of pairs
({'binary-bin': b'value'}, [('binary-bin', 'dmFsdWU')])
])
def test_encode_metadata(value, output):
assert encode_metadata(value) == output
def test_encode_metadata_errors():
with pytest.raises(TypeError) as e1:
encode_metadata({'regular': b'invalid'})
e1.match('Invalid metadata value type, str expected')
with pytest.raises(TypeError) as e2:
encode_metadata({'binary-bin': 'invalid'})
e2.match('Invalid metadata value type, bytes expected')
@pytest.mark.parametrize('key', [
'grpc-internal',
'Upper-Case',
'invalid~character',
' spaces ',
'new-line\n',
])
def test_encode_metadata_invalid_key(key):
with pytest.raises(ValueError) as err:
encode_metadata({key: 'anything'})
err.match('Invalid metadata key')
@pytest.mark.parametrize('value', [
'new-line\n',
])
def test_encode_metadata_invalid_value(value):
with pytest.raises(ValueError) as err:
encode_metadata({'foo': value})
err.match('Invalid metadata value')
def test_decode_metadata_empty():
metadata = decode_metadata([
(':method', 'POST'),
('te', 'trailers'),
('content-type', 'application/grpc'),
('user-agent', 'Test'),
('grpc-timeout', '100m'),
])
assert metadata == MultiDict()
@pytest.mark.parametrize('key, value, expected', [
('regular', 'value', 'value'),
('binary-bin', 'dmFsdWU', b'value'),
])
def test_decode_metadata_regular(key, value, expected):
metadata = decode_metadata([
(':method', 'POST'),
('te', 'trailers'),
('content-type', 'application/grpc'),
('user-agent', 'Test'),
('grpc-timeout', '100m'),
(key, value),
])
assert metadata == MultiDict({key: expected})
assert type(metadata[key]) is type(expected)
|
