diff options
Diffstat (limited to 'test/test_flow.py')
| -rw-r--r-- | test/test_flow.py | 223 |
1 files changed, 168 insertions, 55 deletions
diff --git a/test/test_flow.py b/test/test_flow.py index f5d06906..2609b7cb 100644 --- a/test/test_flow.py +++ b/test/test_flow.py @@ -1,4 +1,6 @@ -import Queue, time, os.path +import Queue +import time +import os.path from cStringIO import StringIO import email.utils import mock @@ -33,7 +35,6 @@ def test_app_registry(): assert ar.get(r) - class TestStickyCookieState: def _response(self, cookie, host): s = flow.StickyCookieState(filt.parse(".*")) @@ -115,7 +116,15 @@ class TestClientPlaybackState: class TestServerPlaybackState: def test_hash(self): - s = flow.ServerPlaybackState(None, [], False, False, None, False, None, False) + s = flow.ServerPlaybackState( + None, + [], + False, + False, + None, + False, + None, + False) r = tutils.tflow() r2 = tutils.tflow() @@ -131,7 +140,15 @@ class TestServerPlaybackState: assert s._hash(r) != s._hash(r2) def test_headers(self): - s = flow.ServerPlaybackState(["foo"], [], False, False, None, False, None, False) + s = flow.ServerPlaybackState( + ["foo"], + [], + False, + False, + None, + False, + None, + False) r = tutils.tflow(resp=True) r.request.headers["foo"] = ["bar"] r2 = tutils.tflow(resp=True) @@ -152,7 +169,9 @@ class TestServerPlaybackState: r2 = tutils.tflow(resp=True) r2.request.headers["key"] = ["two"] - s = flow.ServerPlaybackState(None, [r, r2], False, False, None, False, None, False) + s = flow.ServerPlaybackState( + None, [ + r, r2], False, False, None, False, None, False) assert s.count() == 2 assert len(s.fmap.keys()) == 1 @@ -173,34 +192,41 @@ class TestServerPlaybackState: r2 = tutils.tflow(resp=True) r2.request.headers["key"] = ["two"] - s = flow.ServerPlaybackState(None, [r, r2], False, True, None, False, None, False) + s = flow.ServerPlaybackState( + None, [ + r, r2], False, True, None, False, None, False) assert s.count() == 2 s.next_flow(r) assert s.count() == 2 - def test_ignore_params(self): - s = flow.ServerPlaybackState(None, [], False, False, ["param1", "param2"], False, None, False) + s = flow.ServerPlaybackState( + None, [], False, False, [ + "param1", "param2"], False, None, False) r = tutils.tflow(resp=True) - r.request.path="/test?param1=1" + r.request.path = "/test?param1=1" r2 = tutils.tflow(resp=True) - r2.request.path="/test" + r2.request.path = "/test" assert s._hash(r) == s._hash(r2) - r2.request.path="/test?param1=2" + r2.request.path = "/test?param1=2" assert s._hash(r) == s._hash(r2) - r2.request.path="/test?param2=1" + r2.request.path = "/test?param2=1" assert s._hash(r) == s._hash(r2) - r2.request.path="/test?param3=2" + r2.request.path = "/test?param3=2" assert not s._hash(r) == s._hash(r2) def test_ignore_payload_params(self): - s = flow.ServerPlaybackState(None, [], False, False, None, False, ["param1", "param2"], False) + s = flow.ServerPlaybackState( + None, [], False, False, None, False, [ + "param1", "param2"], False) r = tutils.tflow(resp=True) - r.request.headers["Content-Type"] = ["application/x-www-form-urlencoded"] + r.request.headers[ + "Content-Type"] = ["application/x-www-form-urlencoded"] r.request.content = "paramx=x¶m1=1" r2 = tutils.tflow(resp=True) - r2.request.headers["Content-Type"] = ["application/x-www-form-urlencoded"] + r2.request.headers[ + "Content-Type"] = ["application/x-www-form-urlencoded"] r2.request.content = "paramx=x¶m1=1" # same parameters assert s._hash(r) == s._hash(r2) @@ -208,20 +234,22 @@ class TestServerPlaybackState: r2.request.content = "paramx=x¶m1=2" assert s._hash(r) == s._hash(r2) # missing parameter - r2.request.content="paramx=x" + r2.request.content = "paramx=x" assert s._hash(r) == s._hash(r2) # ignorable parameter added - r2.request.content="paramx=x¶m1=2" + r2.request.content = "paramx=x¶m1=2" assert s._hash(r) == s._hash(r2) # not ignorable parameter changed - r2.request.content="paramx=y¶m1=1" + r2.request.content = "paramx=y¶m1=1" assert not s._hash(r) == s._hash(r2) # not ignorable parameter missing - r2.request.content="param1=1" + r2.request.content = "param1=1" assert not s._hash(r) == s._hash(r2) def test_ignore_payload_params_other_content_type(self): - s = flow.ServerPlaybackState(None, [], False, False, None, False, ["param1", "param2"], False) + s = flow.ServerPlaybackState( + None, [], False, False, None, False, [ + "param1", "param2"], False) r = tutils.tflow(resp=True) r.request.headers["Content-Type"] = ["application/json"] r.request.content = '{"param1":"1"}' @@ -235,19 +263,31 @@ class TestServerPlaybackState: assert not s._hash(r) == s._hash(r2) def test_ignore_payload_wins_over_params(self): - #NOTE: parameters are mutually exclusive in options - s = flow.ServerPlaybackState(None, [], False, False, None, True, ["param1", "param2"], False) + # NOTE: parameters are mutually exclusive in options + s = flow.ServerPlaybackState( + None, [], False, False, None, True, [ + "param1", "param2"], False) r = tutils.tflow(resp=True) - r.request.headers["Content-Type"] = ["application/x-www-form-urlencoded"] + r.request.headers[ + "Content-Type"] = ["application/x-www-form-urlencoded"] r.request.content = "paramx=y" r2 = tutils.tflow(resp=True) - r2.request.headers["Content-Type"] = ["application/x-www-form-urlencoded"] + r2.request.headers[ + "Content-Type"] = ["application/x-www-form-urlencoded"] r2.request.content = "paramx=x" # same parameters assert s._hash(r) == s._hash(r2) def test_ignore_content(self): - s = flow.ServerPlaybackState(None, [], False, False, None, False, None, False) + s = flow.ServerPlaybackState( + None, + [], + False, + False, + None, + False, + None, + False) r = tutils.tflow(resp=True) r2 = tutils.tflow(resp=True) @@ -257,8 +297,16 @@ class TestServerPlaybackState: r2.request.content = "bar" assert not s._hash(r) == s._hash(r2) - #now ignoring content - s = flow.ServerPlaybackState(None, [], False, False, None, True, None, False) + # now ignoring content + s = flow.ServerPlaybackState( + None, + [], + False, + False, + None, + True, + None, + False) r = tutils.tflow(resp=True) r2 = tutils.tflow(resp=True) r.request.content = "foo" @@ -272,14 +320,22 @@ class TestServerPlaybackState: assert s._hash(r) == s._hash(r2) def test_ignore_host(self): - s = flow.ServerPlaybackState(None, [], False, False, None, False, None, True) + s = flow.ServerPlaybackState( + None, + [], + False, + False, + None, + False, + None, + True) r = tutils.tflow(resp=True) r2 = tutils.tflow(resp=True) - r.request.host="address" - r2.request.host="address" + r.request.host = "address" + r2.request.host = "address" assert s._hash(r) == s._hash(r2) - r2.request.host="wrong_address" + r2.request.host = "wrong_address" assert s._hash(r) == s._hash(r2) @@ -343,12 +399,14 @@ class TestFlow: def test_getset_state(self): f = tutils.tflow(resp=True) state = f.get_state() - assert f.get_state() == protocol.http.HTTPFlow.from_state(state).get_state() + assert f.get_state() == protocol.http.HTTPFlow.from_state( + state).get_state() f.response = None f.error = Error("error") state = f.get_state() - assert f.get_state() == protocol.http.HTTPFlow.from_state(state).get_state() + assert f.get_state() == protocol.http.HTTPFlow.from_state( + state).get_state() f2 = f.copy() f2.id = f.id # copy creates a different uuid @@ -430,7 +488,6 @@ class TestFlow: assert f.response.content == "abarb" - class TestState: def test_backup(self): c = flow.State() @@ -519,7 +576,7 @@ class TestState: assert c.intercept_txt == "~q" assert "Invalid" in c.set_intercept("~") assert not c.set_intercept(None) - assert c.intercept_txt == None + assert c.intercept_txt is None def _add_request(self, state): f = tutils.tflow() @@ -608,7 +665,13 @@ class TestSerialize: def test_load_flows_reverse(self): r = self._treader() s = flow.State() - conf = ProxyConfig(mode="reverse", upstream_server=[True,True,"use-this-domain",80]) + conf = ProxyConfig( + mode="reverse", + upstream_server=[ + True, + True, + "use-this-domain", + 80]) fm = flow.FlowMaster(DummyServer(conf), s) fm.load_flows(r) assert s.flows[0].request.host == "use-this-domain" @@ -630,7 +693,6 @@ class TestSerialize: r = flow.FlowReader(sio) assert len(list(r.stream())) - def test_error(self): sio = StringIO() sio.write("bogus") @@ -661,7 +723,8 @@ class TestFlowMaster: assert not fm.load_script(tutils.test_data.path("scripts/a.py")) assert not fm.unload_scripts() assert fm.load_script("nonexistent") - assert "ValueError" in fm.load_script(tutils.test_data.path("scripts/starterr.py")) + assert "ValueError" in fm.load_script( + tutils.test_data.path("scripts/starterr.py")) assert len(fm.scripts) == 0 def test_getset_ignore(self): @@ -707,14 +770,14 @@ class TestFlowMaster: assert fm.scripts[0].ns["log"][-1] == "request" fm.handle_response(f) assert fm.scripts[0].ns["log"][-1] == "response" - #load second script + # load second script assert not fm.load_script(tutils.test_data.path("scripts/all.py")) assert len(fm.scripts) == 2 fm.handle_clientdisconnect(f.server_conn) assert fm.scripts[0].ns["log"][-1] == "clientdisconnect" assert fm.scripts[1].ns["log"][-1] == "clientdisconnect" - #unload first script + # unload first script fm.unload_scripts() assert len(fm.scripts) == 0 assert not fm.load_script(tutils.test_data.path("scripts/all.py")) @@ -765,7 +828,16 @@ class TestFlowMaster: f = tutils.tflow(resp=True) pb = [tutils.tflow(resp=True), f] fm = flow.FlowMaster(DummyServer(ProxyConfig()), s) - assert not fm.start_server_playback(pb, False, [], False, False, None, False, None, False) + assert not fm.start_server_playback( + pb, + False, + [], + False, + False, + None, + False, + None, + False) assert not fm.start_client_playback(pb, False) fm.client_playback.testing = True @@ -788,16 +860,43 @@ class TestFlowMaster: fm.refresh_server_playback = True assert not fm.do_server_playback(tutils.tflow()) - fm.start_server_playback(pb, False, [], False, False, None, False, None, False) + fm.start_server_playback( + pb, + False, + [], + False, + False, + None, + False, + None, + False) assert fm.do_server_playback(tutils.tflow()) - fm.start_server_playback(pb, False, [], True, False, None, False, None, False) + fm.start_server_playback( + pb, + False, + [], + True, + False, + None, + False, + None, + False) r = tutils.tflow() r.request.content = "gibble" assert not fm.do_server_playback(r) assert fm.do_server_playback(tutils.tflow()) - fm.start_server_playback(pb, False, [], True, False, None, False, None, False) + fm.start_server_playback( + pb, + False, + [], + True, + False, + None, + False, + None, + False) q = Queue.Queue() fm.tick(q, 0) assert fm.should_exit.is_set() @@ -812,7 +911,16 @@ class TestFlowMaster: pb = [f] fm = flow.FlowMaster(None, s) fm.refresh_server_playback = True - fm.start_server_playback(pb, True, [], False, False, None, False, None, False) + fm.start_server_playback( + pb, + True, + [], + False, + False, + None, + False, + None, + False) f = tutils.tflow() f.request.host = "nonexistent" @@ -862,8 +970,9 @@ class TestFlowMaster: def test_stream(self): with tutils.tmpdir() as tdir: p = os.path.join(tdir, "foo") + def r(): - r = flow.FlowReader(open(p,"rb")) + r = flow.FlowReader(open(p, "rb")) return list(r.stream()) s = flow.State() @@ -884,6 +993,7 @@ class TestFlowMaster: assert not r()[1].response + class TestRequest: def test_simple(self): f = tutils.tflow() @@ -919,7 +1029,7 @@ class TestRequest: r.host = "address" r.port = 22 - assert r.url== "https://address:22/path" + assert r.url == "https://address:22/path" assert r.pretty_url(True) == "https://address:22/path" r.headers["Host"] = ["foo.com"] @@ -1062,6 +1172,7 @@ class TestRequest: resp.headers = h assert resp.headers.get_first("content-type") == "text/plain" + class TestResponse: def test_simple(self): f = tutils.tflow(resp=True) @@ -1077,7 +1188,9 @@ class TestResponse: assert resp.size() == len(resp.assemble()) resp.content = CONTENT_MISSING - tutils.raises("Cannot assemble flow with CONTENT_MISSING", resp.assemble) + tutils.raises( + "Cannot assemble flow with CONTENT_MISSING", + resp.assemble) def test_refresh(self): r = tutils.tresp() @@ -1086,14 +1199,15 @@ class TestResponse: pre = r.headers["date"] r.refresh(n) assert pre == r.headers["date"] - r.refresh(n+60) + r.refresh(n + 60) d = email.utils.parsedate_tz(r.headers["date"][0]) d = email.utils.mktime_tz(d) # Weird that this is not exact... - assert abs(60-(d-n)) <= 1 + assert abs(60 - (d - n)) <= 1 - r.headers["set-cookie"] = ["MOO=BAR; Expires=Tue, 08-Mar-2011 00:20:38 GMT; Path=foo.com; Secure"] + r.headers[ + "set-cookie"] = ["MOO=BAR; Expires=Tue, 08-Mar-2011 00:20:38 GMT; Path=foo.com; Secure"] r.refresh() def test_refresh_cookie(self): @@ -1146,7 +1260,7 @@ class TestResponse: def test_header_size(self): r = tutils.tresp() result = len(r._assemble_headers()) - assert result==44 + assert result == 44 def test_get_content_type(self): h = odict.ODictCaseless() @@ -1178,7 +1292,7 @@ class TestClientConnection: c = tutils.tclient_conn() assert ClientConnection.from_state(c.get_state()).get_state() ==\ - c.get_state() + c.get_state() c2 = tutils.tclient_conn() c2.address.address = (c2.address.host, 4242) @@ -1295,7 +1409,6 @@ def test_setheaders(): h.run(f) assert f.request.content == "foo" - h.clear() h.add("~s", "one", "two") h.add("~s", "one", "three") |