From edb8228dd27c3050f45b536b761eab46672f8eb3 Mon Sep 17 00:00:00 2001 From: Aldo Cortesi Date: Mon, 31 Jan 2011 11:48:53 +1300 Subject: Factor out flow unit tests into speparate file. --- test/test_flow.py | 96 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 test/test_flow.py (limited to 'test/test_flow.py') diff --git a/test/test_flow.py b/test/test_flow.py new file mode 100644 index 00000000..1867616e --- /dev/null +++ b/test/test_flow.py @@ -0,0 +1,96 @@ +from libmproxy import console, proxy, filt, flow +import utils +import libpry + +class uFlow(libpry.AutoTree): + def test_match(self): + f = utils.tflow() + f.response = utils.tresp() + f.request = f.response.request + assert not f.match(filt.parse("~b test")) + + def test_backup(self): + f = utils.tflow() + f.backup() + f.revert() + + def test_getset_state(self): + f = utils.tflow() + state = f.get_state() + assert f == flow.Flow.from_state(state) + f.response = utils.tresp() + f.request = f.response.request + state = f.get_state() + assert f == flow.Flow.from_state(state) + + def test_simple(self): + f = utils.tflow() + assert console.format_flow(f, True) + assert console.format_flow(f, False) + + f.request = utils.treq() + assert console.format_flow(f, True) + assert console.format_flow(f, False) + + f.response = utils.tresp() + f.response.headers["content-type"] = ["text/html"] + assert console.format_flow(f, True) + assert console.format_flow(f, False) + f.response.code = 404 + 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) + + f.connection = flow.ReplayConnection() + assert console.format_flow(f, True) + assert console.format_flow(f, False) + + f.response = None + assert console.format_flow(f, True) + assert console.format_flow(f, False) + + f.error = proxy.Error(200, "test") + assert console.format_flow(f, True) + assert console.format_flow(f, False) + + def test_kill(self): + f = utils.tflow() + f.request = utils.treq() + f.intercept() + assert not f.request.acked + f.kill() + assert f.request.acked + f.intercept() + f.response = utils.tresp() + f.request = f.response.request + f.request.ack() + assert not f.response.acked + f.kill() + assert f.response.acked + + def test_accept_intercept(self): + f = utils.tflow() + f.request = utils.treq() + f.intercept() + assert not f.request.acked + f.accept_intercept() + assert f.request.acked + f.response = utils.tresp() + f.request = f.response.request + f.intercept() + f.request.ack() + assert not f.response.acked + f.accept_intercept() + assert f.response.acked + + def test_serialization(self): + f = flow.Flow(None) + f.request = utils.treq() + + +tests = [ + uFlow() +] -- cgit v1.2.3