tests++

mitmproxy/addons/proxyserver.py

@@ -8,7 +8,7 @@ from mitmproxy.proxy import commands, events, server_hooks
 from mitmproxy.proxy import server
 from mitmproxy.proxy.layers.tcp import TcpMessageInjected
 from mitmproxy.proxy.layers.websocket import WebSocketMessageInjected
-from mitmproxy.utils import asyncio_utils, human, strutils
+from mitmproxy.utils import asyncio_utils, human
 from wsproto.frame_protocol import Opcode
 
 

test/mitmproxy/addons/test_proxyserver.py

@@ -90,7 +90,7 @@ async def test_start_stop():
 
 
 @pytest.mark.asyncio
-async def test_inject():
+async def test_inject() -> None:
     async def server_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
         while s := await reader.read(1):
             writer.write(s.upper())
@@ -112,39 +112,39 @@ async def test_inject():
 
             writer.write(b"a")
             assert await reader.read(1) == b"A"
-            ps.inject_tcp(state.flows[0], False, "b")
+            ps.inject_tcp(state.flows[0], False, b"b")
             assert await reader.read(1) == b"B"
-            ps.inject_tcp(state.flows[0], True, "c")
+            ps.inject_tcp(state.flows[0], True, b"c")
             assert await reader.read(1) == b"c"
 
 
 @pytest.mark.asyncio
-async def test_inject_fail():
+async def test_inject_fail() -> None:
     ps = Proxyserver()
     with taddons.context(ps) as tctx:
         ps.inject_websocket(
             tflow.tflow(),
             True,
-            "test"
+            b"test"
         )
         await tctx.master.await_log("Cannot inject WebSocket messages into non-WebSocket flows.", level="warn")
         ps.inject_tcp(
             tflow.tflow(),
             True,
-            "test"
+            b"test"
         )
         await tctx.master.await_log("Cannot inject TCP messages into non-TCP flows.", level="warn")
 
         ps.inject_websocket(
             tflow.twebsocketflow(),
             True,
-            "test"
+            b"test"
         )
         await tctx.master.await_log("Flow is not from a live connection.", level="warn")
         ps.inject_websocket(
             tflow.ttcpflow(),
             True,
-            "test"
+            b"test"
         )
         await tctx.master.await_log("Flow is not from a live connection.", level="warn")
 

test/mitmproxy/test_types.py

@@ -42,6 +42,17 @@ def test_str():
         assert b.parse(tctx.master.commands, str, "foo") == "foo"
 
 
+def test_bytes():
+    with taddons.context() as tctx:
+        b = mitmproxy.types._BytesType()
+        assert b.is_valid(tctx.master.commands, bytes, b"foo") is True
+        assert b.is_valid(tctx.master.commands, bytes, 1) is False
+        assert b.completion(tctx.master.commands, bytes, "") == []
+        assert b.parse(tctx.master.commands, bytes, "foo") == b"foo"
+        with pytest.raises(mitmproxy.exceptions.TypeError):
+            b.parse(tctx.master.commands, bytes, "incomplete escape sequence\\")
+
+
 def test_unknown():
     with taddons.context() as tctx:
         b = mitmproxy.types._UnknownType()