aboutsummaryrefslogtreecommitdiffstats
path: root/test/http/http1/test_assemble.py
blob: 8a0a54f16b6425492b4abfcf2b10162bc2c9e182 (plain)
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
from __future__ import absolute_import, print_function, division
from netlib.exceptions import HttpException
from netlib.http import CONTENT_MISSING, Headers
from netlib.http.http1.assemble import (
    assemble_request, assemble_request_head, assemble_response,
    assemble_response_head, _assemble_request_line, _assemble_request_headers,
    _assemble_response_headers
)
from netlib.tutils import treq, raises, tresp


def test_assemble_request():
    c = assemble_request(treq()) == (
        b"GET /path HTTP/1.1\r\n"
        b"header: qvalue\r\n"
        b"Host: address:22\r\n"
        b"Content-Length: 7\r\n"
        b"\r\n"
        b"content"
    )

    with raises(HttpException):
        assemble_request(treq(body=CONTENT_MISSING))


def test_assemble_request_head():
    c = assemble_request_head(treq())
    assert b"GET" in c
    assert b"qvalue" in c
    assert b"content" not in c


def test_assemble_response():
    c = assemble_response(tresp()) == (
        b"HTTP/1.1 200 OK\r\n"
        b"header-response: svalue\r\n"
        b"Content-Length: 7\r\n"
        b"\r\n"
        b"message"
    )

    with raises(HttpException):
        assemble_response(tresp(body=CONTENT_MISSING))


def test_assemble_response_head():
    c = assemble_response_head(tresp())
    assert b"200" in c
    assert b"svalue" in c
    assert b"message" not in c


def test_assemble_request_line():
    assert _assemble_request_line(treq()) == b"GET /path HTTP/1.1"

    authority_request = treq(method=b"CONNECT", form_in="authority")
    assert _assemble_request_line(authority_request) == b"CONNECT address:22 HTTP/1.1"

    absolute_request = treq(form_in="absolute")
    assert _assemble_request_line(absolute_request) == b"GET http://address:22/path HTTP/1.1"

    with raises(RuntimeError):
        _assemble_request_line(treq(), "invalid_form")


def test_assemble_request_headers():
    # https://github.com/mitmproxy/mitmproxy/issues/186
    r = treq(body=b"")
    r.headers[b"Transfer-Encoding"] = b"chunked"
    c = _assemble_request_headers(r)
    assert b"Content-Length" in c
    assert b"Transfer-Encoding" not in c

    assert b"Host" in _assemble_request_headers(treq(headers=Headers()))

    assert b"Proxy-Connection" not in _assemble_request_headers(
        treq(headers=Headers(Proxy_Connection="42"))
    )


def test_assemble_response_headers():
    # https://github.com/mitmproxy/mitmproxy/issues/186
    r = tresp(body=b"")
    r.headers["Transfer-Encoding"] = b"chunked"
    c = _assemble_response_headers(r)
    assert b"Content-Length" in c
    assert b"Transfer-Encoding" not in c

    assert b"Proxy-Connection" not in _assemble_response_headers(
        tresp(headers=Headers(Proxy_Connection=b"42"))
    )