tests/extmod: Add websocket tests.
These short unit tests test the base uPy methods as well as parts of the websocket protocol, as implemented by uPy. @dpgeorge converted the original socket based tests by @hosaka to ones that only require io.BytesIO.
This commit is contained in:
parent
38f063ea72
commit
ce0b5e078b
61
tests/extmod/websocket.py
Normal file
61
tests/extmod/websocket.py
Normal file
@ -0,0 +1,61 @@
|
|||||||
|
try:
|
||||||
|
import uio
|
||||||
|
import uerrno
|
||||||
|
import websocket
|
||||||
|
except ImportError:
|
||||||
|
import sys
|
||||||
|
print("SKIP")
|
||||||
|
sys.exit()
|
||||||
|
|
||||||
|
# put raw data in the stream and do a websocket read
|
||||||
|
def ws_read(msg, sz):
|
||||||
|
ws = websocket.websocket(uio.BytesIO(msg))
|
||||||
|
return ws.read(sz)
|
||||||
|
|
||||||
|
# do a websocket write and then return the raw data from the stream
|
||||||
|
def ws_write(msg, sz):
|
||||||
|
s = uio.BytesIO()
|
||||||
|
ws = websocket.websocket(s)
|
||||||
|
ws.write(msg)
|
||||||
|
s.seek(0)
|
||||||
|
return s.read(sz)
|
||||||
|
|
||||||
|
# basic frame
|
||||||
|
print(ws_read(b"\x81\x04ping", 4))
|
||||||
|
print(ws_read(b"\x80\x04ping", 4)) # FRAME_CONT
|
||||||
|
print(ws_write(b"pong", 6))
|
||||||
|
|
||||||
|
# split frames are not supported
|
||||||
|
# print(ws_read(b"\x01\x04ping", 4))
|
||||||
|
|
||||||
|
# extended payloads
|
||||||
|
print(ws_read(b'\x81~\x00\x80' + b'ping' * 32, 128))
|
||||||
|
print(ws_write(b"pong" * 32, 132))
|
||||||
|
|
||||||
|
# mask (returned data will be 'mask' ^ 'mask')
|
||||||
|
print(ws_read(b"\x81\x84maskmask", 4))
|
||||||
|
|
||||||
|
# close control frame
|
||||||
|
s = uio.BytesIO(b'\x88\x00') # FRAME_CLOSE
|
||||||
|
ws = websocket.websocket(s)
|
||||||
|
print(ws.read(1))
|
||||||
|
s.seek(2)
|
||||||
|
print(s.read(4))
|
||||||
|
|
||||||
|
# misc control frames
|
||||||
|
print(ws_read(b"\x89\x00\x81\x04ping", 4)) # FRAME_PING
|
||||||
|
print(ws_read(b"\x8a\x00\x81\x04pong", 4)) # FRAME_PONG
|
||||||
|
|
||||||
|
# close method
|
||||||
|
ws = websocket.websocket(uio.BytesIO())
|
||||||
|
ws.close()
|
||||||
|
|
||||||
|
# ioctl
|
||||||
|
ws = websocket.websocket(uio.BytesIO())
|
||||||
|
print(ws.ioctl(8)) # GET_DATA_OPTS
|
||||||
|
print(ws.ioctl(9, 2)) # SET_DATA_OPTS
|
||||||
|
print(ws.ioctl(9))
|
||||||
|
try:
|
||||||
|
ws.ioctl(-1)
|
||||||
|
except OSError as e:
|
||||||
|
print("ioctl: EINVAL:", e.args[0] == uerrno.EINVAL)
|
14
tests/extmod/websocket.py.exp
Normal file
14
tests/extmod/websocket.py.exp
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
b'ping'
|
||||||
|
b'ping'
|
||||||
|
b'\x81\x04pong'
|
||||||
|
b'pingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingpingping'
|
||||||
|
b'\x81~\x00\x80pongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpongpong'
|
||||||
|
b'\x00\x00\x00\x00'
|
||||||
|
b''
|
||||||
|
b'\x81\x02\x88\x00'
|
||||||
|
b'ping'
|
||||||
|
b'pong'
|
||||||
|
0
|
||||||
|
1
|
||||||
|
2
|
||||||
|
ioctl: EINVAL: True
|
Loading…
x
Reference in New Issue
Block a user