

def test_tunnel(self):
    print('Setting up remote listener...')
    self.remote_server_handle = self.context.Popen(['python', '-c', '"{0}"'.format(HELLO_SERVER_CMD)], stdout=subprocess.PIPE)
    print('Setting up tunnel')
    with self.context.tunnel(2135, 2134):
        print('Tunnel up!')
        server_output = self.remote_server_handle.stdout.read(5)
        self.assertEqual(server_output, b'ready')
        print('Connecting to server via tunnel')
        s = socket.socket()
        s.connect(('localhost', 2135))
        print('Receiving...')
        response = s.recv(5)
        self.assertEqual(response, b'hello')
        print('Closing connection')
        s.close()
        print('Waiting for listener...')
        (output, _) = self.remote_server_handle.communicate()
        self.assertEqual(self.remote_server_handle.returncode, 0)
        print('Closing tunnel')

dataset/ETHPy150Open spotify/luigi/test/contrib/test_ssh.py TestRemoteContext.test_tunnel/original