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
|
import pytest
from google.protobuf.empty_pb2 import Empty
from grpclib.const import Handler, Cardinality, Status
from grpclib.client import UnaryUnaryMethod, StreamUnaryMethod
from grpclib.client import StreamStreamMethod
from grpclib.testing import ChannelFor
from grpclib.exceptions import GRPCError
@pytest.mark.asyncio
@pytest.mark.parametrize('cardinality, method_cls, method_arg', [
(Cardinality.UNARY_UNARY, UnaryUnaryMethod, Empty()),
(Cardinality.STREAM_UNARY, StreamUnaryMethod, [Empty()]),
])
async def test_unary_reply_error(cardinality, method_cls, method_arg):
class Service:
async def Case(self, stream):
await stream.send_initial_metadata()
await stream.send_trailing_metadata(status=Status.PERMISSION_DENIED)
def __mapping__(self):
return {
'/test.Test/Case': Handler(
self.Case,
cardinality,
Empty,
Empty,
)
}
async with ChannelFor([Service()]) as channel:
method = method_cls(channel, '/test.Test/Case', Empty, Empty)
with pytest.raises(GRPCError) as err:
await method(method_arg)
assert err.value.status is Status.PERMISSION_DENIED
@pytest.mark.asyncio
@pytest.mark.parametrize('cardinality, method_cls, method_res', [
(Cardinality.STREAM_UNARY, StreamUnaryMethod, Empty()),
(Cardinality.STREAM_STREAM, StreamStreamMethod, [Empty()]),
])
async def test_empty_streaming_request(cardinality, method_cls, method_res):
class Service:
async def Case(self, stream):
await stream.send_message(Empty())
def __mapping__(self):
return {
'/test.Test/Case': Handler(
self.Case,
cardinality,
Empty,
Empty,
)
}
async with ChannelFor([Service()]) as channel:
method = method_cls(channel, '/test.Test/Case', Empty, Empty)
assert await method([]) == method_res
|
