How to use the websockets.framing.Frame function in websockets

To help you get started, we’ve selected a few websockets examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github LibVNC / libvncserver / test / wsmaketestframe.py View on Github external
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)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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")
        )
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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")
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
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)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
def test_encode_decode_fragmented_binary_frame(self):
        frame1 = Frame(False, OP_TEXT, b"tea ")
        frame2 = Frame(True, OP_CONT, b"time")

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

        self.assertEqual(
            enc_frame1, frame1._replace(rsv1=True, data=b"*IMT\x00\x00\x00\x00\xff\xff")
        )
        self.assertEqual(
            enc_frame2, frame2._replace(rsv1=True, data=b"*\xc9\xccM\x05\x00")
        )

        dec_frame1 = self.extension.decode(enc_frame1)
        dec_frame2 = self.extension.decode(enc_frame2)

        self.assertEqual(dec_frame1, frame1)
github aaugustin / websockets / tests / extensions / test_permessage_deflate.py View on Github external
def test_encode_decode_text_frame(self):
        frame = Frame(True, OP_TEXT, "café".encode("utf-8"))

        enc_frame = self.extension.encode(frame)

        self.assertEqual(enc_frame, frame._replace(rsv1=True, data=b"JNL;\xbc\x12\x00"))

        dec_frame = self.extension.decode(enc_frame)

        self.assertEqual(dec_frame, frame)
github LibVNC / libvncserver / test / wsmaketestframe.py View on Github external
flist = []
### standard text frames with different lengths
flist.append(Testframe(websockets.framing.Frame(1, 1, bytearray("Testit", encoding="utf-8")), "Short valid text frame"))
flist.append(Testframe(websockets.framing.Frame(1, 1, 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 text frame"))
#flist.append(Testframe(websockets.framing.Frame(1, 1, bytearray([(x % 26) + 65 for x in range(100000)])), "100k text frame (ABC..YZABC..)"))

### standard binary frames with different lengths
flist.append(Testframe(websockets.framing.Frame(1, 2, bytearray("Testit", encoding="utf-8")), "Short valid binary frame"))
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__()