Add a simple server playback state object.

We use a loose hash to match incoming requests with recorded flows. At the
moment, this hash is over the host, port, scheme, method, path and content of
the request. Note that headers are not included here - if we do want to include
headers, we would have to do some work to normalize them to remove variations
between user agents, header order, etc. etc.

1 files changed, 38 insertions, 0 deletions

diff --git a/test/test_flow.py b/test/test_flow.py
index 7714ad97..adfeda6e 100644
--- a/test/test_flow.py
+++ b/test/test_flow.py
@@ -3,6 +3,43 @@ from libmproxy import console, proxy, filt, flow
 import utils
 import libpry
 
+
+class uServerPlaybackState(libpry.AutoTree):
+    def test_hash(self):
+        s = flow.ServerPlaybackState()
+        r = utils.tflow()
+        r2 = utils.tflow()
+
+        assert s._hash(r)
+        assert s._hash(r) == s._hash(r2)
+        r.request.headers["foo"] = ["bar"]
+        assert s._hash(r) == s._hash(r2)
+        r.request.path = "voing"
+        assert s._hash(r) != s._hash(r2)
+
+    def test_load(self):
+        s = flow.ServerPlaybackState()
+        r = utils.tflow()
+        r.request.headers["key"] = ["one"]
+
+        r2 = utils.tflow()
+        r2.request.headers["key"] = ["two"]
+
+        s.load([r, r2])
+        assert len(s) == 2
+        assert len(s.fmap.keys()) == 1
+
+        n = s.next_flow(r)
+        assert n.request.headers["key"] == ["one"]
+        assert len(s) == 1
+
+        n = s.next_flow(r)
+        assert n.request.headers["key"] == ["two"]
+        assert len(s) == 0
+
+        assert not s.next_flow(r)
+
+
 class uFlow(libpry.AutoTree):
     def test_run_script(self):
         f = utils.tflow()
@@ -275,6 +312,7 @@ class uFlowMaster(libpry.AutoTree):
 
 
 tests = [
+    uServerPlaybackState(),
     uFlow(),
     uState(),
     uSerialize(),