websocket_test.py 文件源码

def test_getting_messages_from_websocket_76(self):
        connect = [
            "GET /range HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        resp = sock.recv(1024)
        headers, result = resp.split(b'\r\n\r\n')
        msgs = [result[16:].strip(b'\x00\xff')]
        cnt = 10
        while cnt:
            msgs.append(sock.recv(20).strip(b'\x00\xff'))
            cnt -= 1
        # Last item in msgs is an empty string
        self.assertEqual(msgs[:-1], [six.b('msg %d' % i) for i in range(10)])