8000 tests/extmod: Add coverage tests for select module. · micropython/micropython@34b6002 · GitHub
[go: up one dir, main page]
Skip to content

Commit 34b6002

Browse files
dpgeorgedpgeorge
committed
tests/extmod: Add coverage tests for select module.
Signed-off-by: Damien George <damien@micropython.org>
1 parent b87701f commit 34b6002

6 files changed

+191
-0
lines changed

‎tests/extmod/select_ipoll.py

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
# Test select.ipoll().
2+
3+
try:
4+
import socket, select
5+
except ImportError:
6+
print("SKIP")
7+
raise SystemExit
8+
9+
10+
def print_poll_output(lst):
11+
print([(type(obj), flags) for obj, flags in lst])
12+
13+
14+
poller = select.poll()
15+
16+
# Use a new UDP socket for tests, which should be writable but not readable.
17+
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
18+
s.bind(socket.getaddrinfo("127.0.0.1", 8000)[0][-1])
19+
20+
poller.register(s)
21+
22+
# Basic polling.
23+
print_poll_output(poller.ipoll(0))
24+
25+
# Pass in flags=1 for one-shot behaviour.
26+
print_poll_output(poller.ipoll(0, 1))
27+
28+
# Socket should be deregistered and poll should return nothing.
29+
print_poll_output(poller.ipoll(0))
30+
31+
# Create a second socket.
32+
s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
33+
s2.bind(socket.getaddrinfo("127.0.0.1", 8001)[0][-1])
34+
35+
# Register both sockets (to reset the first one).
36+
poller.register(s)
37+
poller.register(s2)
38+
39+
# Basic polling with two sockets.
40+
print_poll_output(poller.ipoll(0))
41+
42+
# Unregister the first socket, to test polling the remaining one.
43+
poller.unregister(s)
44+
print_poll_output(poller.ipoll(0))
45+
46+
# Unregister the second socket, to test polling none.
47+
poller.unregister(s2)
48+
print_poll_output(poller.ipoll(0))
49+
50+
s2.close()
51+
s.close()

‎tests/extmod/select_ipoll.py.exp

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
[(<class 'socket'>, 4)]
2+
[(<class 'socket'>, 4)]
3+
[]
4+
[(<class 'socket'>, 4), (<class 'socket'>, 4)]
5+
[(<class 'socket'>, 4)]
6+
[]

‎tests/extmod/select_poll_basic.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
# "Registering a file descriptor that’s already registered is not an error,
1717
# and has the same effect as registering the descriptor exactly once."
1818
poller.register(s)
19+
poller.register(s, select.POLLIN | select.POLLOUT)
1920

2021
# 2 args are mandatory unlike register()
2122
try:

‎tests/extmod/select_poll_custom.py

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
# Test custom pollable objects implemented in Python.
2+
3+
from micropython import const
4+
5+
try:
6+
import socket, select, io
7+
except ImportError:
8+
print("SKIP")
9+
raise SystemExit
10+
11+
_MP_STREAM_POLL = const(3)
12+
_MP_STREAM_GET_FILENO = const(10)
13+
14+
_MP_STREAM_POLL_RD = const(0x0001)
15+
_MP_STREAM_POLL_WR = const(0x0004)
16+
17+
18+
def print_poll_output(lst):
19+
print([(type(obj), flags) for obj, flags in lst])
20+
21+
22+
class CustomPollable(io.IOBase):
23+
def __init__(self):
24+
self.poll_state = 0
25+
26+
def ioctl(self, cmd, arg):
27+
print("CustomPollable.ioctl", cmd, arg)
28+
if cmd == _MP_STREAM_POLL:
29+
if self.poll_state == "delay_rd":
30+
self.poll_state = _MP_STREAM_POLL_RD
31+
return 0
32+
elif self.poll_state < 0:
33+
return self.poll_state
34+
else:
35+
return self.poll_state & arg
36+
if cmd == _MP_STREAM_GET_FILENO:
37+
return -1
38+
39+
40+
poller = select.poll()
41+
42+
# Use a new UDP socket for tests, which should be writable but not readable.
43+
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
44+
s.bind(socket.getaddrinfo("127.0.0.1", 8000)[0][-1])
45+
46+
x = CustomPollable()
47+
48+
# Register both a file-descriptor-based object and a custom pure-Python object.
49+
poller.register(s)
50+
poller.register(x)
51+
52+
# Modify the flags for the custom object.
53+
poller.modify(x, select.POLLIN)
54+
55+
# Test polling.
56+
print_poll_output(poller.poll(0))
57+
x.poll_state = _MP_STREAM_POLL_WR
58+
print_poll_output(poller.poll(0))
59+
x.poll_state = _MP_STREAM_POLL_RD
60+
print_poll_output(poller.poll(0))
61+
62+
# The custom object becomes readable only after being polled.
63+
poller.modify(s, select.POLLIN)
64+
x.poll_state = "delay_rd"
65+
print_poll_output(poller.poll())
66+
67+
# The custom object returns an error.
68+
x.poll_state = -1000
69+
try:
70+
poller.poll(0)
71+
except OSError as er:
72+
print("OSError", er.errno)
73+
74+
poller.unregister(x)
75+
poller.unregister(s)
76+
77+
s.close()

‎tests/extmod/select_poll_custom.py.exp

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
CustomPollable.ioctl 10 0
2+
CustomPollable.ioctl 3 1
3+
[(<class 'socket'>, 4)]
4+
CustomPollable.ioctl 3 1
5+
[(<class 'socket'>, 4)]
6+
CustomPollable.ioctl 3 1
7+
[(<class 'socket'>, 4), (<class 'CustomPollable'>, 1)]
8+
CustomPollable.ioctl 3 1
9+
CustomPollable.ioctl 3 1
10+
[(<class 'CustomPollable'>, 1)]
11+
CustomPollable.ioctl 3 1
12+
OSError 1000

‎tests/extmod/select_poll_fd.py

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
# Test select.poll in combination with file descriptors.
2+
3+
try:
4+
import select, errno
5+
6+
select.poll # Raises AttributeError for CPython implementations without poll()
7+
except (ImportError, AttributeError):
8+
print("SKIP")
9+
raise SystemExit
10+
11+
# Check that poll supports registering file descriptors (integers).
12+
try:
13+
select.poll().register(0)
14+
except OSError:
15+
print("SKIP")
16+
raise SystemExit
17+
18+
# Register invalid file descriptor.
19+
try:
20+
select.poll().register(-1)
21+
except ValueError:
22+
print("ValueError")
23+
24+
# Test polling stdout, it should be writable.
25+
poller = select.poll()
26+
poller.register(1)
27+
poller.modify(1, select.POLLOUT)
28+
print(poller.poll())
29+
30+
# Unregister then re-register.
31+
poller.unregister(1)
32+
poller.register(1, select.POLLIN)
33+
34+
# Poll for input, should return an empty list.
35+
print(poller.poll(0))
36+
37+
# Test registering a very large number of file descriptors.
38+
poller = select.poll()
39+
for fd in range(10000):
40+
poller.register(fd)
41+
try:
42+
poller.poll()
43+
except OSError as er:
44+
print(er.errno == errno.EINVAL)

0 commit comments

Comments
 (0)
1115
0