(failure, monkeypatch, caplog_async)
| 136 | |
| 137 | @pytest.mark.parametrize("failure", [True, False]) |
| 138 | async def test_transparent(failure, monkeypatch, caplog_async): |
| 139 | caplog_async.set_level("INFO") |
| 140 | manager = MagicMock() |
| 141 | |
| 142 | if failure: |
| 143 | monkeypatch.setattr(mitmproxy.platform, "original_addr", None) |
| 144 | else: |
| 145 | monkeypatch.setattr( |
| 146 | mitmproxy.platform, "original_addr", lambda s: ("address", 42) |
| 147 | ) |
| 148 | |
| 149 | with taddons.context(Proxyserver()) as tctx: |
| 150 | tctx.options.connection_strategy = "lazy" |
| 151 | inst = ServerInstance.make("transparent@127.0.0.1:0", manager) |
| 152 | await inst.start() |
| 153 | await caplog_async.await_log("listening") |
| 154 | |
| 155 | host, port, *_ = inst.listen_addrs[0] |
| 156 | reader, writer = await asyncio.open_connection(host, port) |
| 157 | |
| 158 | if failure: |
| 159 | assert await caplog_async.await_log("Transparent mode failure") |
| 160 | writer.close() |
| 161 | await writer.wait_closed() |
| 162 | else: |
| 163 | assert await caplog_async.await_log("client connect") |
| 164 | writer.close() |
| 165 | await writer.wait_closed() |
| 166 | assert await caplog_async.await_log("client disconnect") |
| 167 | |
| 168 | await inst.stop() |
| 169 | assert await caplog_async.await_log("stopped") |
| 170 | |
| 171 | |
| 172 | async def _echo_server(self: ConnectionHandler): |
nothing calls this directly
no test coverage detected
searching dependent graphs…