Unit test++.

libmproxy/flow.py

@@ -1096,11 +1096,8 @@ class State(object):
             Add an error response to the state. Returns the matching flow, or
             None if there isn't one.
         """
-        if err.request:
+        f = self._flow_map.get(err.request)
-            f = self._flow_map.get(err.request)
+        if not f:
-            if not f:
-                return None
-        else:
             return None
         f.error = err
         if f.match(self._limit) and not f in self.view:

test/scripts/duplicate_flow.py

@@ -0,0 +1,5 @@
+
+def request(ctx, f):
+    f = ctx.duplicate_flow(f)
+    ctx.replay_request(f)
+

test/test_flow.py

@@ -143,6 +143,17 @@ class uFlow(libpry.AutoTree):
         assert not f.request is f2.request
         assert f.request.headers == f2.request.headers
         assert not f.request.headers is f2.request.headers
+        assert f.response == f2.response
+        assert not f.response is f2.response
+
+        f = tutils.tflow_err()
+        f2 = f.copy()
+        assert not f is f2
+        assert not f.request is f2.request
+        assert f.request.headers == f2.request.headers
+        assert not f.request.headers is f2.request.headers
+        assert f.error == f2.error
+        assert not f.error is f2.error
 
     def test_match(self):
         f = tutils.tflow()
@@ -301,7 +312,6 @@ class uState(libpry.AutoTree):
         assert c.add_response(resp)
         assert c.active_flow_count() == 0
 
-
     def test_err(self):
         c = flow.State()
         req = tutils.treq()
@@ -322,7 +332,6 @@ class uState(libpry.AutoTree):
         assert c.add_error(e)
         assert c.view
 
-
     def test_set_limit(self):
         c = flow.State()
 

test/test_proxy.py

@@ -24,6 +24,7 @@ class u_read_chunked(libpry.AutoTree):
 
 class Dummy: pass
 
+
 class u_read_http_body(libpry.AutoTree):
     def test_all(self):
 
@@ -42,7 +43,6 @@ class u_read_http_body(libpry.AutoTree):
         s = cStringIO.StringIO("testing")
         libpry.raises(proxy.ProxyError, proxy.read_http_body, s, d, h, False, 4)
 
-
         h = flow.ODict()
         s = cStringIO.StringIO("testing")
         assert len(proxy.read_http_body(s, d, h, True, 4)) == 4

test/test_script.py

@@ -1,6 +1,7 @@
 import os
 from libmproxy import script, flow
 import libpry
+import tutils
 
 class uScript(libpry.AutoTree):
     def test_simple(self):
@@ -22,6 +23,16 @@ class uScript(libpry.AutoTree):
         p.load()
         assert p.run("here") == (True, 1)
 
+    def test_duplicate_flow(self):
+        s = flow.State()
+        fm = flow.FlowMaster(None, s)
+        fm.load_script(os.path.join("scripts", "duplicate_flow.py"))
+        r = tutils.treq()
+        fm.handle_request(r)
+        assert fm.state.flow_count() == 2
+        assert not fm.state.view[0].request.is_replay()
+        assert fm.state.view[1].request.is_replay()
+
     def test_err(self):
         s = flow.State()
         fm = flow.FlowMaster(None, s)