File: test_web_request.py

package info (click to toggle)
python-aiohttp 3.5.1-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 5,612 kB
  • sloc: python: 36,917; ansic: 15,734; makefile: 365; sh: 83
file content (665 lines) | stat: -rw-r--r-- 21,568 bytes parent folder | download
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
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
import asyncio
import socket
from collections.abc import MutableMapping
from unittest import mock

import pytest
from multidict import CIMultiDict, MultiDict
from yarl import URL

from aiohttp import HttpVersion
from aiohttp.helpers import DEBUG
from aiohttp.streams import StreamReader
from aiohttp.test_utils import make_mocked_request
from aiohttp.web import HTTPRequestEntityTooLarge


@pytest.fixture
def protocol():
    return mock.Mock(_reading_paused=False)


def test_ctor() -> None:
    req = make_mocked_request('GET', '/path/to?a=1&b=2')

    assert 'GET' == req.method
    assert HttpVersion(1, 1) == req.version
    assert req.host == socket.getfqdn()
    assert '/path/to?a=1&b=2' == req.path_qs
    assert '/path/to' == req.path
    assert 'a=1&b=2' == req.query_string
    assert CIMultiDict() == req.headers
    assert () == req.raw_headers

    get = req.query
    assert MultiDict([('a', '1'), ('b', '2')]) == get
    # second call should return the same object
    assert get is req.query

    assert req.keep_alive

    # just make sure that all lines of make_mocked_request covered
    headers = CIMultiDict(FOO='bar')
    payload = mock.Mock()
    protocol = mock.Mock()
    app = mock.Mock()
    req = make_mocked_request('GET', '/path/to?a=1&b=2', headers=headers,
                              protocol=protocol, payload=payload, app=app)
    assert req.app is app
    assert req.content is payload
    assert req.protocol is protocol
    assert req.transport is protocol.transport
    assert req.headers == headers
    assert req.raw_headers == ((b'FOO', b'bar'),)
    assert req.task is req._task


def test_deprecated_message() -> None:
    req = make_mocked_request('GET', '/path/to?a=1&b=2')
    with pytest.warns(DeprecationWarning):
        assert req.message == req._message


def test_doubleslashes() -> None:
    # NB: //foo/bar is an absolute URL with foo netloc and /bar path
    req = make_mocked_request('GET', '/bar//foo/')
    assert '/bar//foo/' == req.path


def test_content_type_not_specified() -> None:
    req = make_mocked_request('Get', '/')
    assert 'application/octet-stream' == req.content_type


def test_content_type_from_spec() -> None:
    req = make_mocked_request('Get', '/',
                              CIMultiDict([('CONTENT-TYPE',
                                            'application/json')]))
    assert 'application/json' == req.content_type


def test_content_type_from_spec_with_charset() -> None:
    req = make_mocked_request(
        'Get', '/',
        CIMultiDict([('CONTENT-TYPE', 'text/html; charset=UTF-8')]))
    assert 'text/html' == req.content_type
    assert 'UTF-8' == req.charset


def test_calc_content_type_on_getting_charset() -> None:
    req = make_mocked_request(
        'Get', '/',
        CIMultiDict([('CONTENT-TYPE', 'text/html; charset=UTF-8')]))
    assert 'UTF-8' == req.charset
    assert 'text/html' == req.content_type


def test_urlencoded_querystring() -> None:
    req = make_mocked_request(
        'GET', '/yandsearch?text=%D1%82%D0%B5%D0%BA%D1%81%D1%82')
    assert {'text': 'текст'} == req.query


def test_non_ascii_path() -> None:
    req = make_mocked_request('GET', '/путь')
    assert '/путь' == req.path


def test_non_ascii_raw_path() -> None:
    req = make_mocked_request('GET', '/путь')
    assert '/путь' == req.raw_path


def test_content_length() -> None:
    req = make_mocked_request('Get', '/',
                              CIMultiDict([('CONTENT-LENGTH', '123')]))

    assert 123 == req.content_length


def test_range_to_slice_head() -> None:
    def bytes_gen(size):
        for i in range(size):
            yield i % 256
    payload = bytearray(bytes_gen(10000))
    req = make_mocked_request(
        'GET', '/',
        headers=CIMultiDict([('RANGE', 'bytes=0-499')]),
        payload=payload)
    assert isinstance(req.http_range, slice)
    assert req.content[req.http_range] == payload[:500]


