69023622ee
CPython changed its non-blocking socket behaviour recently and this test would not run under CPython anymore. So the following steps were taken to get the test working again and then simplify it: - Run the test against CPython 3.10.10 and capture the output into the .exp file for the test. - Run this test on unix port of MicroPython and verify that the output matches the CPython 3.10.10 output in the new .exp file (it did). From now on take unix MicroPython as the source of truth for this test when modifying it. - Remove all code that was there for CPython compatibility. - Make it print out more useful information during the test run, including names of the OSError errno values. - Add polling of the socket before the send/write/recv/read to verify that the poll gives the correct result in non-blocking mode. Tested on unix MicroPython, ESP32_GENERIC, PYBD_SF2 and RPI_PICO_W boards. Signed-off-by: Damien George <damien@micropython.org>
116 lines
3.4 KiB
Python
116 lines
3.4 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
|
|
|
|
import errno
|
|
import select
|
|
import socket
|
|
import ssl
|
|
|
|
# only mbedTLS supports non-blocking mode
|
|
if not hasattr(ssl, "MBEDTLS_VERSION"):
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
|
|
# get the name of an errno error code
|
|
def errno_name(er):
|
|
if er == errno.EAGAIN:
|
|
return "EAGAIN"
|
|
if er == errno.EINPROGRESS:
|
|
return "EINPROGRESS"
|
|
return er
|
|
|
|
|
|
# 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:", errno_name(er.errno))
|
|
# wrap with ssl/tls if desired
|
|
if tls:
|
|
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
try:
|
|
s = ssl_context.wrap_socket(s, do_handshake_on_connect=handshake)
|
|
print("wrap ok: True")
|
|
except Exception as e:
|
|
print("wrap er:", e)
|
|
return s
|
|
|
|
|
|
# poll a socket and print out the result
|
|
def poll(s):
|
|
poller = select.poll()
|
|
poller.register(s)
|
|
print("poll: ", poller.poll(0))
|
|
|
|
|
|
# test runs the test against a specific peer address.
|
|
def test(peer_addr, tls, handshake):
|
|
# MicroPython plain and TLS sockets have read/write
|
|
hasRW = True
|
|
|
|
# MicroPython plain sockets have send/recv
|
|
# MicroPython TLS sockets don't have send/recv
|
|
hasSR = not tls
|
|
|
|
# connect + send
|
|
# non-blocking send should raise EAGAIN
|
|
if hasSR:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
poll(s)
|
|
try:
|
|
ret = s.send(b"1234")
|
|
print("send ok:", ret) # shouldn't get here
|
|
except OSError as er:
|
|
print("send er:", errno_name(er.errno))
|
|
s.close()
|
|
|
|
# connect + write
|
|
# non-blocking write should return None
|
|
if hasRW:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
poll(s)
|
|
ret = s.write(b"1234")
|
|
print("write: ", ret)
|
|
s.close()
|
|
|
|
# connect + recv
|
|
# non-blocking recv should raise EAGAIN
|
|
if hasSR:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
poll(s)
|
|
try:
|
|
ret = s.recv(10)
|
|
print("recv ok:", ret) # shouldn't get here
|
|
except OSError as er:
|
|
print("recv er:", errno_name(er.errno))
|
|
s.close()
|
|
|
|
# connect + read
|
|
# non-blocking read should return None
|
|
if hasRW:
|
|
s = do_connect(peer_addr, tls, handshake)
|
|
poll(s)
|
|
ret = s.read(10)
|
|
print("read: ", ret)
|
|
s.close()
|
|
|
|
|
|
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 ---")
|
|
test(socket.getaddrinfo("192.0.2.1", 443)[0][-1], True, False)
|
|
print("--- Plain sockets ---")
|
|
test(socket.getaddrinfo("micropython.org", 80)[0][-1], False, False)
|
|
print("--- SSL sockets ---")
|
|
test(socket.getaddrinfo("micropython.org", 443)[0][-1], True, True)
|