File: test_rest_streaming_async.py

package info (click to toggle)
python-google-api-core 2.23.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,272 kB
  • sloc: python: 14,967; makefile: 3
file content (378 lines) | stat: -rw-r--r-- 12,038 bytes parent folder | download | duplicates (2)
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# TODO: set random.seed explicitly in each test function.
# See related issue: https://github.com/googleapis/python-api-core/issues/689.

import datetime
import logging
import random
import time
from typing import List, AsyncIterator

try:
    from unittest import mock
    from unittest.mock import AsyncMock  # pragma: NO COVER  # noqa: F401
except ImportError:  # pragma: NO COVER
    import mock  # type: ignore

import pytest  # noqa: I202

import proto

try:
    from google.auth.aio.transport import Response
except ImportError:
    pytest.skip(
        "google-api-core[async_rest] is required to test asynchronous rest streaming.",
        allow_module_level=True,
    )

from google.api_core import rest_streaming_async
from google.api import http_pb2
from google.api import httpbody_pb2


from ..helpers import Composer, Song, EchoResponse, parse_responses


__protobuf__ = proto.module(package=__name__)
SEED = int(time.time())
logging.info(f"Starting async rest streaming tests with random seed: {SEED}")
random.seed(SEED)


async def mock_async_gen(data, chunk_size=1):
    for i in range(0, len(data)):  # pragma: NO COVER
        chunk = data[i : i + chunk_size]
        yield chunk.encode("utf-8")


class ResponseMock(Response):
    class _ResponseItr(AsyncIterator[bytes]):
        def __init__(self, _response_bytes: bytes, random_split=False):
            self._responses_bytes = _response_bytes
            self._idx = 0
            self._random_split = random_split

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._idx >= len(self._responses_bytes):
                raise StopAsyncIteration
            if self._random_split:
                n = random.randint(1, len(self._responses_bytes[self._idx :]))
            else:
                n = 1
            x = self._responses_bytes[self._idx : self._idx + n]
            self._idx += n
            return x

    def __init__(
        self,
        responses: List[proto.Message],
        response_cls,
        random_split=False,
    ):
        self._responses = responses
        self._random_split = random_split
        self._response_message_cls = response_cls

    def _parse_responses(self):
        return parse_responses(self._response_message_cls, self._responses)

    @property
    async def headers(self):
        raise NotImplementedError()

    @property
    async def status_code(self):
        raise NotImplementedError()

    async def close(self):
        raise NotImplementedError()

    async def content(self, chunk_size=None):
        itr = self._ResponseItr(
            self._parse_responses(), random_split=self._random_split
        )
        async for chunk in itr:
            yield chunk

    async def read(self):
        raise NotImplementedError()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "random_split,resp_message_is_proto_plus",
    [(False, True), (False, False)],
)
async def test_next_simple(random_split, resp_message_is_proto_plus):
    if resp_message_is_proto_plus:
        response_type = EchoResponse
        responses = [EchoResponse(content="hello world"), EchoResponse(content="yes")]
    else:
        response_type = httpbody_pb2.HttpBody
        responses = [
            httpbody_pb2.HttpBody(content_type="hello world"),
            httpbody_pb2.HttpBody(content_type="yes"),
        ]

    resp = ResponseMock(
        responses=responses, random_split=random_split, response_cls=response_type
    )
    itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
    idx = 0
    async for response in itr:
        assert response == responses[idx]
        idx += 1


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "random_split,resp_message_is_proto_plus",
    [
        (True, True),
        (False, True),
        (True, False),
        (False, False),
    ],
)
async def test_next_nested(random_split, resp_message_is_proto_plus):
    if resp_message_is_proto_plus:
        response_type = Song
        responses = [
            Song(title="some song", composer=Composer(given_name="some name")),
            Song(title="another song", date_added=datetime.datetime(2021, 12, 17)),
        ]
    else:
        # Although `http_pb2.HttpRule`` is used in the response, any response message
        # can be used which meets this criteria for the test of having a nested field.
        response_type = http_pb2.HttpRule
        responses = [
            http_pb2.HttpRule(
                selector="some selector",
                custom=http_pb2.CustomHttpPattern(kind="some kind"),
            ),
            http_pb2.HttpRule(
                selector="another selector",
                custom=http_pb2.CustomHttpPattern(path="some path"),
            ),
        ]
    resp = ResponseMock(
        responses=responses, random_split=random_split, response_cls=response_type
    )
    itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
    idx = 0
    async for response in itr:
        assert response == responses[idx]
        idx += 1
    assert idx == len(responses)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "random_split,resp_message_is_proto_plus",
    [
        (True, True),
        (False, True),
        (True, False),
        (False, False),
    ],
)
async def test_next_stress(random_split, resp_message_is_proto_plus):
    n = 50
    if resp_message_is_proto_plus:
        response_type = Song
        responses = [
            Song(title="title_%d" % i, composer=Composer(given_name="name_%d" % i))
            for i in range(n)
        ]
    else:
        response_type = http_pb2.HttpRule
        responses = [
            http_pb2.HttpRule(
                selector="selector_%d" % i,
                custom=http_pb2.CustomHttpPattern(path="path_%d" % i),
            )
            for i in range(n)
        ]
    resp = ResponseMock(
        responses=responses, random_split=random_split, response_cls=response_type
    )
    itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
    idx = 0
    async for response in itr:
        assert response == responses[idx]
        idx += 1
    assert idx == n


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "random_split,resp_message_is_proto_plus",
    [
        (True, True),
        (False, True),
        (True, False),
        (False, False),
    ],
)
async def test_next_escaped_characters_in_string(
    random_split, resp_message_is_proto_plus
):
    if resp_message_is_proto_plus:
        response_type = Song
        composer_with_relateds = Composer()
        relateds = ["Artist A", "Artist B"]
        composer_with_relateds.relateds = relateds

        responses = [
            Song(
                title='ti"tle\nfoo\tbar{}', composer=Composer(given_name="name\n\n\n")
            ),
            Song(
                title='{"this is weird": "totally"}',
                composer=Composer(given_name="\\{}\\"),
            ),
            Song(title='\\{"key": ["value",]}\\', composer=composer_with_relateds),
        ]
    else:
        response_type = http_pb2.Http
        responses = [
            http_pb2.Http(
                rules=[
                    http_pb2.HttpRule(
                        selector='ti"tle\nfoo\tbar{}',
                        custom=http_pb2.CustomHttpPattern(kind="name\n\n\n"),
                    )
                ]
            ),
            http_pb2.Http(
                rules=[
                    http_pb2.HttpRule(
                        selector='{"this is weird": "totally"}',
                        custom=http_pb2.CustomHttpPattern(kind="\\{}\\"),
                    )
                ]
            ),
            http_pb2.Http(
                rules=[
                    http_pb2.HttpRule(
                        selector='\\{"key": ["value",]}\\',
                        custom=http_pb2.CustomHttpPattern(kind="\\{}\\"),
                    )
                ]
            ),
        ]
    resp = ResponseMock(
        responses=responses, random_split=random_split, response_cls=response_type
    )
    itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
    idx = 0
    async for response in itr:
        assert response == responses[idx]
        idx += 1
    assert idx == len(responses)


