148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
# test that socket.connect() on a non-blocking socket raises EINPROGRESS
|
|
# and that an immediate write/send/read/recv does the right thing
|
|
|
|
try:
|
|
import sys, time
|
|
import uerrno as errno, usocket as socket, ussl as ssl
|
|
except:
|
|
import socket, errno, ssl
|
|
isMP = sys.implementation.name == "micropython"
|
|
|
|
|
|
def dp(e):
|
|
# uncomment next line for development and testing, to print the actual exceptions
|
|
# print(repr(e))
|
|
pass
|
|
|
|
|
|
# do_connect establishes the socket and wraps it if tls is True.
|
|
# If handshake is true, the initial connect (and TLS handshake) is
|
|
# allowed to be performed before returning.
|
|
def do_connect(peer_addr, tls, handshake):
|
|
s = socket.socket()
|
|
s.setblocking(False)
|
|
try:
|
|
# print("Connecting to", peer_addr)
|
|
s.connect(peer_addr)
|
|
except OSError as er:
|
|
print("connect:", er.errno == errno.EINPROGRESS)
|
|
if er.errno != errno.EINPROGRESS:
|
|
print(" got", er.errno)
|
|
# wrap with ssl/tls if desired
|
|
if tls:
|
|
try:
|
|
if sys.implementation.name == "micropython":
|
|
s = ssl.wrap_socket(s, do_handshake=handshake)
|
|
else:
|
|
s = ssl.wrap_socket(s, do_handshake_on_connect=handshake)
|
|
print("wrap: True")
|
|
except Exception as e:
|
|
dp(e)
|
|
print("wrap:", e)
|
|
elif handshake:
|
|
# just sleep a little bit, this allows any connect() errors to happen
|
|
time.sleep(0.2)
|
|
return s
|
|
|
|
|
|
# test runs the test against a specific peer address.
|
|
def test(peer_addr, tls=False, handshake=False):
|
|
# MicroPython plain sockets have read/write, but CPython's don't
|
|
# MicroPython TLS sockets and CPython's have read/write
|
|
# hasRW captures this wonderful state of affairs
|
|
hasRW = isMP or tls
|
|
|
|
# MicroPython plain sockets and CPython's have send/recv
|
|
# MicroPython TLS sockets don't have send/recv, but CPython's do
|
|
# hasSR captures this wonderful state of affairs
|
|
hasSR = not (isMP and tls)
|
|
|
|
# connect + send
|
|
if hasSR:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
# send -> 4 or EAGAIN
|
|
try:
|
|
ret = s.send(b"1234")
|
|
print("send:", handshake and ret == 4)
|
|
except OSError as er:
|
|
#
|
|
dp(er)
|
|
print("send:", er.errno in (errno.EAGAIN, errno.EINPROGRESS))
|
|
s.close()
|
|
else: # fake it...
|
|
print("connect:", True)
|
|
if tls:
|
|
print("wrap:", True)
|
|
print("send:", True)
|
|
|
|
# connect + write
|
|
if hasRW:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
# write -> None
|
|
try:
|
|
ret = s.write(b"1234")
|
|
print("write:", ret in (4, None)) # SSL may accept 4 into buffer
|
|
except OSError as er:
|
|
dp(er)
|
|
print("write:", False) # should not raise
|
|
except ValueError as er: # CPython
|
|
dp(er)
|
|
print("write:", er.args[0] == "Write on closed or unwrapped SSL socket.")
|
|
s.close()
|
|
else: # fake it...
|
|
print("connect:", True)
|
|
if tls:
|
|
print("wrap:", True)
|
|
print("write:", True)
|
|
|
|
if hasSR:
|
|
# connect + recv
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
# recv -> EAGAIN
|
|
try:
|
|
print("recv:", s.recv(10))
|
|
except OSError as er:
|
|
dp(er)
|
|
print("recv:", er.errno == errno.EAGAIN)
|
|
s.close()
|
|
else: # fake it...
|
|
print("connect:", True)
|
|
if tls:
|
|
print("wrap:", True)
|
|
print("recv:", True)
|
|
|
|
# connect + read
|
|
if hasRW:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
# read -> None
|
|
try:
|
|
ret = s.read(10)
|
|
print("read:", ret is None)
|
|
except OSError as er:
|
|
dp(er)
|
|
print("read:", False) # should not raise
|
|
except ValueError as er: # CPython
|
|
dp(er)
|
|
print("read:", er.args[0] == "Read on closed or unwrapped SSL socket.")
|
|
s.close()
|
|
else: # fake it...
|
|
print("connect:", True)
|
|
if tls:
|
|
print("wrap:", True)
|
|
print("read:", True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# these tests use a non-existent test IP address, this way the connect takes forever and
|
|
# we can see EAGAIN/None (https://tools.ietf.org/html/rfc5737)
|
|
print("--- Plain sockets to nowhere ---")
|
|
test(socket.getaddrinfo("192.0.2.1", 80)[0][-1], False, False)
|
|
print("--- SSL sockets to nowhere ---")
|
|
# this test fails with AXTLS because do_handshake=False blocks on first read/write and
|
|
# there it times out until the connect is aborted
|
|
test(socket.getaddrinfo("192.0.2.1", 443)[0][-1], True, False)
|
|
print("--- Plain sockets ---")
|
|
test(socket.getaddrinfo("micropython.org", 80)[0][-1], False, True)
|
|
print("--- SSL sockets ---")
|
|
test(socket.getaddrinfo("micropython.org", 443)[0][-1], True, True)
|