'Test basic interaction with the websocket server'

def test_websocket_basic(self):

    from calibre.srv.web_socket import EchoHandler
    with WSTestServer(EchoHandler) as server:
        simple_test = partial(self.simple_test, server)
        for q in (u'', (u'*' * 125), (u'*' * 126), (u'*' * 127), (u'*' * 128), (u'*' * 65535), (u'*' * 65536), u'Hello-\xb5@\xdf\xf6\xe4\xfc\xe0\xe1-UTF-8!!'):
            simple_test([q], [q])
        for q in ('', ('\xfe' * 125), ('\xfe' * 126), ('\xfe' * 127), ('\xfe' * 128), ('\xfe' * 65535), ('\xfe' * 65536)):
            simple_test([q], [q])
        for payload in [u'', u'ping', '\x00\xff\xfe\xfd\xfc\xfb\x00\xff', ('\xfe' * 125)]:
            simple_test([(PING, payload)], [(PONG, payload)])
        with server.silence_log:
            simple_test([(PING, (u'a' * 126))], close_code=PROTOCOL_ERROR, send_close=False)
        for payload in ('', 'pong'):
            simple_test([(PONG, payload)], [])
        fragments = u'Hello-\xb5@\xdf\xf6\xe4    \xfc\xe0\xe1-UTF-8!!'.split()
        nc = struct.pack('!H', NORMAL_CLOSE)
        with server.silence_log:
            isf_test = partial(simple_test, ignore_send_failures=True)
            for rsv in xrange(1, 7):
                isf_test([{u'rsv': rsv, u'opcode': BINARY}], [], close_code=PROTOCOL_ERROR, send_close=False)
            for opcode in (3, 4, 5, 6, 7, 11, 12, 13, 14, 15):
                isf_test([{u'opcode': opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
            for opcode in (PING, PONG):
                isf_test([{u'opcode': opcode, u'payload': u'f1', u'fin': 0}, {u'opcode': opcode, u'payload': u'f2'}], close_code=PROTOCOL_ERROR, send_close=False)
            isf_test([(CLOSE, (nc + ('x' * 124)))], send_close=False, close_code=PROTOCOL_ERROR)
            for fin in (0, 1):
                isf_test([{u'opcode': 0, u'fin': fin, u'payload': 'non-continuation    frame'}, u'some    text'], close_code=PROTOCOL_ERROR, send_close=False)
            isf_test([{u'opcode': TEXT, u'payload': fragments[0], u'fin': 0}, {u'opcode': CONTINUATION, u'payload': fragments[1]}, {u'opcode': 0, u'fin': 0}], [u''.join(fragments)], close_code=PROTOCOL_ERROR, send_close=False)
            isf_test([{u'opcode': TEXT, u'payload': fragments[0], u'fin': 0}, {u'opcode': TEXT, u'payload': fragments[1]}], close_code=PROTOCOL_ERROR, send_close=False)
            frags = []
            for payload in ('\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5', '\xed\xa0\x80', '\x80edited'):
                frags.append({u'opcode': (CONTINUATION if frags else TEXT), u'fin': (1 if (len(frags) == 2) else 0), u'payload': payload})
            isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False)
            (frags, q) = ([], '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5\xed\xa0\x80\x80edited')
            for (i, b) in enumerate(q):
                frags.append({u'opcode': (TEXT if (i == 0) else CONTINUATION), u'fin': (1 if (i == (len(q) - 1)) else 0), u'payload': b})
            isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False, ignore_send_failures=True)
            for q in ('\xce', '\xce\xba\xe1'):
                isf_test([{u'opcode': TEXT, u'payload': q}], close_code=INCONSISTENT_DATA, send_close=False)
        simple_test([{u'opcode': TEXT, u'payload': fragments[0], u'fin': 0}, {u'opcode': CONTINUATION, u'payload': fragments[1]}], [u''.join(fragments)])
        simple_test([{u'opcode': TEXT, u'payload': fragments[0], u'fin': 0}, (PING, 'pong'), {u'opcode': CONTINUATION, u'payload': fragments[1]}], [(PONG, 'pong'), u''.join(fragments)])
        fragments = u'12345'
        simple_test([{u'opcode': TEXT, u'payload': fragments[0], u'fin': 0}, {u'opcode': CONTINUATION, u'payload': fragments[1], u'fin': 0}, (PING, '1'), {u'opcode': CONTINUATION, u'payload': fragments[2], u'fin': 0}, {u'opcode': CONTINUATION, u'payload': fragments[3], u'fin': 0}, (PING, '2'), {u'opcode': CONTINUATION, u'payload': fragments[4]}], [(PONG, '1'), (PONG, '2'), fragments])
        simple_test([{u'opcode': TEXT, u'fin': 0}, {u'opcode': CONTINUATION, u'fin': 0}, {u'opcode': CONTINUATION}], [u''])
        simple_test([{u'opcode': TEXT, u'fin': 0}, {u'opcode': CONTINUATION, u'fin': 0, u'payload': u'x'}, {u'opcode': CONTINUATION}], [u'x'])
        for q in ('\xc2\xb5', '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5', u'Hello-\xb5@\xdf\xf6\xe4\xfc\xe0\xe1-UTF-8!!'.encode(u'utf-8')):
            frags = []
            for (i, b) in enumerate(q):
                frags.append({u'opcode': (TEXT if (i == 0) else CONTINUATION), u'fin': (1 if (i == (len(q) - 1)) else 0), u'payload': b})
            simple_test(frags, [q.decode(u'utf-8')])
        simple_test([(CLOSE, nc), (CLOSE, '\x01\x01')], send_close=False)
        simple_test([(CLOSE, nc), (PING, 'ping')], send_close=False)
        simple_test([(CLOSE, nc), u'xxx'], send_close=False)
        simple_test([{u'opcode': TEXT, u'payload': u'xxx', u'fin': 0}, (CLOSE, nc), {u'opcode': CONTINUATION, u'payload': u'yyy'}], send_close=False)
        simple_test([(CLOSE, '')], send_close=False)
        simple_test([(CLOSE, '\x01')], send_close=False, close_code=PROTOCOL_ERROR)
        simple_test([(CLOSE, (nc + ('x' * 123)))], send_close=False)
        simple_test([(CLOSE, (nc + 'a\x80\x80'))], send_close=False, close_code=PROTOCOL_ERROR)
        for code in (1000, 1001, 1002, 1003, 1007, 1008, 1009, 1010, 1011, 3000, 3999, 4000, 4999):
            simple_test([(CLOSE, struct.pack('!H', code))], send_close=False, close_code=code)
        for code in (0, 999, 1004, 1005, 1006, 1012, 1013, 1014, 1015, 1016, 1100, 2000, 2999):
            simple_test([(CLOSE, struct.pack('!H', code))], send_close=False, close_code=PROTOCOL_ERROR)
