aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--netlib/http/cookies.py1
-rw-r--r--netlib/http/message.py10
-rw-r--r--netlib/http/request.py12
-rw-r--r--netlib/http/response.py14
-rw-r--r--test/http/http1/test_assemble.py13
-rw-r--r--test/http/http1/test_read.py3
-rw-r--r--test/http/test_cookies.py1
-rw-r--r--test/http/test_message.py91
-rw-r--r--test/http/test_models.py94
-rw-r--r--test/http/test_request.py42
-rw-r--r--test/http/test_response.py99
11 files changed, 208 insertions, 172 deletions
diff --git a/netlib/http/cookies.py b/netlib/http/cookies.py
index 78b03a83..18544b5e 100644
--- a/netlib/http/cookies.py
+++ b/netlib/http/cookies.py
@@ -58,6 +58,7 @@ def _read_quoted_string(s, start):
escaping = False
ret = []
# Skip the first quote
+ i = start # initialize in case the loop doesn't run.
for i in range(start + 1, len(s)):
if escaping:
ret.append(s[i])
diff --git a/netlib/http/message.py b/netlib/http/message.py
index 7cb18f52..e4e799ca 100644
--- a/netlib/http/message.py
+++ b/netlib/http/message.py
@@ -18,6 +18,16 @@ else:
_always_bytes = lambda x: utils.always_bytes(x, "utf-8", "surrogateescape")
+class MessageData(object):
+ def __eq__(self, other):
+ if isinstance(other, MessageData):
+ return self.__dict__ == other.__dict__
+ return False
+
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+
class Message(object):
def __init__(self, data):
self.data = data
diff --git a/netlib/http/request.py b/netlib/http/request.py
index 325c0080..095b5945 100644
--- a/netlib/http/request.py
+++ b/netlib/http/request.py
@@ -10,10 +10,10 @@ from netlib.http import cookies
from netlib.odict import ODict
from .. import encoding
from .headers import Headers
-from .message import Message, _native, _always_bytes
+from .message import Message, _native, _always_bytes, MessageData
-class RequestData(object):
+class RequestData(MessageData):
def __init__(self, first_line_format, method, scheme, host, port, path, http_version, headers=None, content=None,
timestamp_start=None, timestamp_end=None):
if not headers:
@@ -32,14 +32,6 @@ class RequestData(object):
self.timestamp_start = timestamp_start
self.timestamp_end = timestamp_end
- def __eq__(self, other):
- if isinstance(other, RequestData):
- return self.__dict__ == other.__dict__
- return False
-
- def __ne__(self, other):
- return not self.__eq__(other)
-
class Request(Message):
"""
diff --git a/netlib/http/response.py b/netlib/http/response.py
index db31d2b9..66e5ded6 100644
--- a/netlib/http/response.py
+++ b/netlib/http/response.py
@@ -4,12 +4,12 @@ import warnings
from . import cookies
from .headers import Headers
-from .message import Message, _native, _always_bytes
+from .message import Message, _native, _always_bytes, MessageData
from .. import utils
from ..odict import ODict
-class ResponseData(object):
+class ResponseData(MessageData):
def __init__(self, http_version, status_code, reason=None, headers=None, content=None,
timestamp_start=None, timestamp_end=None):
if not headers:
@@ -24,14 +24,6 @@ class ResponseData(object):
self.timestamp_start = timestamp_start
self.timestamp_end = timestamp_end
- def __eq__(self, other):
- if isinstance(other, ResponseData):
- return self.__dict__ == other.__dict__
- return False
-
- def __ne__(self, other):
- return not self.__eq__(other)
-
class Response(Message):
"""
@@ -48,7 +40,7 @@ class Response(Message):
utils.pretty_size(len(self.content))
)
else:
- details = "content missing"
+ details = "no content"
return "Response({status_code} {reason}, {details})".format(
status_code=self.status_code,
reason=self.reason,
diff --git a/test/http/http1/test_assemble.py b/test/http/http1/test_assemble.py
index 460e22c5..ed94292d 100644
--- a/test/http/http1/test_assemble.py
+++ b/test/http/http1/test_assemble.py
@@ -78,10 +78,19 @@ def test_assemble_request_headers():
# https://github.com/mitmproxy/mitmproxy/issues/186
r = treq(content=b"")
r.headers["Transfer-Encoding"] = "chunked"
- c = _assemble_request_headers(r)
+ c = _assemble_request_headers(r.data)
assert b"Transfer-Encoding" in c
- assert b"host" in _assemble_request_headers(treq(headers=Headers()))
+
+def test_assemble_request_headers_host_header():
+ r = treq()
+ r.headers = Headers()
+ c = _assemble_request_headers(r.data)
+ assert b"host" in c
+
+ r.host = None
+ c = _assemble_request_headers(r.data)
+ assert b"host" not in c
def test_assemble_response_headers():
diff --git a/test/http/http1/test_read.py b/test/http/http1/test_read.py
index a0085db9..84a43f8b 100644
--- a/test/http/http1/test_read.py
+++ b/test/http/http1/test_read.py
@@ -117,6 +117,9 @@ def test_connection_close():
headers["connection"] = "close"
assert connection_close(b"HTTP/1.1", headers)
+ headers["connection"] = "foobar"
+ assert connection_close(b"HTTP/1.0", headers)
+ assert not connection_close(b"HTTP/1.1", headers)
def test_expected_http_body_size():
# Expect: 100-continue
diff --git a/test/http/test_cookies.py b/test/http/test_cookies.py
index 413b6241..34bb64f2 100644
--- a/test/http/test_cookies.py
+++ b/test/http/test_cookies.py
@@ -21,6 +21,7 @@ def test_read_quoted_string():
[(r'"f\\o" x', 0), (r"f\o", 6)],
[(r'"f\\" x', 0), (r"f" + '\\', 5)],
[('"fo\\\"" x', 0), ("fo\"", 6)],
+ [('"foo" x', 7), ("", 8)],
]
for q, a in tokens:
assert cookies._read_quoted_string(*q) == a
diff --git a/test/http/test_message.py b/test/http/test_message.py
index b0b7e27f..2c37dc3e 100644
--- a/test/http/test_message.py
+++ b/test/http/test_message.py
@@ -1,43 +1,53 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, division
-from netlib.http import decoded
-from netlib.tutils import tresp
+from netlib.http import decoded, Headers
+from netlib.tutils import tresp, raises
def _test_passthrough_attr(message, attr):
- def t(self=None):
- assert getattr(message, attr) == getattr(message.data, attr)
- setattr(message, attr, "foo")
- assert getattr(message.data, attr) == "foo"
- return t
+ assert getattr(message, attr) == getattr(message.data, attr)
+ setattr(message, attr, "foo")
+ assert getattr(message.data, attr) == "foo"
def _test_decoded_attr(message, attr):
- def t(self=None):
- assert getattr(message, attr) == getattr(message.data, attr).decode("utf8")
- # Set str, get raw bytes
- setattr(message, attr, "foo")
- assert getattr(message.data, attr) == b"foo"
- # Set raw bytes, get decoded
- setattr(message.data, attr, b"bar")
- assert getattr(message, attr) == "bar"
- # Set bytes, get raw bytes
- setattr(message, attr, b"baz")
- assert getattr(message.data, attr) == b"baz"
-
- # Set UTF8
- setattr(message, attr, "Non-Autorisé")
- assert getattr(message.data, attr) == b"Non-Autoris\xc3\xa9"
- # Don't fail on garbage
- setattr(message.data, attr, b"foo\xFF\x00bar")
- assert getattr(message, attr).startswith("foo")
- assert getattr(message, attr).endswith("bar")
- # foo.bar = foo.bar should not cause any side effects.
- d = getattr(message, attr)
- setattr(message, attr, d)
- assert getattr(message.data, attr) == b"foo\xFF\x00bar"
- return t
+ assert getattr(message, attr) == getattr(message.data, attr).decode("utf8")
+ # Set str, get raw bytes
+ setattr(message, attr, "foo")
+ assert getattr(message.data, attr) == b"foo"
+ # Set raw bytes, get decoded
+ setattr(message.data, attr, b"bar")
+ assert getattr(message, attr) == "bar"
+ # Set bytes, get raw bytes
+ setattr(message, attr, b"baz")
+ assert getattr(message.data, attr) == b"baz"
+
+ # Set UTF8
+ setattr(message, attr, "Non-Autorisé")
+ assert getattr(message.data, attr) == b"Non-Autoris\xc3\xa9"
+ # Don't fail on garbage
+ setattr(message.data, attr, b"foo\xFF\x00bar")
+ assert getattr(message, attr).startswith("foo")
+ assert getattr(message, attr).endswith("bar")
+ # foo.bar = foo.bar should not cause any side effects.
+ d = getattr(message, attr)
+ setattr(message, attr, d)
+ assert getattr(message.data, attr) == b"foo\xFF\x00bar"
+
+
+class TestMessageData(object):
+ def test_eq_ne(self):
+ data = tresp(timestamp_start=42, timestamp_end=42).data
+ same = tresp(timestamp_start=42, timestamp_end=42).data
+ assert data == same
+ assert not data != same
+
+ other = tresp(content=b"foo").data
+ assert not data == other
+ assert data != other
+
+ assert data != 0
class TestMessage(object):
@@ -67,12 +77,20 @@ class TestMessage(object):
assert resp.data.content == b""
assert resp.headers["content-length"] == "0"
- test_content_basic = _test_passthrough_attr(tresp(), "content")
- test_headers = _test_passthrough_attr(tresp(), "headers")
- test_timestamp_start = _test_passthrough_attr(tresp(), "timestamp_start")
- test_timestamp_end = _test_passthrough_attr(tresp(), "timestamp_end")
+ def test_content_basic(self):
+ _test_passthrough_attr(tresp(), "content")
+
+ def test_headers(self):
+ _test_passthrough_attr(tresp(), "headers")
- test_http_version = _test_decoded_attr(tresp(), "http_version")
+ def test_timestamp_start(self):
+ _test_passthrough_attr(tresp(), "timestamp_start")
+
+ def test_timestamp_end(self):
+ _test_passthrough_attr(tresp(), "timestamp_end")
+
+ def teste_http_version(self):
+ _test_decoded_attr(tresp(), "http_version")
class TestDecodedDecorator(object):
@@ -133,4 +151,3 @@ class TestDecodedDecorator(object):
assert "content-encoding" not in r.headers
assert r.content is None
-
diff --git a/test/http/test_models.py b/test/http/test_models.py
deleted file mode 100644
index 76a05446..00000000
--- a/test/http/test_models.py
+++ /dev/null
@@ -1,94 +0,0 @@
-
-from netlib import tutils
-from netlib.odict import ODict, ODictCaseless
-from netlib.http import Response, Headers, CONTENT_MISSING
-
-class TestResponse(object):
- def test_headers(self):
- tutils.raises(AssertionError, Response,
- b"HTTP/1.1",
- 200,
- headers='foobar',
- )
-
- resp = Response(
- b"HTTP/1.1",
- 200,
- )
- assert isinstance(resp.headers, Headers)
-
- def test_equal(self):
- a = tutils.tresp(timestamp_start=42, timestamp_end=43)
- b = tutils.tresp(timestamp_start=42, timestamp_end=43)
- assert a == b
-
- assert not a == 'foo'
- assert not b == 'foo'
- assert not 'foo' == a
- assert not 'foo' == b
-
- def test_repr(self):
- r = tutils.tresp()
- assert "unknown content type" in repr(r)
- r.headers["content-type"] = "foo"
- assert "foo" in repr(r)
- assert repr(tutils.tresp(content=CONTENT_MISSING))
-
- def test_get_cookies_none(self):
- resp = tutils.tresp()
- resp.headers = Headers()
- assert not resp.get_cookies()
-
- def test_get_cookies_simple(self):
- resp = tutils.tresp()
- resp.headers = Headers(set_cookie="cookiename=cookievalue")
- result = resp.get_cookies()
- assert len(result) == 1
- assert "cookiename" in result
- assert result["cookiename"][0] == ["cookievalue", ODict()]
-
- def test_get_cookies_with_parameters(self):
- resp = tutils.tresp()
- resp.headers = Headers(set_cookie="cookiename=cookievalue;domain=example.com;expires=Wed Oct 21 16:29:41 2015;path=/; HttpOnly")
- result = resp.get_cookies()
- assert len(result) == 1
- assert "cookiename" in result
- assert result["cookiename"][0][0] == "cookievalue"
- attrs = result["cookiename"][0][1]
- assert len(attrs) == 4
- assert attrs["domain"] == ["example.com"]
- assert attrs["expires"] == ["Wed Oct 21 16:29:41 2015"]
- assert attrs["path"] == ["/"]
- assert attrs["httponly"] == [None]
-
- def test_get_cookies_no_value(self):
- resp = tutils.tresp()
- resp.headers = Headers(set_cookie="cookiename=; Expires=Thu, 01-Jan-1970 00:00:01 GMT; path=/")
- result = resp.get_cookies()
- assert len(result) == 1
- assert "cookiename" in result
- assert result["cookiename"][0][0] == ""
- assert len(result["cookiename"][0][1]) == 2
-
- def test_get_cookies_twocookies(self):
- resp = tutils.tresp()
- resp.headers = Headers([
- [b"Set-Cookie", b"cookiename=cookievalue"],
- [b"Set-Cookie", b"othercookie=othervalue"]
- ])
- result = resp.get_cookies()
- assert len(result) == 2
- assert "cookiename" in result
- assert result["cookiename"][0] == ["cookievalue", ODict()]
- assert "othercookie" in result
- assert result["othercookie"][0] == ["othervalue", ODict()]
-
- def test_set_cookies(self):
- resp = tutils.tresp()
- v = resp.get_cookies()
- v.add("foo", ["bar", ODictCaseless()])
- resp.set_cookies(v)
-
- v = resp.get_cookies()
- assert len(v) == 1
- assert v["foo"] == [["bar", ODictCaseless()]]
diff --git a/test/http/test_request.py b/test/http/test_request.py
index 15bdd3e3..8cf69ffe 100644
--- a/test/http/test_request.py
+++ b/test/http/test_request.py
@@ -17,31 +17,31 @@ class TestRequestData(object):
assert isinstance(treq(headers=None).headers, Headers)
- def test_eq_ne(self):
- request_data = treq().data
- same = treq().data
- assert request_data == same
- assert not request_data != same
-
- other = treq(content=b"foo").data
- assert not request_data == other
- assert request_data != other
-
- assert request_data != 0
-
class TestRequestCore(object):
+ """
+ Tests for builtins and the attributes that are directly proxied from the data structure
+ """
def test_repr(self):
request = treq()
assert repr(request) == "Request(GET address:22/path)"
request.host = None
assert repr(request) == "Request(GET /path)"
- test_first_line_format = _test_passthrough_attr(treq(), "first_line_format")
- test_method = _test_decoded_attr(treq(), "method")
- test_scheme = _test_decoded_attr(treq(), "scheme")
- test_port = _test_passthrough_attr(treq(), "port")
- test_path = _test_decoded_attr(treq(), "path")
+ def test_first_line_format(self):
+ _test_passthrough_attr(treq(), "first_line_format")
+
+ def test_method(self):
+ _test_decoded_attr(treq(), "method")
+
+ def test_scheme(self):
+ _test_decoded_attr(treq(), "scheme")
+
+ def test_port(self):
+ _test_passthrough_attr(treq(), "port")
+
+ def test_path(self):
+ _test_decoded_attr(treq(), "path")
def test_host(self):
if six.PY2:
@@ -86,6 +86,9 @@ class TestRequestCore(object):
class TestRequestUtils(object):
+ """
+ Tests for additional convenience methods.
+ """
def test_url(self):
request = treq()
assert request.url == "http://address:22/path"
@@ -199,6 +202,11 @@ class TestRequestUtils(object):
def test_constrain_encoding(self):
request = treq()
+
+ h = request.headers.copy()
+ request.constrain_encoding() # no-op if there is no accept_encoding header.
+ assert request.headers == h
+
request.headers["Accept-Encoding"] = "identity, gzip, foo"
request.constrain_encoding()
assert "foo" not in request.headers["Accept-Encoding"]
diff --git a/test/http/test_response.py b/test/http/test_response.py
index 02fac3df..a1f4abd7 100644
--- a/test/http/test_response.py
+++ b/test/http/test_response.py
@@ -1,3 +1,100 @@
from __future__ import absolute_import, print_function, division
-# TODO \ No newline at end of file
+from netlib.http import Headers
+from netlib.odict import ODict, ODictCaseless
+from netlib.tutils import raises, tresp
+from .test_message import _test_passthrough_attr, _test_decoded_attr
+
+
+class TestResponseData(object):
+ def test_init(self):
+ with raises(AssertionError):
+ tresp(headers="foobar")
+
+ assert isinstance(tresp(headers=None).headers, Headers)
+
+
+class TestResponseCore(object):
+ """
+ Tests for builtins and the attributes that are directly proxied from the data structure
+ """
+ def test_repr(self):
+ response = tresp()
+ assert repr(response) == "Response(200 OK, unknown content type, 7B)"
+ response.content = None
+ assert repr(response) == "Response(200 OK, no content)"
+
+ def test_status_code(self):
+ _test_passthrough_attr(tresp(), "status_code")
+
+ def test_reason(self):
+ _test_decoded_attr(tresp(), "reason")
+
+
+class TestResponseUtils(object):
+ """
+ Tests for additional convenience methods.
+ """
+ def test_get_cookies_none(self):
+ resp = tresp()
+ resp.headers = Headers()
+ assert not resp.cookies
+
+ def test_get_cookies_empty(self):
+ resp = tresp()
+ resp.headers = Headers(set_cookie="")
+ assert not resp.cookies
+
+ def test_get_cookies_simple(self):
+ resp = tresp()
+ resp.headers = Headers(set_cookie="cookiename=cookievalue")
+ result = resp.cookies
+ assert len(result) == 1
+ assert "cookiename" in result
+ assert result["cookiename"][0] == ["cookievalue", ODict()]
+
+ def test_get_cookies_with_parameters(self):
+ resp = tresp()
+ resp.headers = Headers(set_cookie="cookiename=cookievalue;domain=example.com;expires=Wed Oct 21 16:29:41 2015;path=/; HttpOnly")
+ result = resp.cookies
+ assert len(result) == 1
+ assert "cookiename" in result
+ assert result["cookiename"][0][0] == "cookievalue"
+ attrs = result["cookiename"][0][1]
+ assert len(attrs) == 4
+ assert attrs["domain"] == ["example.com"]
+ assert attrs["expires"] == ["Wed Oct 21 16:29:41 2015"]
+ assert attrs["path"] == ["/"]
+ assert attrs["httponly"] == [None]
+
+ def test_get_cookies_no_value(self):
+ resp = tresp()
+ resp.headers = Headers(set_cookie="cookiename=; Expires=Thu, 01-Jan-1970 00:00:01 GMT; path=/")
+ result = resp.cookies
+ assert len(result) == 1
+ assert "cookiename" in result
+ assert result["cookiename"][0][0] == ""
+ assert len(result["cookiename"][0][1]) == 2
+
+ def test_get_cookies_twocookies(self):
+ resp = tresp()
+ resp.headers = Headers([
+ [b"Set-Cookie", b"cookiename=cookievalue"],
+ [b"Set-Cookie", b"othercookie=othervalue"]
+ ])
+ result = resp.cookies
+ assert len(result) == 2
+ assert "cookiename" in result
+ assert result["cookiename"][0] == ["cookievalue", ODict()]
+ assert "othercookie" in result
+ assert result["othercookie"][0] == ["othervalue", ODict()]
+
+ def test_set_cookies(self):
+ resp = tresp()
+ v = resp.cookies
+ v.add("foo", ["bar", ODictCaseless()])
+ resp.set_cookies(v)
+
+ v = resp.cookies
+ assert len(v) == 1
+ assert v["foo"] == [["bar", ODictCaseless()]]
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-08 17:07:29 +0000