forked from python-websockets/websockets
-
Notifications
You must be signed in to change notification settings - Fork 0
/
protocol.py
29 lines (20 loc) · 767 Bytes
/
protocol.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from websockets.protocol import Protocol
class RecordingProtocol(Protocol):
"""
Protocol subclass that records incoming frames.
By interfacing with this protocol, you can check easily what the component
being testing sends during a test.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.frames_rcvd = []
def get_frames_rcvd(self):
"""
Get incoming frames received up to this point.
Calling this method clears the list. Each frame is returned only once.
"""
frames_rcvd, self.frames_rcvd = self.frames_rcvd, []
return frames_rcvd
def recv_frame(self, frame):
self.frames_rcvd.append(frame)
super().recv_frame(frame)