(result, monkeypatch)
| 34 | |
| 35 | @pytest.mark.parametrize("result", ("success", "killed", "failed")) |
| 36 | async def test_open_connection(result, monkeypatch): |
| 37 | handler = MockConnectionHandler() |
| 38 | server_connect = handler.hook_handlers["server_connect"] |
| 39 | server_connected = handler.hook_handlers["server_connected"] |
| 40 | server_connect_error = handler.hook_handlers["server_connect_error"] |
| 41 | server_disconnected = handler.hook_handlers["server_disconnected"] |
| 42 | |
| 43 | match result: |
| 44 | case "success": |
| 45 | monkeypatch.setattr( |
| 46 | asyncio, |
| 47 | "open_connection", |
| 48 | mock.AsyncMock(return_value=(mock.MagicMock(), mock.MagicMock())), |
| 49 | ) |
| 50 | monkeypatch.setattr( |
| 51 | MockConnectionHandler, "handle_connection", mock.AsyncMock() |
| 52 | ) |
| 53 | case "failed": |
| 54 | monkeypatch.setattr( |
| 55 | asyncio, "open_connection", mock.AsyncMock(side_effect=OSError) |
| 56 | ) |
| 57 | case "killed": |
| 58 | |
| 59 | def _kill(d: server_hooks.ServerConnectionHookData) -> None: |
| 60 | d.server.error = "do not connect" |
| 61 | |
| 62 | server_connect.side_effect = _kill |
| 63 | |
| 64 | await handler.open_connection( |
| 65 | commands.OpenConnection(connection=Server(address=("server", 1234))) |
| 66 | ) |
| 67 | |
| 68 | assert server_connect.call_args[0][0].server.address == ("server", 1234) |
| 69 | |
| 70 | assert server_connected.called == (result == "success") |
| 71 | assert server_connect_error.called == (result != "success") |
| 72 | |
| 73 | assert server_disconnected.called == (result == "success") |
| 74 | |
| 75 | |
| 76 | async def test_no_reentrancy(capsys): |
nothing calls this directly
no test coverage detected
searching dependent graphs…