(tdata, mode, concurrency)
| 57 | @pytest.mark.parametrize("mode", ["http", "https", "upstream", "err"]) |
| 58 | @pytest.mark.parametrize("concurrency", [-1, 1]) |
| 59 | async def test_playback(tdata, mode, concurrency): |
| 60 | async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): |
| 61 | if mode == "err": |
| 62 | return |
| 63 | req = await reader.readline() |
| 64 | if mode == "upstream": |
| 65 | assert req == b"GET http://address:22/path HTTP/1.1\r\n" |
| 66 | else: |
| 67 | assert req == b"GET /path HTTP/1.1\r\n" |
| 68 | req = await reader.readuntil(b"data") |
| 69 | assert req == ( |
| 70 | b"header: qvalue\r\n" |
| 71 | b"content-length: 4\r\nHost: example.mitmproxy.org\r\n\r\n" |
| 72 | b"data" |
| 73 | ) |
| 74 | writer.write(b"HTTP/1.1 204 No Content\r\n\r\n") |
| 75 | await writer.drain() |
| 76 | assert not await reader.read() |
| 77 | |
| 78 | cp = ClientPlayback() |
| 79 | ps = Proxyserver() |
| 80 | tls = TlsConfig() |
| 81 | with taddons.context(cp, ps, tls) as tctx: |
| 82 | tctx.configure(cp, client_replay_concurrency=concurrency) |
| 83 | |
| 84 | server_args = {} |
| 85 | if mode == "https": |
| 86 | server_args["ssl"] = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) |
| 87 | server_args["ssl"].load_cert_chain( |
| 88 | certfile=tdata.path( |
| 89 | "mitmproxy/net/data/verificationcerts/trusted-leaf.crt" |
| 90 | ), |
| 91 | keyfile=tdata.path( |
| 92 | "mitmproxy/net/data/verificationcerts/trusted-leaf.key" |
| 93 | ), |
| 94 | ) |
| 95 | tctx.configure( |
| 96 | tls, |
| 97 | ssl_verify_upstream_trusted_ca=tdata.path( |
| 98 | "mitmproxy/net/data/verificationcerts/trusted-root.crt" |
| 99 | ), |
| 100 | ) |
| 101 | |
| 102 | async with tcp_server(handler, **server_args) as addr: |
| 103 | cp.running() |
| 104 | flow = tflow.tflow(live=False) |
| 105 | flow.request.content = b"data" |
| 106 | if mode == "upstream": |
| 107 | tctx.options.mode = [f"upstream:http://{addr[0]}:{addr[1]}"] |
| 108 | flow.request.authority = f"{addr[0]}:{addr[1]}" |
| 109 | flow.request.host, flow.request.port = "address", 22 |
| 110 | else: |
| 111 | flow.request.host, flow.request.port = addr |
| 112 | if mode == "https": |
| 113 | flow.request.scheme = "https" |
| 114 | # Used for SNI |
| 115 | flow.request.host_header = "example.mitmproxy.org" |
| 116 | cp.start_replay([flow]) |
nothing calls this directly
no test coverage detected
searching dependent graphs…