How to use websockets - 10 common examples

async def test_websocket_failure_receive(liveclient):

    @liveclient.app.route('/failure', protocol="websocket")
    async def failme(request, ws):
        raise NotImplementedError('OUCH')

    websocket = await websockets.connect(liveclient.wsl + '/failure')
    with pytest.raises(websockets.exceptions.ConnectionClosed):
        # Receiving, on the other hand, will raise immediatly an
        # error as the reader is closed. Only the writer is opened
        # as we are expected to send back the closing frame.
        await websocket.recv()

    await websocket.close()
    assert websocket.state == websockets.protocol.State.CLOSED
    assert websocket.close_code == 1011
    assert websocket.close_reason == 'Handler died prematurely.'

def main():
    c_source = asn1tools.compile_files([
        'asn1/c_source.asn',
        'asn1/programming_types.asn'
    ], 'oer')

    start_server = websockets.serve(
        functools.partial(echo, c_source=c_source),
        'localhost',
        8765)

    print('Listening for websocket clients on localhost:8765...')

    asyncio.get_event_loop().run_until_complete(start_server)
    asyncio.get_event_loop().run_forever()

async def test_explicit_permanent_key_unavailable(
            self, server_no_key, server, client_factory
    ):
        """
        Check that the server rejects a permanent key if the server
        has none.
        """
        key = libnacl.public.SecretKey()

        # Expect invalid key
        with pytest.raises(websockets.ConnectionClosed) as exc_info:
            await client_factory(
                server=server_no_key, permanent_key=key.pk, explicit_permanent_key=True,
                initiator_handshake=True)
        assert exc_info.value.code == CloseCode.invalid_key
        await server.wait_connections_closed()

flist.append(Testframe(websockets.framing.Frame(1, 2, bytearray("Frame2 does contain much more text and even goes beyond the 126 byte len field. Frame2 does contain much more text and even goes beyond the 126 byte len field.", encoding="utf-8")),
    "Mid-long valid binary frame"))
#flist.append(Testframe(websockets.framing.Frame(1, 2, bytearray([(x % 26) + 65 for x in range(100000)])), "100k binary frame (ABC..YZABC..)"))

### some conn reset frames, one with no close message, one with close message
flist.append(Testframe(websockets.framing.Frame(1, 8, bytearray(list([0x03, 0xEB]))), "Close frame (Reason 1003)", experrno="ECONNRESET"))
flist.append(Testframe(websockets.framing.Frame(1, 8, bytearray(list([0x03, 0xEB])) + bytearray("I'm a close reason and much more than that!", encoding="utf-8")), "Close frame (Reason 1003) and msg", experrno="ECONNRESET"))

### invalid header values
flist.append(Testframe(websockets.framing.Frame(1, 1, bytearray("Testit", encoding="utf-8")), "Invalid frame: Wrong masking", experrno="EPROTO", mask=False))
flist.append(Testframe(websockets.framing.Frame(1, 1, bytearray("..Lore Ipsum", encoding="utf-8")), "Invalid frame: Length of < 126 with add. 16 bit len field", experrno="EPROTO", modify_bytes={ 1: 0xFE, 2: 0x00, 3: 0x0F}))
flist.append(Testframe(websockets.framing.Frame(1, 1, bytearray("........Lore Ipsum", encoding="utf-8")), "Invalid frame: Length of < 126 with add. 64 bit len field", experrno="EPROTO", modify_bytes={ 1: 0xFF, 2: 0x00, 3: 0x00, 4: 0x00, 5: 0x00, 6: 0x00, 7: 0x00, 8: 0x80, 9: 0x40}))

frag1 = websockets.framing.Frame(0, 1, bytearray("This is a fragmented websocket...", encoding="utf-8"))
frag2 = websockets.framing.Frame(0, 0, bytearray("... and it goes on...", encoding="utf-8"))
frag3 = websockets.framing.Frame(1, 0, bytearray("and on and stop", encoding="utf-8"))
flist.append(Testframe(frag1, "Continuation test frag1"))
flist.append(Testframe(frag2, "Continuation test frag2", opcode_overwrite=1))
flist.append(Testframe(frag3, "Continuation test frag3", opcode_overwrite=1))

s = "struct ws_frame_test tests[] = {\n"
for i in range(len(flist)):
    s += flist[i].__str__()
    if (i + 1 < len(flist)):
        s += ","
    s += "\n"
s += "};\n"

with open("wstestdata.inc", "w") as cdatafile:
    cdatafile.write(s)

def test_no_encode_decode_pong_frame(self):
        frame = Frame(True, OP_PONG, b"")

        self.assertEqual(self.extension.encode(frame), frame)

        self.assertEqual(self.extension.decode(frame), frame)

def test_no_decode_fragmented_binary_frame(self):
        frame1 = Frame(False, OP_TEXT, b"tea ")
        frame2 = Frame(True, OP_CONT, b"time")

        dec_frame1 = self.extension.decode(frame1)
        dec_frame2 = self.extension.decode(frame2)

        self.assertEqual(dec_frame1, frame1)
        self.assertEqual(dec_frame2, frame2)

def test_no_encode_decode_ping_frame(self):
        frame = Frame(True, OP_PING, b"")

        self.assertEqual(self.extension.encode(frame), frame)

        self.assertEqual(self.extension.decode(frame), frame)

def test_encode_decode_fragmented_text_frame(self):
        frame1 = Frame(False, OP_TEXT, "café".encode("utf-8"))
        frame2 = Frame(False, OP_CONT, " & ".encode("utf-8"))
        frame3 = Frame(True, OP_CONT, "croissants".encode("utf-8"))

        enc_frame1 = self.extension.encode(frame1)
        enc_frame2 = self.extension.encode(frame2)
        enc_frame3 = self.extension.encode(frame3)

        self.assertEqual(
            enc_frame1,
            frame1._replace(rsv1=True, data=b"JNL;\xbc\x12\x00\x00\x00\xff\xff"),
        )
        self.assertEqual(
            enc_frame2, frame2._replace(rsv1=True, data=b"RPS\x00\x00\x00\x00\xff\xff")
        )
        self.assertEqual(
            enc_frame3, frame3._replace(rsv1=True, data=b"J.\xca\xcf,.N\xcc+)\x06\x00")
        )

def test_context_takeover(self):
        frame = Frame(True, OP_TEXT, "café".encode("utf-8"))

        enc_frame1 = self.extension.encode(frame)
        enc_frame2 = self.extension.encode(frame)

        self.assertEqual(enc_frame1.data, b"JNL;\xbc\x12\x00")
        self.assertEqual(enc_frame2.data, b"J\x06\x11\x00\x00")

def test_no_decode_fragmented_text_frame(self):
        frame1 = Frame(False, OP_TEXT, "café".encode("utf-8"))
        frame2 = Frame(False, OP_CONT, " & ".encode("utf-8"))
        frame3 = Frame(True, OP_CONT, "croissants".encode("utf-8"))

        dec_frame1 = self.extension.decode(frame1)
        dec_frame2 = self.extension.decode(frame2)
        dec_frame3 = self.extension.decode(frame3)

        self.assertEqual(dec_frame1, frame1)
        self.assertEqual(dec_frame2, frame2)
        self.assertEqual(dec_frame3, frame3)