def test_range_to_slice_mid() -> None:
    def bytes_gen(size):
        for i in range(size):
            yield i % 256
    payload = bytearray(bytes_gen(10000))
    req = make_mocked_request(
        'GET', '/',
        headers=CIMultiDict([('RANGE', 'bytes=500-999')]),
        payload=payload)
    assert isinstance(req.http_range, slice)
    assert req.content[req.http_range] == payload[500:1000]


def test_range_to_slice_tail_start() -> None:
    def bytes_gen(size):
        for i in range(size):
            yield i % 256
    payload = bytearray(bytes_gen(10000))
    req = make_mocked_request(
        'GET', '/',
        headers=CIMultiDict([('RANGE', 'bytes=9500-')]),
        payload=payload)
    assert isinstance(req.http_range, slice)
    assert req.content[req.http_range] == payload[-500:]


def test_range_to_slice_tail_stop() -> None:
    def bytes_gen(size):
        for i in range(size):
            yield i % 256
    payload = bytearray(bytes_gen(10000))
    req = make_mocked_request(
        'GET', '/',
        headers=CIMultiDict([('RANGE', 'bytes=-500')]),
        payload=payload)
    assert isinstance(req.http_range, slice)
    assert req.content[req.http_range] == payload[-500:]


def test_non_keepalive_on_http10() -> None:
    req = make_mocked_request('GET', '/', version=HttpVersion(1, 0))
    assert not req.keep_alive


def test_non_keepalive_on_closing() -> None:
    req = make_mocked_request('GET', '/', closing=True)
    assert not req.keep_alive


async def test_call_POST_on_GET_request() -> None:
    req = make_mocked_request('GET', '/')

    ret = await req.post()
    assert CIMultiDict() == ret


async def test_call_POST_on_weird_content_type() -> None:
    req = make_mocked_request(
        'POST', '/',
        headers=CIMultiDict({'CONTENT-TYPE': 'something/weird'}))

    ret = await req.post()
    assert CIMultiDict() == ret


async def test_call_POST_twice() -> None:
    req = make_mocked_request('GET', '/')

    ret1 = await req.post()
    ret2 = await req.post()
    assert ret1 is ret2


def test_no_request_cookies() -> None:
    req = make_mocked_request('GET', '/')

    assert req.cookies == {}

    cookies = req.cookies
    assert cookies is req.cookies


def test_request_cookie() -> None:
    headers = CIMultiDict(COOKIE='cookie1=value1; cookie2=value2')
    req = make_mocked_request('GET', '/', headers=headers)

    assert req.cookies == {'cookie1': 'value1',
                           'cookie2': 'value2'}


def test_request_cookie__set_item() -> None:
    headers = CIMultiDict(COOKIE='name=value')
    req = make_mocked_request('GET', '/', headers=headers)

    assert req.cookies == {'name': 'value'}

    with pytest.raises(TypeError):
        req.cookies['my'] = 'value'


def test_match_info() -> None:
    req = make_mocked_request('GET', '/')
    assert req._match_info is req.match_info


def test_request_is_mutable_mapping() -> None:
    req = make_mocked_request('GET', '/')
    assert isinstance(req, MutableMapping)
    req['key'] = 'value'
    assert 'value' == req['key']


def test_request_delitem() -> None:
    req = make_mocked_request('GET', '/')
    req['key'] = 'value'
    assert 'value' == req['key']
    del req['key']
    assert 'key' not in req


def test_request_len() -> None:
    req = make_mocked_request('GET', '/')
    assert len(req) == 0
    req['key'] = 'value'
    assert len(req) == 1


def test_request_iter() -> None:
    req = make_mocked_request('GET', '/')
    req['key'] = 'value'
    req['key2'] = 'value2'
    assert set(req) == {'key', 'key2'}


def test___repr__() -> None:
    req = make_mocked_request('GET', '/path/to')
    assert "<Request GET /path/to >" == repr(req)


def test___repr___non_ascii_path() -> None:
    req = make_mocked_request('GET', '/path/\U0001f415\U0001f308')
    assert "<Request GET /path/\\U0001f415\\U0001f308 >" == repr(req)


def test_http_scheme() -> None:
    req = make_mocked_request('GET', '/', headers={'Host': 'example.com'})
    assert "http" == req.scheme
    assert req.secure is False