@pytest.mark.asyncio
@pytest.mark.parametrize("response_type", [EchoResponse, httpbody_pb2.HttpBody])
async def test_next_not_array(response_type):

    data = '{"hello": 0}'
    with mock.patch.object(
        ResponseMock, "content", return_value=mock_async_gen(data)
    ) as mock_method:
        resp = ResponseMock(responses=[], response_cls=response_type)
        itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
        with pytest.raises(ValueError):
            await itr.__anext__()
        mock_method.assert_called_once()


@pytest.mark.asyncio
@pytest.mark.parametrize("response_type", [EchoResponse, httpbody_pb2.HttpBody])
async def test_cancel(response_type):
    with mock.patch.object(
        ResponseMock, "close", new_callable=mock.AsyncMock
    ) as mock_method:
        resp = ResponseMock(responses=[], response_cls=response_type)
        itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
        await itr.cancel()
        mock_method.assert_called_once()


@pytest.mark.asyncio
@pytest.mark.parametrize("response_type", [EchoResponse, httpbody_pb2.HttpBody])
async def test_iterator_as_context_manager(response_type):
    with mock.patch.object(
        ResponseMock, "close", new_callable=mock.AsyncMock
    ) as mock_method:
        resp = ResponseMock(responses=[], response_cls=response_type)
        async with rest_streaming_async.AsyncResponseIterator(resp, response_type):
            pass
        mock_method.assert_called_once()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "response_type,return_value",
    [
        (EchoResponse, bytes('[{"content": "hello"}, {', "utf-8")),
        (httpbody_pb2.HttpBody, bytes('[{"content_type": "hello"}, {', "utf-8")),
    ],
)
async def test_check_buffer(response_type, return_value):
    with mock.patch.object(
        ResponseMock,
        "_parse_responses",
        return_value=return_value,
    ):
        resp = ResponseMock(responses=[], response_cls=response_type)
        itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
        with pytest.raises(ValueError):
            await itr.__anext__()
            await itr.__anext__()


@pytest.mark.asyncio
@pytest.mark.parametrize("response_type", [EchoResponse, httpbody_pb2.HttpBody])
async def test_next_html(response_type):

    data = "<!DOCTYPE html><html></html>"
    with mock.patch.object(
        ResponseMock, "content", return_value=mock_async_gen(data)
    ) as mock_method:
        resp = ResponseMock(responses=[], response_cls=response_type)

        itr = rest_streaming_async.AsyncResponseIterator(resp, response_type)
        with pytest.raises(ValueError):
            await itr.__anext__()
        mock_method.assert_called_once()


@pytest.mark.asyncio
async def test_invalid_response_class():
    class SomeClass:
        pass

    resp = ResponseMock(responses=[], response_cls=SomeClass)
    with pytest.raises(
        ValueError,
        match="Response message class must be a subclass of proto.Message or google.protobuf.message.Message",
    ):
        rest_streaming_async.AsyncResponseIterator(resp, SomeClass)