diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_console.py | 6 | ||||
| -rw-r--r-- | test/test_console_contentview.py | 6 | ||||
| -rw-r--r-- | test/test_dump.py | 11 | ||||
| -rw-r--r-- | test/test_flow.py | 76 | ||||
| -rw-r--r-- | test/test_protocol_http.py | 259 | ||||
| -rw-r--r-- | test/test_server.py | 7 | ||||
| -rw-r--r-- | test/test_utils.py | 16 | ||||
| -rw-r--r-- | test/tutils.py | 82 |
8 files changed, 76 insertions, 387 deletions
diff --git a/test/test_console.py b/test/test_console.py index ed8408a5..3304fdbb 100644 --- a/test/test_console.py +++ b/test/test_console.py @@ -4,6 +4,8 @@ import mock import gc from os.path import normpath import mock_urwid + +import netlib.tutils from libmproxy import console from libmproxy.console import common @@ -60,13 +62,13 @@ class TestConsoleState: def _add_response(self, state): f = self._add_request(state) - f.response = tutils.tresp() + f.response = netlib.tutils.tresp() state.update_flow(f) def test_add_response(self): c = console.ConsoleState() f = self._add_request(c) - f.response = tutils.tresp() + f.response = netlib.tutils.tresp() c.focus = None c.update_flow(f) diff --git a/test/test_console_contentview.py b/test/test_console_contentview.py index f2d82419..b98e1021 100644 --- a/test/test_console_contentview.py +++ b/test/test_console_contentview.py @@ -4,7 +4,9 @@ if os.name == "nt": raise SkipTest("Skipped on Windows.") import sys +import netlib.utils from netlib import odict + import libmproxy.console.contentview as cv from libmproxy import utils, flow, encoding import tutils @@ -65,10 +67,10 @@ class TestContentView: assert f[0].startswith("XML") def test_view_urlencoded(self): - d = utils.urlencode([("one", "two"), ("three", "four")]) + d = netlib.utils.urlencode([("one", "two"), ("three", "four")]) v = cv.ViewURLEncoded() assert v([], d, 100) - d = utils.urlencode([("adsfa", "")]) + d = netlib.utils.urlencode([("adsfa", "")]) v = cv.ViewURLEncoded() assert v([], d, 100) diff --git a/test/test_dump.py b/test/test_dump.py index 46c832d3..b3d724a5 100644 --- a/test/test_dump.py +++ b/test/test_dump.py @@ -1,17 +1,18 @@ import os from cStringIO import StringIO +import netlib.tutils from netlib.http.semantics import CONTENT_MISSING from libmproxy import dump, flow -from libmproxy.protocol import http +from libmproxy.protocol import http, http_wrappers from libmproxy.proxy.primitives import Log import tutils import mock def test_strfuncs(): - t = tutils.tresp() + t = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) t.is_replay = True dump.str_response(t) @@ -26,14 +27,14 @@ def test_strfuncs(): class TestDumpMaster: def _cycle(self, m, content): - f = tutils.tflow(req=tutils.treq(content)) + f = tutils.tflow(req=netlib.tutils.treq(content)) l = Log("connect") l.reply = mock.MagicMock() m.handle_log(l) m.handle_clientconnect(f.client_conn) m.handle_serverconnect(f.server_conn) m.handle_request(f) - f.response = tutils.tresp(content) + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp(content)) f = m.handle_response(f) m.handle_clientdisconnect(f.client_conn) return f @@ -70,7 +71,7 @@ class TestDumpMaster: f = tutils.tflow() f.request.content = CONTENT_MISSING m.handle_request(f) - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) f.response.content = CONTENT_MISSING m.handle_response(f) assert "content missing" in cs.getvalue() diff --git a/test/test_flow.py b/test/test_flow.py index c72a583c..711688da 100644 --- a/test/test_flow.py +++ b/test/test_flow.py @@ -3,18 +3,20 @@ import time import os.path from cStringIO import StringIO import email.utils +import mock +import netlib.utils from netlib import odict -from netlib.http.semantics import CONTENT_MISSING +from netlib.http.semantics import CONTENT_MISSING, HDR_FORM_URLENCODED, HDR_FORM_MULTIPART from libmproxy import filt, protocol, controller, utils, tnetstring, flow +from libmproxy.protocol import http_wrappers from libmproxy.protocol.primitives import Error, Flow from libmproxy.protocol.http import decoded from libmproxy.proxy.config import HostMatcher from libmproxy.proxy import ProxyConfig from libmproxy.proxy.server import DummyServer from libmproxy.proxy.connection import ClientConnection -import mock import tutils @@ -22,7 +24,7 @@ def test_app_registry(): ar = flow.AppRegistry() ar.add("foo", "domain", 80) - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.host = "domain" r.port = 80 assert ar.get(r) @@ -30,7 +32,7 @@ def test_app_registry(): r.port = 81 assert not ar.get(r) - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.host = "domain2" r.port = 80 assert not ar.get(r) @@ -41,7 +43,7 @@ def test_app_registry(): class TestStickyCookieState: def _response(self, cookie, host): s = flow.StickyCookieState(filt.parse(".*")) - f = tutils.tflow(req=tutils.treq(host=host, port=80), resp=True) + f = tutils.tflow(req=netlib.tutils.treq(host=host, port=80), resp=True) f.response.headers["Set-Cookie"] = [cookie] s.handle_response(f) return s, f @@ -383,7 +385,7 @@ class TestFlow: def test_backup(self): f = tutils.tflow() - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) f.request.content = "foo" assert not f.modified() f.backup() @@ -516,16 +518,16 @@ class TestState: assert c.add_flow(newf) assert c.active_flow_count() == 2 - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) assert c.update_flow(f) assert c.flow_count() == 2 assert c.active_flow_count() == 1 - _ = tutils.tresp() + _ = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) assert not c.update_flow(None) assert c.active_flow_count() == 1 - newf.response = tutils.tresp() + newf.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) assert c.update_flow(newf) assert c.active_flow_count() == 0 @@ -557,7 +559,7 @@ class TestState: c.set_limit("~s") assert c.limit_txt == "~s" assert len(c.view) == 0 - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) c.update_flow(f) assert len(c.view) == 1 c.set_limit(None) @@ -589,7 +591,7 @@ class TestState: def _add_response(self, state): f = tutils.tflow() state.add_flow(f) - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) state.update_flow(f) def _add_error(self, state): @@ -807,11 +809,11 @@ class TestFlowMaster: fm.anticomp = True f = tutils.tflow(req=None) fm.handle_clientconnect(f.client_conn) - f.request = tutils.treq() + f.request = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) fm.handle_request(f) assert s.flow_count() == 1 - f.response = tutils.tresp() + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) fm.handle_response(f) assert not fm.handle_response(None) assert s.flow_count() == 1 @@ -856,7 +858,7 @@ class TestFlowMaster: s = flow.State() f = tutils.tflow() - f.response = tutils.tresp(f.request) + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp(f.request)) pb = [f] fm = flow.FlowMaster(None, s) @@ -910,7 +912,7 @@ class TestFlowMaster: def test_server_playback_kill(self): s = flow.State() f = tutils.tflow() - f.response = tutils.tresp(f.request) + f.response = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp(f.request)) pb = [f] fm = flow.FlowMaster(None, s) fm.refresh_server_playback = True @@ -1009,7 +1011,7 @@ class TestRequest: assert r.get_state() == r2.get_state() def test_get_url(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) assert r.url == "http://address:22/path" @@ -1030,7 +1032,7 @@ class TestRequest: assert r.pretty_url(True) == "https://foo.com:22/path" def test_path_components(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.path = "/" assert r.get_path_components() == [] r.path = "/foo/bar" @@ -1050,8 +1052,8 @@ class TestRequest: def test_getset_form_urlencoded(self): d = odict.ODict([("one", "two"), ("three", "four")]) - r = tutils.treq(content=utils.urlencode(d.lst)) - r.headers["content-type"] = [protocol.http.HDR_FORM_URLENCODED] + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq(content=netlib.utils.urlencode(d.lst))) + r.headers["content-type"] = [HDR_FORM_URLENCODED] assert r.get_form_urlencoded() == d d = odict.ODict([("x", "y")]) @@ -1064,7 +1066,7 @@ class TestRequest: def test_getset_query(self): h = odict.ODictCaseless() - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.path = "/foo?x=y&a=b" q = r.get_query() assert q.lst == [("x", "y"), ("a", "b")] @@ -1087,7 +1089,7 @@ class TestRequest: def test_anticache(self): h = odict.ODictCaseless() - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.headers = h h["if-modified-since"] = ["test"] h["if-none-match"] = ["test"] @@ -1096,7 +1098,7 @@ class TestRequest: assert not "if-none-match" in r.headers def test_replace(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.path = "path/foo" r.headers["Foo"] = ["fOo"] r.content = "afoob" @@ -1106,31 +1108,31 @@ class TestRequest: assert r.headers["boo"] == ["boo"] def test_constrain_encoding(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.headers["accept-encoding"] = ["gzip", "oink"] r.constrain_encoding() assert "oink" not in r.headers["accept-encoding"] def test_decodeencode(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" r.decode() assert not r.headers["content-encoding"] assert r.content == "falafel" - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.content = "falafel" assert not r.decode() - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" r.encode("identity") assert r.headers["content-encoding"] == ["identity"] assert r.content == "falafel" - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" r.encode("gzip") @@ -1141,7 +1143,7 @@ class TestRequest: assert r.content == "falafel" def test_get_decoded_content(self): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) r.content = None r.headers["content-encoding"] = ["identity"] assert r.get_decoded_content() == None @@ -1153,7 +1155,7 @@ class TestRequest: def test_get_content_type(self): h = odict.ODictCaseless() h["Content-Type"] = ["text/plain"] - resp = tutils.tresp() + resp = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) resp.headers = h assert resp.headers.get_first("content-type") == "text/plain" @@ -1166,7 +1168,7 @@ class TestResponse: assert resp2.get_state() == resp.get_state() def test_refresh(self): - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) n = time.time() r.headers["date"] = [email.utils.formatdate(n)] pre = r.headers["date"] @@ -1184,7 +1186,7 @@ class TestResponse: r.refresh() def test_refresh_cookie(self): - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) # Invalid expires format, sent to us by Reddit. c = "rfoo=bar; Domain=reddit.com; expires=Thu, 31 Dec 2037 23:59:59 GMT; Path=/" @@ -1194,7 +1196,7 @@ class TestResponse: assert "00:21:38" in r._refresh_cookie(c, 60) def test_replace(self): - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) r.headers["Foo"] = ["fOo"] r.content = "afoob" assert r.replace("foo(?i)", "boo") == 3 @@ -1202,21 +1204,21 @@ class TestResponse: assert r.headers["boo"] == ["boo"] def test_decodeencode(self): - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" assert r.decode() assert not r.headers["content-encoding"] assert r.content == "falafel" - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" r.encode("identity") assert r.headers["content-encoding"] == ["identity"] assert r.content == "falafel" - r = tutils.tresp() + r = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) r.headers["content-encoding"] = ["identity"] r.content = "falafel" r.encode("gzip") @@ -1233,7 +1235,7 @@ class TestResponse: def test_get_content_type(self): h = odict.ODictCaseless() h["Content-Type"] = ["text/plain"] - resp = tutils.tresp() + resp = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) resp.headers = h assert resp.headers.get_first("content-type") == "text/plain" @@ -1277,7 +1279,7 @@ class TestClientConnection: def test_decoded(): - r = tutils.treq() + r = http_wrappers.HTTPRequest.wrap(netlib.tutils.treq()) assert r.content == "content" assert not r.headers["content-encoding"] r.encode("gzip") diff --git a/test/test_protocol_http.py b/test/test_protocol_http.py index 75f0a7b9..81c39f09 100644 --- a/test/test_protocol_http.py +++ b/test/test_protocol_http.py @@ -17,14 +17,7 @@ def mock_protocol(data='', chunked=False): return http1.HTTP1Protocol(rfile=rfile, wfile=wfile) - -def test_HttpAuthenticationError(): - x = HttpAuthenticationError({"foo": "bar"}) - assert str(x) - assert "foo" in x.headers - - -# TODO: move test to netlib +# TODO: move test to netlib http1 protocol # def test_stripped_chunked_encoding_no_content(): # """ # https://github.com/mitmproxy/mitmproxy/issues/186 @@ -38,183 +31,6 @@ def test_HttpAuthenticationError(): # assert "Content-Length" in r._assemble_headers() # -class TestHTTPRequest: - def test_asterisk_form_in(self): - f = tutils.tflow(req=None) - protocol = mock_protocol("OPTIONS * HTTP/1.1") - f.request = HTTPRequest.from_protocol(protocol) - - assert f.request.form_in == "relative" - f.request.host = f.server_conn.address.host - f.request.port = f.server_conn.address.port - f.request.scheme = "http" - assert protocol.assemble(f.request) == ( - "OPTIONS * HTTP/1.1\r\n" - "Host: address:22\r\n" - "Content-Length: 0\r\n\r\n") - - def test_relative_form_in(self): - protocol = mock_protocol("GET /foo\xff HTTP/1.1") - tutils.raises("Bad HTTP request line", HTTPRequest.from_protocol, protocol) - - protocol = mock_protocol("GET /foo HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: h2c") - r = HTTPRequest.from_protocol(protocol) - assert r.headers["Upgrade"] == ["h2c"] - - def test_expect_header(self): - protocol = mock_protocol( - "GET / HTTP/1.1\r\nContent-Length: 3\r\nExpect: 100-continue\r\n\r\nfoobar") - r = HTTPRequest.from_protocol(protocol) - assert protocol.tcp_handler.wfile.getvalue() == "HTTP/1.1 100 Continue\r\n\r\n" - assert r.content == "foo" - assert protocol.tcp_handler.rfile.read(3) == "bar" - - def test_authority_form_in(self): - protocol = mock_protocol("CONNECT oops-no-port.com HTTP/1.1") - tutils.raises("Bad HTTP request line", HTTPRequest.from_protocol, protocol) - - protocol = mock_protocol("CONNECT address:22 HTTP/1.1") - r = HTTPRequest.from_protocol(protocol) - r.scheme, r.host, r.port = "http", "address", 22 - assert protocol.assemble(r) == ( - "CONNECT address:22 HTTP/1.1\r\n" - "Host: address:22\r\n" - "Content-Length: 0\r\n\r\n") - assert r.pretty_url(False) == "address:22" - - def test_absolute_form_in(self): - protocol = mock_protocol("GET oops-no-protocol.com HTTP/1.1") - tutils.raises("Bad HTTP request line", HTTPRequest.from_protocol, protocol) - - protocol = mock_protocol("GET http://address:22/ HTTP/1.1") - r = HTTPRequest.from_protocol(protocol) - assert protocol.assemble(r) == ( - "GET http://address:22/ HTTP/1.1\r\n" - "Host: address:22\r\n" - "Content-Length: 0\r\n\r\n") - - def test_http_options_relative_form_in(self): - """ - Exercises fix for Issue #392. - """ - protocol = mock_protocol("OPTIONS /secret/resource HTTP/1.1") - r = HTTPRequest.from_protocol(protocol) - r.host = 'address' - r.port = 80 - r.scheme = "http" - assert protocol.assemble(r) == ( - "OPTIONS /secret/resource HTTP/1.1\r\n" - "Host: address\r\n" - "Content-Length: 0\r\n\r\n") - - def test_http_options_absolute_form_in(self): - protocol = mock_protocol("OPTIONS http://address/secret/resource HTTP/1.1") - r = HTTPRequest.from_protocol(protocol) - r.host = 'address' - r.port = 80 - r.scheme = "http" - assert protocol.assemble(r) == ( - "OPTIONS http://address:80/secret/resource HTTP/1.1\r\n" - "Host: address\r\n" - "Content-Length: 0\r\n\r\n") - - def test_set_url(self): - r = tutils.treq_absolute() - r.url = "https://otheraddress:42/ORLY" - assert r.scheme == "https" - assert r.host == "otheraddress" - assert r.port == 42 - assert r.path == "/ORLY" - - def test_repr(self): - r = tutils.treq() - assert repr(r) - - def test_pretty_host(self): - r = tutils.treq() - assert r.pretty_host(True) == "address" - assert r.pretty_host(False) == "address" - r.headers["host"] = ["other"] - assert r.pretty_host(True) == "other" - assert r.pretty_host(False) == "address" - r.host = None - assert r.pretty_host(True) == "other" - assert r.pretty_host(False) is None - del r.headers["host"] - assert r.pretty_host(True) is None - assert r.pretty_host(False) is None - - # Invalid IDNA - r.headers["host"] = [".disqus.com"] - assert r.pretty_host(True) == ".disqus.com" - - def test_get_form_for_urlencoded(self): - r = tutils.treq() - r.headers.add("content-type", "application/x-www-form-urlencoded") - r.get_form_urlencoded = MagicMock() - - r.get_form() - - assert r.get_form_urlencoded.called - - def test_get_form_for_multipart(self): - r = tutils.treq() - r.headers.add("content-type", "multipart/form-data") - r.get_form_multipart = MagicMock() - - r.get_form() - - assert r.get_form_multipart.called - - def test_get_cookies_none(self): - h = odict.ODictCaseless() - r = tutils.treq() - r.headers = h - assert len(r.get_cookies()) == 0 - - def test_get_cookies_single(self): - h = odict.ODictCaseless() - h["Cookie"] = ["cookiename=cookievalue"] - r = tutils.treq() - r.headers = h - result = r.get_cookies() - assert len(result) == 1 - assert result['cookiename'] == ['cookievalue'] - - def test_get_cookies_double(self): - h = odict.ODictCaseless() - h["Cookie"] = [ - "cookiename=cookievalue;othercookiename=othercookievalue" - ] - r = tutils.treq() - r.headers = h - result = r.get_cookies() - assert len(result) == 2 - assert result['cookiename'] == ['cookievalue'] - assert result['othercookiename'] == ['othercookievalue'] - - def test_get_cookies_withequalsign(self): - h = odict.ODictCaseless() - h["Cookie"] = [ - "cookiename=coo=kievalue;othercookiename=othercookievalue" - ] - r = tutils.treq() - r.headers = h - result = r.get_cookies() - assert len(result) == 2 - assert result['cookiename'] == ['coo=kievalue'] - assert result['othercookiename'] == ['othercookievalue'] - - def test_set_cookies(self): - h = odict.ODictCaseless() - h["Cookie"] = ["cookiename=cookievalue"] - r = tutils.treq() - r.headers = h - result = r.get_cookies() - result["cookiename"] = ["foo"] - r.set_cookies(result) - assert r.get_cookies()["cookiename"] == ["foo"] - class TestHTTPResponse: def test_read_from_stringio(self): @@ -241,80 +57,7 @@ class TestHTTPResponse: HTTPResponse.from_protocol, protocol, "GET" ) - 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): - h = odict.ODictCaseless() - resp = tutils.tresp() - resp.headers = h - assert not resp.get_cookies() - - def test_get_cookies_simple(self): - h = odict.ODictCaseless() - h["Set-Cookie"] = ["cookiename=cookievalue"] - resp = tutils.tresp() - resp.headers = h - result = resp.get_cookies() - assert len(result) == 1 - assert "cookiename" in result - assert result["cookiename"][0] == ["cookievalue", odict.ODict()] - - def test_get_cookies_with_parameters(self): - h = odict.ODictCaseless() - h["Set-Cookie"] = [ - "cookiename=cookievalue;domain=example.com;expires=Wed Oct 21 16:29:41 2015;path=/; HttpOnly"] - resp = tutils.tresp() - resp.headers = h - 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): - h = odict.ODictCaseless() - h["Set-Cookie"] = [ - "cookiename=; Expires=Thu, 01-Jan-1970 00:00:01 GMT; path=/" - ] - resp = tutils.tresp() - resp.headers = h - 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): - h = odict.ODictCaseless() - h["Set-Cookie"] = ["cookiename=cookievalue", "othercookie=othervalue"] - resp = tutils.tresp() - resp.headers = h - result = resp.get_cookies() - assert len(result) == 2 - assert "cookiename" in result - assert result["cookiename"][0] == ["cookievalue", odict.ODict()] - assert "othercookie" in result - assert result["othercookie"][0] == ["othervalue", odict.ODict()] - - def test_set_cookies(self): - resp = tutils.tresp() - v = resp.get_cookies() - v.add("foo", ["bar", odict.ODictCaseless()]) - resp.set_cookies(v) - v = resp.get_cookies() - assert len(v) == 1 - assert v["foo"] == [["bar", odict.ODictCaseless()]] class TestHTTPFlow(object): diff --git a/test/test_server.py b/test/test_server.py index 27b8aad3..77ba4576 100644 --- a/test/test_server.py +++ b/test/test_server.py @@ -2,6 +2,7 @@ import socket import time from OpenSSL import SSL +import netlib.tutils from netlib import tcp, http, socks from netlib.certutils import SSLCert from netlib.http import authentication @@ -9,7 +10,7 @@ from netlib.http.semantics import CONTENT_MISSING from libpathod import pathoc, pathod from libmproxy.proxy.config import HostMatcher -from libmproxy.protocol import KILL, Error +from libmproxy.protocol import KILL, Error, http_wrappers import tutils import tservers @@ -783,7 +784,7 @@ class TestStreamRequest(tservers.HTTPProxTest): class MasterFakeResponse(tservers.TestMaster): def handle_request(self, f): - resp = tutils.tresp() + resp = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) f.reply(resp) @@ -848,7 +849,7 @@ class TestTransparentResolveError(tservers.TransparentProxTest): class MasterIncomplete(tservers.TestMaster): def handle_request(self, f): - resp = tutils.tresp() + resp = http_wrappers.HTTPResponse.wrap(netlib.tutils.tresp()) resp.content = CONTENT_MISSING f.reply(resp) diff --git a/test/test_utils.py b/test/test_utils.py index 0c514f5d..0ee23b97 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -44,11 +44,6 @@ def test_pretty_json(): assert not utils.pretty_json("moo") -def test_urldecode(): - s = "one=two&three=four" - assert len(utils.urldecode(s)) == 2 - - def test_multipartdecode(): boundary = 'somefancyboundary' headers = odict.ODict( @@ -116,13 +111,6 @@ def test_LRUCache(): assert len(cache.cache) == 2 -def test_unparse_url(): - assert utils.unparse_url("http", "foo.com", 99, "") == "http://foo.com:99" - assert utils.unparse_url("http", "foo.com", 80, "") == "http://foo.com" - assert utils.unparse_url("https", "foo.com", 80, "") == "https://foo.com:80" - assert utils.unparse_url("https", "foo.com", 443, "") == "https://foo.com" - - def test_parse_size(): assert not utils.parse_size("") assert utils.parse_size("1") == 1 @@ -144,7 +132,3 @@ def test_parse_content_type(): def test_safe_subn(): assert utils.safe_subn("foo", u"bar", "\xc2foo") - - -def test_urlencode(): - assert utils.urlencode([('foo', 'bar')]) diff --git a/test/tutils.py b/test/tutils.py index 7c7d1db3..61b1154c 100644 --- a/test/tutils.py +++ b/test/tutils.py @@ -1,21 +1,24 @@ -from cStringIO import StringIO import os import shutil import tempfile import argparse -from contextlib import contextmanager import sys +import mock_urwid +from cStringIO import StringIO +from contextlib import contextmanager +from nose.plugins.skip import SkipTest +from mock import Mock +from time import time + +from netlib import certutils, odict +import netlib.tutils + from libmproxy import flow, utils, controller -from libmproxy.protocol import http +from libmproxy.protocol import http, http_wrappers from libmproxy.proxy.connection import ClientConnection, ServerConnection -import mock_urwid from libmproxy.console.flowview import FlowView from libmproxy.console import ConsoleState from libmproxy.protocol.primitives import Error -from netlib import certutils, odict -from nose.plugins.skip import SkipTest -from mock import Mock -from time import time def _SkipWindows(): @@ -43,12 +46,17 @@ def tflow(client_conn=True, server_conn=True, req=True, resp=None, err=None): if server_conn is True: server_conn = tserver_conn() if req is True: - req = treq() + req = netlib.tutils.treq() if resp is True: - resp = tresp() + resp = netlib.tutils.tresp() if err is True: err = terr() + if req: + req = http_wrappers.HTTPRequest.wrap(req) + if resp: + resp = http_wrappers.HTTPResponse.wrap(resp) + f = http.HTTPFlow(client_conn, server_conn) f.request = req f.response = resp @@ -83,60 +91,6 @@ def tserver_conn(): return c -def treq(content="content", scheme="http", host="address", port=22): - """ - @return: libmproxy.protocol.http.HTTPRequest - """ - headers = odict.ODictCaseless() - headers["header"] = ["qvalue"] - req = http.HTTPRequest( - "relative", - "GET", - scheme, - host, - port, - "/path", - (1, 1), - headers, - content, - None, - None, - None, - ) - return req - - -def treq_absolute(content="content"): - """ - @return: libmproxy.protocol.http.HTTPRequest - """ - r = treq(content) - r.form_in = r.form_out = "absolute" - r.host = "address" - r.port = 22 - r.scheme = "http" - return r - - -def tresp(content="message"): - """ - @return: libmproxy.protocol.http.HTTPResponse - """ - - headers = odict.ODictCaseless() - headers["header_response"] = ["svalue"] - - resp = http.HTTPResponse( - (1, 1), - 200, - "OK", - headers, - content, - time(), - time(), - ) - return resp - def terr(content="error"): """ |