def test_https_scheme_by_ssl_transport() -> None:
    req = make_mocked_request('GET', '/', headers={'Host': 'example.com'},
                              sslcontext=True)
    assert "https" == req.scheme
    assert req.secure is True


def test_single_forwarded_header() -> None:
    header = 'by=identifier;for=identifier;host=identifier;proto=identifier'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['by'] == 'identifier'
    assert req.forwarded[0]['for'] == 'identifier'
    assert req.forwarded[0]['host'] == 'identifier'
    assert req.forwarded[0]['proto'] == 'identifier'


@pytest.mark.parametrize(
    "forward_for_in, forward_for_out",
    [
        ("1.2.3.4:1234", "1.2.3.4:1234"),
        ("1.2.3.4", "1.2.3.4"),
        ('"[2001:db8:cafe::17]:1234"', '[2001:db8:cafe::17]:1234'),
        ('"[2001:db8:cafe::17]"', '[2001:db8:cafe::17]'),
    ])
def test_forwarded_node_identifier(forward_for_in, forward_for_out) -> None:
    header = 'for={}'.format(forward_for_in)
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded == ({'for': forward_for_out},)


def test_single_forwarded_header_camelcase() -> None:
    header = 'bY=identifier;fOr=identifier;HOst=identifier;pRoTO=identifier'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['by'] == 'identifier'
    assert req.forwarded[0]['for'] == 'identifier'
    assert req.forwarded[0]['host'] == 'identifier'
    assert req.forwarded[0]['proto'] == 'identifier'


def test_single_forwarded_header_single_param() -> None:
    header = 'BY=identifier'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['by'] == 'identifier'


def test_single_forwarded_header_multiple_param() -> None:
    header = 'By=identifier1,BY=identifier2,  By=identifier3 ,  BY=identifier4'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert len(req.forwarded) == 4
    assert req.forwarded[0]['by'] == 'identifier1'
    assert req.forwarded[1]['by'] == 'identifier2'
    assert req.forwarded[2]['by'] == 'identifier3'
    assert req.forwarded[3]['by'] == 'identifier4'


def test_single_forwarded_header_quoted_escaped() -> None:
    header = r'BY=identifier;pROTO="\lala lan\d\~ 123\!&"'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['by'] == 'identifier'
    assert req.forwarded[0]['proto'] == 'lala land~ 123!&'


def test_single_forwarded_header_custom_param() -> None:
    header = r'BY=identifier;PROTO=https;SOME="other, \"value\""'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert len(req.forwarded) == 1
    assert req.forwarded[0]['by'] == 'identifier'
    assert req.forwarded[0]['proto'] == 'https'
    assert req.forwarded[0]['some'] == 'other, "value"'


def test_single_forwarded_header_empty_params() -> None:
    # This is allowed by the grammar given in RFC 7239
    header = ';For=identifier;;PROTO=https;;;'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['for'] == 'identifier'
    assert req.forwarded[0]['proto'] == 'https'


def test_single_forwarded_header_bad_separator() -> None:
    header = 'BY=identifier PROTO=https'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert 'proto' not in req.forwarded[0]


def test_single_forwarded_header_injection1() -> None:
    # We might receive a header like this if we're sitting behind a reverse
    # proxy that blindly appends a forwarded-element without checking
    # the syntax of existing field-values. We should be able to recover
    # the appended element anyway.
    header = 'for=_injected;by=", for=_real'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert len(req.forwarded) == 2
    assert 'by' not in req.forwarded[0]
    assert req.forwarded[1]['for'] == '_real'


def test_single_forwarded_header_injection2() -> None:
    header = 'very bad syntax, for=_real'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert len(req.forwarded) == 2
    assert 'for' not in req.forwarded[0]
    assert req.forwarded[1]['for'] == '_real'


def test_single_forwarded_header_long_quoted_string() -> None:
    header = 'for="' + '\\\\' * 5000 + '"'
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Forwarded': header}))
    assert req.forwarded[0]['for'] == '\\' * 5000


