| 6 | |
| 7 | |
| 8 | class WsFrame: |
| 9 | |
| 10 | CONT = 0 |
| 11 | TEXT = 1 |
| 12 | BINARY = 2 |
| 13 | RSVD3 = 3 |
| 14 | RSVD4 = 4 |
| 15 | RSVD5 = 5 |
| 16 | RSVD6 = 6 |
| 17 | RSVD7 = 7 |
| 18 | CLOSE = 8 |
| 19 | PING = 9 |
| 20 | PONG = 10 |
| 21 | RSVD11 = 11 |
| 22 | RSVD12 = 12 |
| 23 | RSVD13 = 13 |
| 24 | RSVD14 = 14 |
| 25 | RSVD15 = 15 |
| 26 | |
| 27 | OP_NAMES = [ |
| 28 | "CONT", |
| 29 | "TEXT", |
| 30 | "BINARY", |
| 31 | "RSVD3", |
| 32 | "RSVD4", |
| 33 | "RSVD5", |
| 34 | "RSVD6", |
| 35 | "RSVD7", |
| 36 | "CLOSE", |
| 37 | "PING", |
| 38 | "PONG", |
| 39 | "RSVD11", |
| 40 | "RSVD12", |
| 41 | "RSVD13", |
| 42 | "RSVD14", |
| 43 | "RSVD15", |
| 44 | ] |
| 45 | |
| 46 | def __init__(self, opcode: int, fin: bool, mask: bytes, data: bytes): |
| 47 | self.opcode = opcode |
| 48 | self.fin = fin |
| 49 | self.mask = mask |
| 50 | self.data = data |
| 51 | self.length = len(data) |
| 52 | |
| 53 | def __repr__(self): |
| 54 | return f'WsFrame[{self.OP_NAMES[self.opcode]} fin={self.fin}, mask={self.mask}, len={len(self.data)}]' |
| 55 | |
| 56 | @property |
| 57 | def data_len(self) -> int: |
| 58 | return len(self.data) if self.data else 0 |
| 59 | |
| 60 | def to_network(self) -> bytes: |
| 61 | nd = bytearray() |
| 62 | h1 = self.opcode |
| 63 | if self.fin: |
| 64 | h1 |= 0x80 |
| 65 | nd.extend(struct.pack("!B", h1)) |
no outgoing calls
no test coverage detected