1from __future__ import absolute_import
2from __future__ import division
3from __future__ import print_function
4
5import email.utils
6import errno
7import httplib2
8import mock
9import os
10import pytest
11from six.moves import http_client, urllib
12import socket
13import tests
14
15
16def _raise_connection_refused_exception(*args, **kwargs):
17    raise socket.error(errno.ECONNREFUSED, "Connection refused.")
18
19
20def test_connection_type():
21    http = httplib2.Http()
22    http.force_exception_to_status_code = False
23    response, content = http.request(
24        tests.DUMMY_URL, connection_type=tests.MockHTTPConnection
25    )
26    assert response["content-location"] == tests.DUMMY_URL
27    assert content == b"the body"
28
29
30def test_bad_status_line_retry():
31    http = httplib2.Http()
32    old_retries = httplib2.RETRIES
33    httplib2.RETRIES = 1
34    http.force_exception_to_status_code = False
35    try:
36        response, content = http.request(
37            tests.DUMMY_URL, connection_type=tests.MockHTTPBadStatusConnection
38        )
39    except http_client.BadStatusLine:
40        assert tests.MockHTTPBadStatusConnection.num_calls == 2
41    httplib2.RETRIES = old_retries
42
43
44def test_unknown_server():
45    http = httplib2.Http()
46    http.force_exception_to_status_code = False
47    with tests.assert_raises(httplib2.ServerNotFoundError):
48        with mock.patch("socket.socket.connect", side_effect=socket.gaierror):
49            http.request("http://no-such-hostname./")
50
51    # Now test with exceptions turned off
52    http.force_exception_to_status_code = True
53    response, content = http.request("http://no-such-hostname./")
54    assert response["content-type"] == "text/plain"
55    assert content.startswith(b"Unable to find")
56    assert response.status == 400
57
58
59@pytest.mark.skipif(
60    os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
61    reason="Fails on Travis py27/pypy, works elsewhere. "
62    "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.",
63)
64@mock.patch("socket.socket.connect", spec=True)
65def test_connection_refused_raises_exception(mock_socket_connect):
66    mock_socket_connect.side_effect = _raise_connection_refused_exception
67    http = httplib2.Http()
68    http.force_exception_to_status_code = False
69    with tests.assert_raises(socket.error):
70        http.request(tests.DUMMY_URL)
71
72
73@pytest.mark.skipif(
74    os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
75    reason="Fails on Travis py27/pypy, works elsewhere. "
76    "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.",
77)
78@mock.patch("socket.socket.connect", spec=True)
79def test_connection_refused_returns_response(mock_socket_connect):
80    mock_socket_connect.side_effect = _raise_connection_refused_exception
81    http = httplib2.Http()
82    http.force_exception_to_status_code = True
83    response, content = http.request(tests.DUMMY_URL)
84    content = content.lower()
85    assert response["content-type"] == "text/plain"
86    assert (
87        b"connection refused" in content
88        or b"actively refused" in content
89        or b"socket is not connected" in content
90    )
91    assert response.status == 400
92
93
94def test_get_iri():
95    http = httplib2.Http()
96    query = u"?a=\N{CYRILLIC CAPITAL LETTER DJE}"
97    with tests.server_reflect() as uri:
98        response, content = http.request(uri + query, "GET")
99        assert response.status == 200
100        reflected = tests.HttpRequest.from_bytes(content)
101        assert reflected.uri == "/?a=%D0%82"
102
103
104def test_get_is_default_method():
105    # Test that GET is the default method
106    http = httplib2.Http()
107    with tests.server_reflect() as uri:
108        response, content = http.request(uri)
109        assert response.status == 200
110        reflected = tests.HttpRequest.from_bytes(content)
111        assert reflected.method == "GET"
112
113
114def test_different_methods():
115    # Test that all methods can be used
116    http = httplib2.Http()
117    methods = ["GET", "PUT", "DELETE", "POST", "unknown"]
118    with tests.server_reflect(request_count=len(methods)) as uri:
119        for method in methods:
120            response, content = http.request(uri, method, body=b" ")
121            assert response.status == 200
122            reflected = tests.HttpRequest.from_bytes(content)
123            assert reflected.method == method
124
125
126def test_head_read():
127    # Test that we don't try to read the response of a HEAD request
128    # since httplib blocks response.read() for HEAD requests.
129    http = httplib2.Http()
130    respond_with = b"HTTP/1.0 200 OK\r\ncontent-length: " b"14\r\n\r\nnon-empty-body"
131    with tests.server_const_bytes(respond_with) as uri:
132        response, content = http.request(uri, "HEAD")
133    assert response.status == 200
134    assert content == b""
135
136
137def test_get_no_cache():
138    # Test that can do a GET w/o the cache turned on.
139    http = httplib2.Http()
140    with tests.server_const_http() as uri:
141        response, content = http.request(uri, "GET")
142    assert response.status == 200
143    assert response.previous is None
144
145
146def test_user_agent():
147    # Test that we provide a default user-agent
148    http = httplib2.Http()
149    with tests.server_reflect() as uri:
150        response, content = http.request(uri, "GET")
151        assert response.status == 200
152        reflected = tests.HttpRequest.from_bytes(content)
153        assert reflected.headers.get("user-agent", "").startswith("Python-httplib2/")
154
155
156def test_user_agent_non_default():
157    # Test that the default user-agent can be over-ridden
158    http = httplib2.Http()
159    with tests.server_reflect() as uri:
160        response, content = http.request(uri, "GET", headers={"User-Agent": "fred/1.0"})
161        assert response.status == 200
162        reflected = tests.HttpRequest.from_bytes(content)
163        assert reflected.headers.get("user-agent") == "fred/1.0"
164
165
166def test_get_300_with_location():
167    # Test the we automatically follow 300 redirects if a Location: header is provided
168    http = httplib2.Http()
169    final_content = b"This is the final destination.\n"
170    routes = {
171        "/final": tests.http_response_bytes(body=final_content),
172        "": tests.http_response_bytes(
173            status="300 Multiple Choices", headers={"location": "/final"}
174        ),
175    }
176    with tests.server_route(routes, request_count=2) as uri:
177        response, content = http.request(uri, "GET")
178    assert response.status == 200
179    assert content == final_content
180    assert response.previous.status == 300
181    assert not response.previous.fromcache
182
183    # Confirm that the intermediate 300 is not cached
184    with tests.server_route(routes, request_count=2) as uri:
185        response, content = http.request(uri, "GET")
186    assert response.status == 200
187    assert content == final_content
188    assert response.previous.status == 300
189    assert not response.previous.fromcache
190
191
192def test_get_300_with_location_noredirect():
193    # Test the we automatically follow 300 redirects if a Location: header is provided
194    http = httplib2.Http()
195    http.follow_redirects = False
196    response = tests.http_response_bytes(
197        status="300 Multiple Choices",
198        headers={"location": "/final"},
199        body=b"redirect body",
200    )
201    with tests.server_const_bytes(response) as uri:
202        response, content = http.request(uri, "GET")
203    assert response.status == 300
204
205
206def test_get_300_without_location():
207    # Not giving a Location: header in a 300 response is acceptable
208    # In which case we just return the 300 response
209    http = httplib2.Http()
210    with tests.server_const_http(
211        status="300 Multiple Choices", body=b"redirect body"
212    ) as uri:
213        response, content = http.request(uri, "GET")
214    assert response.status == 300
215    assert response.previous is None
216    assert content == b"redirect body"
217
218
219def test_get_301():
220    # Test that we automatically follow 301 redirects
221    # and that we cache the 301 response
222    http = httplib2.Http(cache=tests.get_cache_path())
223    destination = ""
224    routes = {
225        "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
226        "": tests.http_response_bytes(
227            status="301 Now where did I leave that URL",
228            headers={"location": "/final"},
229            body=b"redirect body",
230        ),
231    }
232    with tests.server_route(routes, request_count=3) as uri:
233        destination = urllib.parse.urljoin(uri, "/final")
234        response1, content1 = http.request(uri, "GET")
235        response2, content2 = http.request(uri, "GET")
236    assert response1.status == 200
237    assert "content-location" in response2
238    assert response1["content-location"] == destination
239    assert content1 == b"This is the final destination.\n"
240    assert response1.previous.status == 301
241    assert not response1.previous.fromcache
242
243    assert response2.status == 200
244    assert response2["content-location"] == destination
245    assert content2 == b"This is the final destination.\n"
246    assert response2.previous.status == 301
247    assert response2.previous.fromcache
248
249
250@pytest.mark.skip(
251    not os.environ.get("httplib2_test_still_run_skipped")
252    and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
253    reason="FIXME: timeout on Travis py27 and pypy, works elsewhere",
254)
255def test_head_301():
256    # Test that we automatically follow 301 redirects
257    http = httplib2.Http()
258    destination = ""
259    routes = {
260        "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
261        "": tests.http_response_bytes(
262            status="301 Now where did I leave that URL",
263            headers={"location": "/final"},
264            body=b"redirect body",
265        ),
266    }
267    with tests.server_route(routes, request_count=2) as uri:
268        destination = urllib.parse.urljoin(uri, "/final")
269        response, content = http.request(uri, "HEAD")
270    assert response.status == 200
271    assert response["content-location"] == destination
272    assert response.previous.status == 301
273    assert not response.previous.fromcache
274
275
276@pytest.mark.xfail(
277    reason=(
278        "FIXME: 301 cache works only with follow_redirects, should work " "regardless"
279    )
280)
281def test_get_301_no_redirect():
282    # Test that we cache the 301 response
283    http = httplib2.Http(cache=tests.get_cache_path(), timeout=0.5)
284    http.follow_redirects = False
285    response = tests.http_response_bytes(
286        status="301 Now where did I leave that URL",
287        headers={"location": "/final", "cache-control": "max-age=300"},
288        body=b"redirect body",
289        add_date=True,
290    )
291    with tests.server_const_bytes(response) as uri:
292        response, _ = http.request(uri, "GET")
293        assert response.status == 301
294        assert not response.fromcache
295        response, _ = http.request(uri, "GET")
296        assert response.status == 301
297        assert response.fromcache
298
299
300def test_get_302():
301    # Test that we automatically follow 302 redirects
302    # and that we DO NOT cache the 302 response
303    http = httplib2.Http(cache=tests.get_cache_path())
304    second_url, final_url = "", ""
305    routes = {
306        "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
307        "/second": tests.http_response_bytes(
308            status="302 Found", headers={"location": "/final"}, body=b"second redirect"
309        ),
310        "": tests.http_response_bytes(
311            status="302 Found", headers={"location": "/second"}, body=b"redirect body"
312        ),
313    }
314    with tests.server_route(routes, request_count=7) as uri:
315        second_url = urllib.parse.urljoin(uri, "/second")
316        final_url = urllib.parse.urljoin(uri, "/final")
317        response1, content1 = http.request(second_url, "GET")
318        response2, content2 = http.request(second_url, "GET")
319        response3, content3 = http.request(uri, "GET")
320    assert response1.status == 200
321    assert response1["content-location"] == final_url
322    assert content1 == b"This is the final destination.\n"
323    assert response1.previous.status == 302
324    assert not response1.previous.fromcache
325
326    assert response2.status == 200
327    # FIXME:
328    # assert response2.fromcache
329    assert response2["content-location"] == final_url
330    assert content2 == b"This is the final destination.\n"
331    assert response2.previous.status == 302
332    assert not response2.previous.fromcache
333    assert response2.previous["content-location"] == second_url
334
335    assert response3.status == 200
336    # FIXME:
337    # assert response3.fromcache
338    assert content3 == b"This is the final destination.\n"
339    assert response3.previous.status == 302
340    assert not response3.previous.fromcache
341
342
343def test_get_302_redirection_limit():
344    # Test that we can set a lower redirection limit
345    # and that we raise an exception when we exceed
346    # that limit.
347    http = httplib2.Http()
348    http.force_exception_to_status_code = False
349    routes = {
350        "/second": tests.http_response_bytes(
351            status="302 Found", headers={"location": "/final"}, body=b"second redirect"
352        ),
353        "": tests.http_response_bytes(
354            status="302 Found", headers={"location": "/second"}, body=b"redirect body"
355        ),
356    }
357    with tests.server_route(routes, request_count=4) as uri:
358        try:
359            http.request(uri, "GET", redirections=1)
360            assert False, "This should not happen"
361        except httplib2.RedirectLimit:
362            pass
363        except Exception:
364            assert False, "Threw wrong kind of exception "
365
366        # Re-run the test with out the exceptions
367        http.force_exception_to_status_code = True
368        response, content = http.request(uri, "GET", redirections=1)
369
370    assert response.status == 500
371    assert response.reason.startswith("Redirected more")
372    assert response["status"] == "302"
373    assert content == b"second redirect"
374    assert response.previous is not None
375
376
377def test_get_302_no_location():
378    # Test that we throw an exception when we get
379    # a 302 with no Location: header.
380    http = httplib2.Http()
381    http.force_exception_to_status_code = False
382    with tests.server_const_http(status="302 Found", request_count=2) as uri:
383        try:
384            http.request(uri, "GET")
385            assert False, "Should never reach here"
386        except httplib2.RedirectMissingLocation:
387            pass
388        except Exception:
389            assert False, "Threw wrong kind of exception "
390
391        # Re-run the test with out the exceptions
392        http.force_exception_to_status_code = True
393        response, content = http.request(uri, "GET")
394
395    assert response.status == 500
396    assert response.reason.startswith("Redirected but")
397    assert "302" == response["status"]
398    assert content == b""
399
400
401@pytest.mark.skip(
402    not os.environ.get("httplib2_test_still_run_skipped")
403    and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
404    reason="FIXME: timeout on Travis py27 and pypy, works elsewhere",
405)
406def test_303():
407    # Do a follow-up GET on a Location: header
408    # returned from a POST that gave a 303.
409    http = httplib2.Http()
410    routes = {
411        "/final": tests.make_http_reflect(),
412        "": tests.make_http_reflect(
413            status="303 See Other", headers={"location": "/final"}
414        ),
415    }
416    with tests.server_route(routes, request_count=2) as uri:
417        response, content = http.request(uri, "POST", " ")
418    assert response.status == 200
419    reflected = tests.HttpRequest.from_bytes(content)
420    assert reflected.uri == "/final"
421    assert response.previous.status == 303
422
423    # Skip follow-up GET
424    http = httplib2.Http()
425    http.follow_redirects = False
426    with tests.server_route(routes, request_count=1) as uri:
427        response, content = http.request(uri, "POST", " ")
428    assert response.status == 303
429
430    # All methods can be used
431    http = httplib2.Http()
432    cases = "DELETE GET HEAD POST PUT EVEN_NEW_ONES".split(" ")
433    with tests.server_route(routes, request_count=len(cases) * 2) as uri:
434        for method in cases:
435            response, content = http.request(uri, method, body=b"q q")
436            assert response.status == 200
437            reflected = tests.HttpRequest.from_bytes(content)
438            assert reflected.method == "GET"
439
440
441def test_etag_used():
442    # Test that we use ETags properly to validate our cache
443    cache_path = tests.get_cache_path()
444    http = httplib2.Http(cache=cache_path)
445    response_kwargs = dict(
446        add_date=True,
447        add_etag=True,
448        body=b"something",
449        headers={"cache-control": "public,max-age=300"},
450    )
451
452    def handler(request):
453        if request.headers.get("range"):
454            return tests.http_response_bytes(status=206, **response_kwargs)
455        return tests.http_response_bytes(**response_kwargs)
456
457    with tests.server_request(handler, request_count=2) as uri:
458        response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"})
459        assert response["etag"] == '"437b930db84b8079c2dd804a71936b5f"'
460
461        http.request(uri, "GET", headers={"accept-encoding": "identity"})
462        response, _ = http.request(
463            uri,
464            "GET",
465            headers={"accept-encoding": "identity", "cache-control": "must-revalidate"},
466        )
467        assert response.status == 200
468        assert response.fromcache
469
470        # TODO: API to read cache item, at least internal to tests
471        cache_file_name = os.path.join(
472            cache_path, httplib2.safename(httplib2.urlnorm(uri)[-1])
473        )
474        with open(cache_file_name, "r") as f:
475            status_line = f.readline()
476        assert status_line.startswith("status:")
477
478        response, content = http.request(
479            uri, "HEAD", headers={"accept-encoding": "identity"}
480        )
481        assert response.status == 200
482        assert response.fromcache
483
484        response, content = http.request(
485            uri, "GET", headers={"accept-encoding": "identity", "range": "bytes=0-0"}
486        )
487        assert response.status == 206
488        assert not response.fromcache
489
490
491def test_etag_ignore():
492    # Test that we can forcibly ignore ETags
493    http = httplib2.Http(cache=tests.get_cache_path())
494    response_kwargs = dict(add_date=True, add_etag=True)
495    with tests.server_reflect(request_count=3, **response_kwargs) as uri:
496        response, content = http.request(
497            uri, "GET", headers={"accept-encoding": "identity"}
498        )
499        assert response.status == 200
500        assert response["etag"] != ""
501
502        response, content = http.request(
503            uri,
504            "GET",
505            headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
506        )
507        reflected = tests.HttpRequest.from_bytes(content)
508        assert reflected.headers.get("if-none-match")
509
510        http.ignore_etag = True
511        response, content = http.request(
512            uri,
513            "GET",
514            headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
515        )
516        assert not response.fromcache
517        reflected = tests.HttpRequest.from_bytes(content)
518        assert not reflected.headers.get("if-none-match")
519
520
521def test_etag_override():
522    # Test that we can forcibly ignore ETags
523    http = httplib2.Http(cache=tests.get_cache_path())
524    response_kwargs = dict(add_date=True, add_etag=True)
525    with tests.server_reflect(request_count=3, **response_kwargs) as uri:
526        response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"})
527        assert response.status == 200
528        assert response["etag"] != ""
529
530        response, content = http.request(
531            uri,
532            "GET",
533            headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
534        )
535        assert response.status == 200
536        reflected = tests.HttpRequest.from_bytes(content)
537        assert reflected.headers.get("if-none-match")
538        assert reflected.headers.get("if-none-match") != "fred"
539
540        response, content = http.request(
541            uri,
542            "GET",
543            headers={
544                "accept-encoding": "identity",
545                "cache-control": "max-age=0",
546                "if-none-match": "fred",
547            },
548        )
549        assert response.status == 200
550        reflected = tests.HttpRequest.from_bytes(content)
551        assert reflected.headers.get("if-none-match") == "fred"
552
553
554@pytest.mark.skip(reason="was commented in legacy code")
555def test_get_304_end_to_end():
556    pass
557    # Test that end to end headers get overwritten in the cache
558    # uri = urllib.parse.urljoin(base, "304/end2end.cgi")
559    # response, content = http.request(uri, 'GET')
560    # assertNotEqual(response['etag'], "")
561    # old_date = response['date']
562    # time.sleep(2)
563
564    # response, content = http.request(uri, 'GET', headers = {'Cache-Control': 'max-age=0'})
565    # # The response should be from the cache, but the Date: header should be updated.
566    # new_date = response['date']
567    # assert new_date != old_date
568    # assert response.status == 200
569    # assert response.fromcache == True
570
571
572def test_get_304_last_modified():
573    # Test that we can still handle a 304
574    # by only using the last-modified cache validator.
575    http = httplib2.Http(cache=tests.get_cache_path())
576    date = email.utils.formatdate()
577
578    def handler(read):
579        read()
580        yield tests.http_response_bytes(
581            status=200, body=b"something", headers={"date": date, "last-modified": date}
582        )
583
584        request2 = read()
585        assert request2.headers["if-modified-since"] == date
586        yield tests.http_response_bytes(status=304)
587
588    with tests.server_yield(handler, request_count=2) as uri:
589        response, content = http.request(uri, "GET")
590        assert response.get("last-modified") == date
591
592        response, content = http.request(uri, "GET")
593        assert response.status == 200
594        assert response.fromcache
595
596
597def test_get_307():
598    # Test that we do follow 307 redirects but
599    # do not cache the 307
600    http = httplib2.Http(cache=tests.get_cache_path(), timeout=1)
601    r307 = tests.http_response_bytes(status=307, headers={"location": "/final"})
602    r200 = tests.http_response_bytes(
603        status=200,
604        add_date=True,
605        body=b"final content\n",
606        headers={"cache-control": "max-age=300"},
607    )
608
609    with tests.server_list_http([r307, r200, r307]) as uri:
610        response, content = http.request(uri, "GET")
611        assert response.previous.status == 307
612        assert not response.previous.fromcache
613        assert response.status == 200
614        assert not response.fromcache
615        assert content == b"final content\n"
616
617        response, content = http.request(uri, "GET")
618        assert response.previous.status == 307
619        assert not response.previous.fromcache
620        assert response.status == 200
621        assert response.fromcache
622        assert content == b"final content\n"
623
624
625def test_post_307():
626    # 307: follow with same method
627    http = httplib2.Http(cache=tests.get_cache_path(), timeout=1)
628    http.follow_all_redirects = True
629    r307 = tests.http_response_bytes(status=307, headers={"location": "/final"})
630    r200 = tests.http_response_bytes(status=200, body=b"final content\n")
631
632    with tests.server_list_http([r307, r200, r307, r200]) as uri:
633        response, content = http.request(uri, "POST")
634        assert response.previous.status == 307
635        assert not response.previous.fromcache
636        assert response.status == 200
637        assert not response.fromcache
638        assert content == b"final content\n"
639
640        response, content = http.request(uri, "POST")
641        assert response.previous.status == 307
642        assert not response.previous.fromcache
643        assert response.status == 200
644        assert not response.fromcache
645        assert content == b"final content\n"
646
647
648def test_change_308():
649    # 308: follow with same method, cache redirect
650    http = httplib2.Http(cache=tests.get_cache_path(), timeout=1)
651    routes = {
652        "/final": tests.make_http_reflect(),
653        "": tests.http_response_bytes(
654            status="308 Permanent Redirect",
655            add_date=True,
656            headers={"cache-control": "max-age=300", "location": "/final"},
657        ),
658    }
659
660    with tests.server_route(routes, request_count=3) as uri:
661        response, content = http.request(uri, "CHANGE", body=b"hello308")
662        assert response.previous.status == 308
663        assert not response.previous.fromcache
664        assert response.status == 200
665        assert not response.fromcache
666        assert content.startswith(b"CHANGE /final HTTP")
667
668        response, content = http.request(uri, "CHANGE")
669        assert response.previous.status == 308
670        assert response.previous.fromcache
671        assert response.status == 200
672        assert not response.fromcache
673        assert content.startswith(b"CHANGE /final HTTP")
674
675
676def test_get_410():
677    # Test that we pass 410's through
678    http = httplib2.Http()
679    with tests.server_const_http(status=410) as uri:
680        response, content = http.request(uri, "GET")
681        assert response.status == 410
682
683
684def test_get_duplicate_headers():
685    # Test that duplicate headers get concatenated via ','
686    http = httplib2.Http()
687    response = b"""HTTP/1.0 200 OK\r\n\
688Link: link1\r\n\
689Content-Length: 7\r\n\
690Link: link2\r\n\r\n\
691content"""
692    with tests.server_const_bytes(response) as uri:
693        response, content = http.request(uri, "GET")
694        assert response.status == 200
695        assert content == b"content"
696        assert response["link"], "link1, link2"
697
698
699def test_custom_redirect_codes():
700    http = httplib2.Http()
701    http.redirect_codes = set([300])
702    with tests.server_const_http(status=301, request_count=1) as uri:
703        response, content = http.request(uri, "GET")
704        assert response.status == 301
705        assert response.previous is None
706
707
708def test_cwe93_inject_crlf():
709    # https://cwe.mitre.org/data/definitions/93.html
710    # GET /?q= HTTP/1.1      <- injected "HTTP/1.1" from attacker
711    # injected: attack
712    # ignore-http: HTTP/1.1  <- nominal "HTTP/1.1" from library
713    # Host: localhost:57285
714    http = httplib2.Http()
715    with tests.server_reflect() as uri:
716        danger_url = urllib.parse.urljoin(
717            uri, "?q= HTTP/1.1\r\ninjected: attack\r\nignore-http:"
718        )
719        response, content = http.request(danger_url, "GET")
720        assert response.status == 200
721        req = tests.HttpRequest.from_bytes(content)
722        assert req.headers.get("injected") is None
723
724
725def test_inject_space():
726    # Injecting space into request line is precursor to CWE-93 and possibly other injections
727    http = httplib2.Http()
728    with tests.server_reflect() as uri:
729        # "\r\nignore-http:" suffix is nuance for current server implementation
730        # please only pay attention to space after "?q="
731        danger_url = urllib.parse.urljoin(uri, "?q= HTTP/1.1\r\nignore-http:")
732        response, content = http.request(danger_url, "GET")
733        assert response.status == 200
734        req = tests.HttpRequest.from_bytes(content)
735        assert req.uri == "/?q=%20HTTP/1.1%0D%0Aignore-http:"
736