def test_multiple_forwarded_headers() -> None:
    headers = CIMultiDict()
    headers.add('Forwarded', 'By=identifier1;for=identifier2, BY=identifier3')
    headers.add('Forwarded', 'By=identifier4;fOr=identifier5')
    req = make_mocked_request('GET', '/', headers=headers)
    assert len(req.forwarded) == 3
    assert req.forwarded[0]['by'] == 'identifier1'
    assert req.forwarded[0]['for'] == 'identifier2'
    assert req.forwarded[1]['by'] == 'identifier3'
    assert req.forwarded[2]['by'] == 'identifier4'
    assert req.forwarded[2]['for'] == 'identifier5'


def test_multiple_forwarded_headers_bad_syntax() -> None:
    headers = CIMultiDict()
    headers.add('Forwarded', 'for=_1;by=_2')
    headers.add('Forwarded', 'invalid value')
    headers.add('Forwarded', '')
    headers.add('Forwarded', 'for=_3;by=_4')
    req = make_mocked_request('GET', '/', headers=headers)
    assert len(req.forwarded) == 4
    assert req.forwarded[0]['for'] == '_1'
    assert 'for' not in req.forwarded[1]
    assert 'for' not in req.forwarded[2]
    assert req.forwarded[3]['by'] == '_4'


def test_multiple_forwarded_headers_injection() -> None:
    headers = CIMultiDict()
    # This could be sent by an attacker, hoping to "shadow" the second header.
    headers.add('Forwarded', 'for=_injected;by="')
    # This is added by our trusted reverse proxy.
    headers.add('Forwarded', 'for=_real;by=_actual_proxy')
    req = make_mocked_request('GET', '/', headers=headers)
    assert len(req.forwarded) == 2
    assert 'by' not in req.forwarded[0]
    assert req.forwarded[1]['for'] == '_real'
    assert req.forwarded[1]['by'] == '_actual_proxy'


def test_host_by_host_header() -> None:
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'Host': 'example.com'}))
    assert req.host == 'example.com'


def test_raw_headers() -> None:
    req = make_mocked_request('GET', '/',
                              headers=CIMultiDict({'X-HEADER': 'aaa'}))
    assert req.raw_headers == ((b'X-HEADER', b'aaa'),)


def test_rel_url() -> None:
    req = make_mocked_request('GET', '/path')
    assert URL('/path') == req.rel_url


def test_url_url() -> None:
    req = make_mocked_request('GET', '/path', headers={'HOST': 'example.com'})
    assert URL('http://example.com/path') == req.url


def test_clone() -> None:
    req = make_mocked_request('GET', '/path')
    req2 = req.clone()
    assert req2.method == 'GET'
    assert req2.rel_url == URL('/path')


def test_clone_client_max_size() -> None:
    req = make_mocked_request('GET', '/path', client_max_size=1024)
    req2 = req.clone()
    assert req._client_max_size == req2._client_max_size
    assert req2._client_max_size == 1024


def test_clone_method() -> None:
    req = make_mocked_request('GET', '/path')
    req2 = req.clone(method='POST')
    assert req2.method == 'POST'
    assert req2.rel_url == URL('/path')


def test_clone_rel_url() -> None:
    req = make_mocked_request('GET', '/path')
    req2 = req.clone(rel_url=URL('/path2'))
    assert req2.rel_url == URL('/path2')


def test_clone_rel_url_str() -> None:
    req = make_mocked_request('GET', '/path')
    req2 = req.clone(rel_url='/path2')
    assert req2.rel_url == URL('/path2')


def test_clone_headers() -> None:
    req = make_mocked_request('GET', '/path', headers={'A': 'B'})
    req2 = req.clone(headers=CIMultiDict({'B': 'C'}))
    assert req2.headers == CIMultiDict({'B': 'C'})
    assert req2.raw_headers == ((b'B', b'C'),)


def test_clone_headers_dict() -> None:
    req = make_mocked_request('GET', '/path', headers={'A': 'B'})
    req2 = req.clone(headers={'B': 'C'})
    assert req2.headers == CIMultiDict({'B': 'C'})
    assert req2.raw_headers == ((b'B', b'C'),)


async def test_cannot_clone_after_read(protocol) -> None:
    payload = StreamReader(protocol)
    payload.feed_data(b'data')
    payload.feed_eof()
    req = make_mocked_request('GET', '/path', payload=payload)
    await req.read()
    with pytest.raises(RuntimeError):
        req.clone()


