Function test_inject

test/mitmproxy/addons/test_proxyserver.py:128–163

()

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

    ps = Proxyserver()
    nl = NextLayer()
    state = HelperAddon()

    with taddons.context(ps, nl, state) as tctx:
        async with tcp_server(server_handler) as addr:
            tctx.configure(ps, listen_host="127.0.0.1", listen_port=0)
            assert await ps.setup_servers()
            ps.running()
            proxy_addr = ps.servers["regular"].listen_addrs[0]
            reader, writer = await asyncio.open_connection(*proxy_addr)

            req = f"CONNECT {addr[0]}:{addr[1]} HTTP/1.1\r\n\r\n"
            writer.write(req.encode())
            assert (
                await reader.readuntil(b"\r\n\r\n")
                == b"HTTP/1.1 200 Connection established\r\n\r\n"
            )

            writer.write(b"a")
            assert await reader.read(1) == b"A"
            ps.inject_tcp(state.flows[0], False, b"b")
            assert await reader.read(1) == b"B"
            ps.inject_tcp(state.flows[0], True, b"c")
            assert await reader.read(1) == b"c"

            writer.close()
            await writer.wait_closed()
            await _wait_for_connection_closes(ps)


async def test_inject_fail(caplog) -> None: