add tests

mitmproxy/contrib/wsproto/__init__.py

@@ -0,0 +1,13 @@
+from . import compat
+from . import connection
+from . import events
+from . import extensions
+from . import frame_protocol
+
+__all__ = [
+    'compat',
+    'connection',
+    'events',
+    'extensions',
+    'frame_protocol',
+]

mitmproxy/test/tflow.py

@@ -75,7 +75,9 @@ def twebsocketflow(client_conn=True, server_conn=True, messages=True, err=None,
         handshake_flow.response = resp
 
     f = websocket.WebSocketFlow(client_conn, server_conn, handshake_flow)
-    handshake_flow.metadata['websocket_flow'] = f
+    f.metadata['websocket_handshake'] = handshake_flow.id
+    handshake_flow.metadata['websocket_flow'] = f.id
+    handshake_flow.metadata['websocket'] = True
 
     if messages is True:
         messages = [

mitmproxy/websocket.py

@@ -13,7 +13,7 @@ class WebSocketMessage(serializable.Serializable):
     def __init__(
         self, type: int, from_client: bool, content: bytes, timestamp: Optional[int]=None
     ) -> None:
-        self.type = wsproto.frame_protocol.Opcode(type)
+        self.type = wsproto.frame_protocol.Opcode(type)  # type: ignore
         self.from_client = from_client
         self.content = content
         self.timestamp = timestamp or int(time.time())  # type: int

test/mitmproxy/addons/test_save.py

@@ -52,6 +52,7 @@ def test_websocket(tmpdir):
 
         f = tflow.twebsocketflow()
         sa.websocket_start(f)
+        sa.websocket_end(f)
         tctx.configure(sa, save_stream_file=None)
         assert rd(p)
 

test/mitmproxy/test_flow.py

@@ -97,7 +97,7 @@ class TestSerialize:
 
 
 class TestFlowMaster:
-    def test_load_flow_reverse(self):
+    def test_load_http_flow_reverse(self):
         s = tservers.TestState()
         opts = options.Options(
             mode="reverse:https://use-this-domain"
@@ -108,6 +108,20 @@ class TestFlowMaster:
         fm.load_flow(f)
         assert s.flows[0].request.host == "use-this-domain"
 
+    def test_load_websocket_flow(self):
+        s = tservers.TestState()
+        opts = options.Options(
+            mode="reverse:https://use-this-domain"
+        )
+        fm = master.Master(opts)
+        fm.addons.add(s)
+        f = tflow.twebsocketflow()
+        fm.load_flow(f.handshake_flow)
+        fm.load_flow(f)
+        assert s.flows[0].request.host == "use-this-domain"
+        assert s.flows[1].handshake_flow == f.handshake_flow
+        assert len(s.flows[1].messages) == len(f.messages)
+
     def test_replay(self):
         opts = options.Options()
         fm = master.Master(opts)