Factor out flow unit tests into speparate file.

1 files changed, 96 insertions, 0 deletions

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()
+]