From db02553e2afee044faac898d12bd8d1adadbcd21 Mon Sep 17 00:00:00 2001 From: Thomas Kriechbaumer Date: Sat, 1 Aug 2015 10:40:19 +0200 Subject: move code from mitmproxy to netlib --- test/test_flow.py | 76 ++++++++++++++++++++++++++++--------------------------- 1 file changed, 39 insertions(+), 37 deletions(-) (limited to 'test/test_flow.py') 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") -- cgit v1.2.3