| 122 | |
| 123 | |
| 124 | def assert_cat_socket_detached_with_keys(sock, inputs): |
| 125 | if hasattr(sock, '_sock'): |
| 126 | sock = sock._sock |
| 127 | |
| 128 | for i in inputs: |
| 129 | sock.sendall(i) |
| 130 | time.sleep(0.5) |
| 131 | |
| 132 | # If we're using a Unix socket, the sock.send call will fail with a |
| 133 | # BrokenPipeError ; INET sockets will just stop receiving / sending data |
| 134 | # but will not raise an error |
| 135 | if isinstance(sock, paramiko.Channel): |
| 136 | with pytest.raises(OSError): |
| 137 | sock.sendall(b'make sure the socket is closed\n') |
| 138 | else: |
| 139 | if getattr(sock, 'family', -9) == getattr(socket, 'AF_UNIX', -1): |
| 140 | # We do not want to use pytest.raises here because future versions |
| 141 | # of the daemon no longer cause this to raise an error. |
| 142 | try: |
| 143 | sock.sendall(b'make sure the socket is closed\n') |
| 144 | except OSError: |
| 145 | return |
| 146 | |
| 147 | sock.sendall(b"make sure the socket is closed\n") |
| 148 | data = sock.recv(128) |
| 149 | # New in 18.06: error message is broadcast over the socket when reading |
| 150 | # after detach |
| 151 | assert data == b'' or data.startswith( |
| 152 | b'exec attach failed: error on attach stdin: read escape sequence' |
| 153 | ) |
| 154 | |
| 155 | |
| 156 | def ctrl_with(char): |