blob: 81c39f09b1299ed23e2e5b9a85cd36dcd7337a93 (
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
|
import cStringIO
from cStringIO import StringIO
from mock import MagicMock
from libmproxy.protocol.http import *
from netlib import odict
from netlib.http import http1
from netlib.http.semantics import CONTENT_MISSING
import tutils
import tservers
def mock_protocol(data='', chunked=False):
rfile = cStringIO.StringIO(data)
wfile = cStringIO.StringIO()
return http1.HTTP1Protocol(rfile=rfile, wfile=wfile)
# TODO: move test to netlib http1 protocol
# def test_stripped_chunked_encoding_no_content():
# """
# https://github.com/mitmproxy/mitmproxy/issues/186
# """
# r = tutils.tresp(content="")
# r.headers["Transfer-Encoding"] = ["chunked"]
# assert "Content-Length" in r._assemble_headers()
#
# r = tutils.treq(content="")
# r.headers["Transfer-Encoding"] = ["chunked"]
# assert "Content-Length" in r._assemble_headers()
#
class TestHTTPResponse:
def test_read_from_stringio(self):
s = "HTTP/1.1 200 OK\r\n" \
"Content-Length: 7\r\n" \
"\r\n"\
"content\r\n" \
"HTTP/1.1 204 OK\r\n" \
"\r\n"
protocol = mock_protocol(s)
r = HTTPResponse.from_protocol(protocol, "GET")
assert r.status_code == 200
assert r.content == "content"
assert HTTPResponse.from_protocol(protocol, "GET").status_code == 204
protocol = mock_protocol(s)
# HEAD must not have content by spec. We should leave it on the pipe.
r = HTTPResponse.from_protocol(protocol, "HEAD")
assert r.status_code == 200
assert r.content == ""
tutils.raises(
"Invalid server response: 'content",
HTTPResponse.from_protocol, protocol, "GET"
)
class TestHTTPFlow(object):
def test_repr(self):
f = tutils.tflow(resp=True, err=True)
assert repr(f)
class TestInvalidRequests(tservers.HTTPProxTest):
ssl = True
def test_double_connect(self):
p = self.pathoc()
r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
assert r.status_code == 400
assert "Must not CONNECT on already encrypted connection" in r.body
def test_relative_request(self):
p = self.pathoc_raw()
p.connect()
r = p.request("get:/p/200")
assert r.status_code == 400
assert "Invalid HTTP request form" in r.body
|
