sift/python/lib/sift_py/_internal/test_util/channel.py at 9a75eeaeeedecc9de014ca1b58cd95048092099b · sift-stack/sift · GitHub

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
133
134
135
136
137
138
139
140
from collections.abc import AsyncIterable, Iterable
from typing import Any, Callable, Optional, Union

import grpc
import grpc.aio as grpc_aio
from grpc.aio import Channel as AsyncChannel
from grpc_testing import Channel

from sift_py.grpc.transport import SiftChannelConfig

SerializingFunction = Callable[[Any], bytes]
DeserializingFunction = Callable[[bytes], Any]
DoneCallbackType = Callable[[Any], None]
RequestIterableType = Union[Iterable, AsyncIterable]
ResponseIterableType = AsyncIterable


class MockChannel(Channel):
    """
    Used as a mock gRPC channel
    """

    config = SiftChannelConfig(uri="localhost:50051", apikey="fake-api-key", use_ssl=False)

    def take_unary_unary(self, method_descriptor):
        pass

    def take_unary_stream(self, method_descriptor):
        pass

    def take_stream_unary(self, method_descriptor):
        pass

    def take_stream_stream(self, method_descriptor):
        pass

    def subscribe(self, callback, try_to_connect=False):
        pass

    def unsubscribe(self, callback):
        pass

    def unary_unary(
        self,
        method,
        request_serializer=None,
        response_deserializer=None,
        _registered_method=False,
    ):
        pass

    def unary_stream(
        self,
        method,
        request_serializer=None,
        response_deserializer=None,
        _registered_method=False,
    ):
        pass

    def stream_unary(
        self,
        method,
        request_serializer=None,
        response_deserializer=None,
        _registered_method=False,
    ):
        pass

    def stream_stream(
        self,
        method,
        request_serializer=None,
        response_deserializer=None,
        _registered_method=False,
    ):
        pass

    def close(self):
        pass

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


class MockAsyncChannel(AsyncChannel):
    async def __aenter__(self):
        pass

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass

    async def close(self, grace: Optional[float] = None):
        pass

    def get_state(self, try_to_connect: bool = False) -> grpc.ChannelConnectivity: ...

    async def wait_for_state_change(
        self,
        last_observed_state: grpc.ChannelConnectivity,
    ) -> None:
        return None

    async def channel_ready(self) -> None:
        return None

    def unary_unary(
        self,
        method: str,
        request_serializer: Optional[SerializingFunction] = None,
        response_deserializer: Optional[DeserializingFunction] = None,
        _registered_method: Optional[bool] = False,
    ) -> grpc_aio.UnaryUnaryMultiCallable: ...

    def unary_stream(
        self,
        method: str,
        request_serializer: Optional[SerializingFunction] = None,
        response_deserializer: Optional[DeserializingFunction] = None,
        _registered_method: Optional[bool] = False,
    ) -> grpc_aio.UnaryStreamMultiCallable: ...

    def stream_unary(
        self,
        method: str,
        request_serializer: Optional[SerializingFunction] = None,
        response_deserializer: Optional[DeserializingFunction] = None,
        _registered_method: Optional[bool] = False,
    ) -> grpc_aio.StreamUnaryMultiCallable: ...

    def stream_stream(
        self,
        method: str,
        request_serializer: Optional[SerializingFunction] = None,
        response_deserializer: Optional[DeserializingFunction] = None,
        _registered_method: Optional[bool] = False,
    ) -> grpc_aio.StreamStreamMultiCallable: ...