async def test_make_too_big_request(protocol) -> None:
    payload = StreamReader(protocol)
    large_file = 1024 ** 2 * b'x'
    too_large_file = large_file + b'x'
    payload.feed_data(too_large_file)
    payload.feed_eof()
    req = make_mocked_request('POST', '/', payload=payload)
    with pytest.raises(HTTPRequestEntityTooLarge) as err:
        await req.read()

    assert err.value.status_code == 413


async def test_make_too_big_request_adjust_limit(protocol) -> None:
    payload = StreamReader(protocol)
    large_file = 1024 ** 2 * b'x'
    too_large_file = large_file + b'x'
    payload.feed_data(too_large_file)
    payload.feed_eof()
    max_size = 1024**2 + 2
    req = make_mocked_request('POST', '/', payload=payload,
                              client_max_size=max_size)
    txt = await req.read()
    assert len(txt) == 1024**2 + 1


async def test_multipart_formdata(protocol) -> None:
    payload = StreamReader(protocol)
    payload.feed_data(b"""-----------------------------326931944431359\r
Content-Disposition: form-data; name="a"\r
\r
b\r
-----------------------------326931944431359\r
Content-Disposition: form-data; name="c"\r
\r
d\r
-----------------------------326931944431359--\r\n""")
    content_type = "multipart/form-data; boundary="\
                   "---------------------------326931944431359"
    payload.feed_eof()
    req = make_mocked_request('POST', '/',
                              headers={'CONTENT-TYPE': content_type},
                              payload=payload)
    result = await req.post()
    assert dict(result) == {'a': 'b', 'c': 'd'}


async def test_make_too_big_request_limit_None(protocol) -> None:
    payload = StreamReader(protocol)
    large_file = 1024 ** 2 * b'x'
    too_large_file = large_file + b'x'
    payload.feed_data(too_large_file)
    payload.feed_eof()
    max_size = None
    req = make_mocked_request('POST', '/', payload=payload,
                              client_max_size=max_size)
    txt = await req.read()
    assert len(txt) == 1024**2 + 1


def test_remote_peername_tcp() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = ('10.10.10.10', 1234)
    req = make_mocked_request('GET', '/', transport=transp)
    assert req.remote == '10.10.10.10'


def test_remote_peername_unix() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = '/path/to/sock'
    req = make_mocked_request('GET', '/', transport=transp)
    assert req.remote == '/path/to/sock'


def test_save_state_on_clone() -> None:
    req = make_mocked_request('GET', '/')
    req['key'] = 'val'
    req2 = req.clone()
    req2['key'] = 'val2'
    assert req['key'] == 'val'
    assert req2['key'] == 'val2'


def test_clone_scheme() -> None:
    req = make_mocked_request('GET', '/')
    req2 = req.clone(scheme='https')
    assert req2.scheme == 'https'


def test_clone_host() -> None:
    req = make_mocked_request('GET', '/')
    req2 = req.clone(host='example.com')
    assert req2.host == 'example.com'


def test_clone_remote() -> None:
    req = make_mocked_request('GET', '/')
    req2 = req.clone(remote='11.11.11.11')
    assert req2.remote == '11.11.11.11'


@pytest.mark.skipif(not DEBUG,
                    reason="The check is applied in DEBUG mode only")
def test_request_custom_attr() -> None:
    req = make_mocked_request('GET', '/')
    with pytest.warns(DeprecationWarning):
        req.custom = None


def test_remote_with_closed_transport() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = ('10.10.10.10', 1234)
    req = make_mocked_request('GET', '/', transport=transp)
    req._protocol = None
    assert req.remote == '10.10.10.10'


def test_url_http_with_closed_transport() -> None:
    req = make_mocked_request('GET', '/')
    req._protocol = None
    assert str(req.url).startswith('http://')


def test_url_https_with_closed_transport() -> None:
    req = make_mocked_request('GET', '/', sslcontext=True)
    req._protocol = None
    assert str(req.url).startswith('https://')


def test_eq() -> None:
    req1 = make_mocked_request('GET', '/path/to?a=1&b=2')
    req2 = make_mocked_request('GET', '/path/to?a=1&b=2')
    assert req1 != req2
    assert req1 == req1


async def test_loop_prop() -> None:
    loop = asyncio.get_event_loop()
    req = make_mocked_request('GET', '/path', loop=loop)
    with pytest.warns(DeprecationWarning):
        assert req.loop is loop