diff options
Diffstat (limited to 'test/test_flow.py')
-rw-r--r-- | test/test_flow.py | 142 |
1 files changed, 140 insertions, 2 deletions
diff --git a/test/test_flow.py b/test/test_flow.py index 470547a5..59b277e4 100644 --- a/test/test_flow.py +++ b/test/test_flow.py @@ -26,6 +26,10 @@ class uFlow(libpry.AutoTree): f.request = f.response.request assert not f.match(filt.parse("~b test")) + def test_dump(self): + f = utils.tflow() + assert f.dump() + def test_backup(self): f = utils.tflow() assert not f.modified() @@ -59,7 +63,6 @@ class uFlow(libpry.AutoTree): assert console.format_flow(f, True) assert console.format_flow(f, False) - f.focus = True assert console.format_flow(f, True) assert console.format_flow(f, False) @@ -110,6 +113,141 @@ class uFlow(libpry.AutoTree): f.request = utils.treq() +class uState(libpry.AutoTree): + def test_backup(self): + bc = proxy.BrowserConnection("address", 22) + c = flow.State() + f = flow.Flow(bc) + c.add_browserconnect(f) + + f.backup() + c.revert(f) + + def test_flow(self): + """ + normal flow: + + connect -> request -> response + """ + bc = proxy.BrowserConnection("address", 22) + c = flow.State() + f = flow.Flow(bc) + c.add_browserconnect(f) + assert c.lookup(bc) + + req = utils.treq(bc) + assert c.add_request(req) + assert len(c.flow_list) == 1 + assert c.lookup(req) + + newreq = utils.treq() + assert not c.add_request(newreq) + assert not c.lookup(newreq) + + resp = utils.tresp(req) + assert c.add_response(resp) + assert len(c.flow_list) == 1 + assert c.lookup(resp) + + newresp = utils.tresp() + assert not c.add_response(newresp) + assert not c.lookup(newresp) + + def test_err(self): + bc = proxy.BrowserConnection("address", 22) + c = flow.State() + f = flow.Flow(bc) + c.add_browserconnect(f) + e = proxy.Error(bc, "message") + assert c.add_error(e) + + e = proxy.Error(proxy.BrowserConnection("address", 22), "message") + assert not c.add_error(e) + + def test_view(self): + c = flow.State() + + f = utils.tflow() + c.add_browserconnect(f) + assert len(c.view) == 1 + c.set_limit(filt.parse("~q")) + assert len(c.view) == 0 + c.set_limit(None) + + + f = utils.tflow() + req = utils.treq(f.connection) + c.add_browserconnect(f) + c.add_request(req) + assert len(c.view) == 2 + c.set_limit(filt.parse("~q")) + assert len(c.view) == 1 + c.set_limit(filt.parse("~s")) + assert len(c.view) == 0 + + def _add_request(self, state): + f = utils.tflow() + state.add_browserconnect(f) + q = utils.treq(f.connection) + state.add_request(q) + return f + + def _add_response(self, state): + f = self._add_request(state) + r = utils.tresp(f.request) + state.add_response(r) + + def _add_error(self, state): + f = utils.tflow() + f.error = proxy.Error(None, "msg") + state.add_browserconnect(f) + q = utils.treq(f.connection) + state.add_request(q) + + def test_kill_flow(self): + c = flow.State() + f = utils.tflow() + c.add_browserconnect(f) + c.kill_flow(f) + assert not c.flow_list + + def test_clear(self): + c = flow.State() + f = utils.tflow() + c.add_browserconnect(f) + f.intercepting = True + + c.clear() + assert len(c.flow_list) == 1 + f.intercepting = False + c.clear() + assert len(c.flow_list) == 0 + + def test_dump_flows(self): + c = flow.State() + self._add_request(c) + self._add_response(c) + self._add_request(c) + self._add_response(c) + self._add_request(c) + self._add_response(c) + self._add_error(c) + + dump = c.dump_flows() + c.clear() + c.load_flows(dump) + assert isinstance(c.flow_list[0], flow.Flow) + + def test_accept_all(self): + c = flow.State() + self._add_request(c) + self._add_response(c) + self._add_request(c) + c.accept_all() + + + tests = [ - uFlow() + uFlow(), + uState(), ] |