(tctx: Context, success)
| 95 | |
| 96 | @pytest.mark.parametrize("success", ["success", "fail"]) |
| 97 | def test_tunnel_handshake_command(tctx: Context, success): |
| 98 | server = Server(address=("proxy", 1234)) |
| 99 | |
| 100 | tl = TTunnelLayer(tctx, server, tctx.server) |
| 101 | tl.child_layer = TChildLayer(tctx) |
| 102 | |
| 103 | playbook = Playbook(tl, logs=True) |
| 104 | ( |
| 105 | playbook |
| 106 | << Log("Got start. Server state: CLOSED") |
| 107 | >> DataReceived(tctx.client, b"client-hello") |
| 108 | << SendData(tctx.client, b"client-hello-reply") |
| 109 | >> DataReceived(tctx.client, b"open") |
| 110 | << OpenConnection(server) |
| 111 | >> reply(None) |
| 112 | << SendData(server, b"handshake-hello") |
| 113 | >> DataReceived(server, b"handshake-" + success.encode()) |
| 114 | << SendData(server, b"handshake-" + success.encode()) |
| 115 | ) |
| 116 | if success == "success": |
| 117 | assert ( |
| 118 | playbook |
| 119 | << Log(f"Opened: err=None. Server state: OPEN") |
| 120 | >> DataReceived(server, b"tunneled-server-hello") |
| 121 | << SendData(server, b"tunneled-server-hello-reply") |
| 122 | >> ConnectionClosed(tctx.client) |
| 123 | << Log("Got client close.") |
| 124 | << CloseConnection(tctx.client) |
| 125 | ) |
| 126 | assert tl.tunnel_state is tunnel.TunnelState.OPEN |
| 127 | assert ( |
| 128 | playbook |
| 129 | >> ConnectionClosed(server) |
| 130 | << Log("Got server close.") |
| 131 | << CloseConnection(server) |
| 132 | ) |
| 133 | assert tl.tunnel_state is tunnel.TunnelState.CLOSED |
| 134 | else: |
| 135 | assert ( |
| 136 | playbook |
| 137 | << CloseConnection(server) |
| 138 | << Log("Opened: err='handshake error'. Server state: CLOSED") |
| 139 | ) |
| 140 | assert tl.tunnel_state is tunnel.TunnelState.CLOSED |
| 141 | |
| 142 | |
| 143 | def test_tunnel_default_impls(tctx: Context): |
nothing calls this directly
no test coverage detected
searching dependent graphs…