summaryrefslogtreecommitdiffstatshomepage
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/extmod/framebuf_blit.py68
-rw-r--r--tests/extmod/framebuf_blit.py.exp45
-rw-r--r--tests/multi_net/tcp_recv_peek.py46
-rw-r--r--tests/multi_net/udp_recv_dontwait.py59
-rw-r--r--tests/multi_net/udp_recv_dontwait.py.exp7
-rw-r--r--tests/multi_net/udp_recv_peek.py36
-rw-r--r--tests/ports/unix/extra_coverage.py.exp7
-rwxr-xr-xtests/run-natmodtests.py15
-rwxr-xr-xtests/run-tests.py27
9 files changed, 297 insertions, 13 deletions
diff --git a/tests/extmod/framebuf_blit.py b/tests/extmod/framebuf_blit.py
new file mode 100644
index 0000000000..b1d98b330a
--- /dev/null
+++ b/tests/extmod/framebuf_blit.py
@@ -0,0 +1,68 @@
+# Test FrameBuffer.blit method.
+
+try:
+ import framebuf
+except ImportError:
+ print("SKIP")
+ raise SystemExit
+
+
+def printbuf():
+ print("--8<--")
+ for y in range(h):
+ for x in range(w):
+ print("%02x" % buf[(x + y * w)], end="")
+ print()
+ print("-->8--")
+
+
+w = 5
+h = 4
+buf = bytearray(w * h)
+fbuf = framebuf.FrameBuffer(buf, w, h, framebuf.GS8)
+
+fbuf2 = framebuf.FrameBuffer(bytearray(4), 2, 2, framebuf.GS8)
+fbuf2.fill(0xFF)
+
+# Blit another FrameBuffer, at various locations.
+for x, y in ((-1, -1), (0, 0), (1, 1), (4, 3)):
+ fbuf.fill(0)
+ fbuf.blit(fbuf2, x, y)
+ printbuf()
+
+# Blit a bytes object.
+fbuf.fill(0)
+image = (b"\x10\x11\x12\x13", 2, 2, framebuf.GS8)
+fbuf.blit(image, 1, 1)
+printbuf()
+
+# Blit a bytes object that has a stride.
+fbuf.fill(0)
+image = (b"\x20\x21\xff\x22\x23\xff", 2, 2, framebuf.GS8, 3)
+fbuf.blit(image, 1, 1)
+printbuf()
+
+# Blit a bytes object with a bytes palette.
+fbuf.fill(0)
+image = (b"\x00\x01\x01\x00", 2, 2, framebuf.GS8)
+palette = (b"\xa1\xa2", 2, 1, framebuf.GS8)
+fbuf.blit(image, 1, 1, -1, palette)
+printbuf()
+
+# Not enough elements in the tuple.
+try:
+ fbuf.blit((0, 0, 0), 0, 0)
+except ValueError:
+ print("ValueError")
+
+# Too many elements in the tuple.
+try:
+ fbuf.blit((0, 0, 0, 0, 0, 0), 0, 0)
+except ValueError:
+ print("ValueError")
+
+# Bytes too small.
+try:
+ fbuf.blit((b"", 1, 1, framebuf.GS8), 0, 0)
+except ValueError:
+ print("ValueError")
diff --git a/tests/extmod/framebuf_blit.py.exp b/tests/extmod/framebuf_blit.py.exp
new file mode 100644
index 0000000000..e340f1990c
--- /dev/null
+++ b/tests/extmod/framebuf_blit.py.exp
@@ -0,0 +1,45 @@
+--8<--
+ff00000000
+0000000000
+0000000000
+0000000000
+-->8--
+--8<--
+ffff000000
+ffff000000
+0000000000
+0000000000
+-->8--
+--8<--
+0000000000
+00ffff0000
+00ffff0000
+0000000000
+-->8--
+--8<--
+0000000000
+0000000000
+0000000000
+00000000ff
+-->8--
+--8<--
+0000000000
+0010110000
+0012130000
+0000000000
+-->8--
+--8<--
+0000000000
+0020210000
+0022230000
+0000000000
+-->8--
+--8<--
+0000000000
+00a1a20000
+00a2a10000
+0000000000
+-->8--
+ValueError
+ValueError
+ValueError
diff --git a/tests/multi_net/tcp_recv_peek.py b/tests/multi_net/tcp_recv_peek.py
new file mode 100644
index 0000000000..ff540dd3c3
--- /dev/null
+++ b/tests/multi_net/tcp_recv_peek.py
@@ -0,0 +1,46 @@
+# Test TCP recv with MSG_PEEK
+#
+# Note that bare metal LWIP only returns at most one packet's worth of TCP data
+# in any recv() call - including when peeking - so can't be too clever with
+# different recv() combinations
+import socket
+import random
+
+
+# Server
+def instance0():
+ PORT = random.randrange(10000, 50000)
+ multitest.globals(IP=multitest.get_network_ip(), PORT=PORT)
+ s = socket.socket()
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ s.bind(socket.getaddrinfo("0.0.0.0", PORT)[0][-1])
+ s.listen()
+ multitest.next()
+ s2, _ = s.accept()
+ print(s2.recv(8, socket.MSG_PEEK))
+ print(s2.recv(8))
+ s2.send(b"1234567890")
+ multitest.broadcast("0-sent")
+ multitest.wait("1-sent")
+ print(s2.recv(5, socket.MSG_PEEK))
+ print(s2.recv(5, socket.MSG_PEEK))
+ multitest.broadcast("0-recved")
+ multitest.wait("1-recved") # sync here necessary as MP sends RST if closing TCP early
+ s2.close()
+ s.close()
+
+
+# Client
+def instance1():
+ multitest.next()
+ s = socket.socket()
+ s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
+ s.send(b"abcdefgh")
+ multitest.broadcast("1-sent")
+ multitest.wait("0-sent")
+ s.send(b"klmnopqr")
+ print(s.recv(5, socket.MSG_PEEK))
+ print(s.recv(10))
+ multitest.broadcast("1-recved")
+ multitest.wait("0-recved")
+ s.close()
diff --git a/tests/multi_net/udp_recv_dontwait.py b/tests/multi_net/udp_recv_dontwait.py
new file mode 100644
index 0000000000..640f3f060e
--- /dev/null
+++ b/tests/multi_net/udp_recv_dontwait.py
@@ -0,0 +1,59 @@
+# Test UDP recv and recvfrom with MSG_DONTWAIT
+import random
+import socket
+
+try:
+ import errno, time
+except ImportError:
+ print("SKIP")
+ raise SystemExit
+
+
+# Server
+def instance0():
+ PORT = random.randrange(10000, 50000)
+ multitest.globals(IP=multitest.get_network_ip(), PORT=PORT)
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ s.bind(socket.getaddrinfo("0.0.0.0", PORT)[0][-1])
+ multitest.next()
+ begin = time.ticks_ms()
+
+ # do some recvs before instance1 starts, when we know no packet is waiting
+ try:
+ print(s.recvfrom(8, socket.MSG_DONTWAIT))
+ except OSError as e:
+ print(e.errno == errno.EAGAIN)
+ try:
+ print(s.recv(8, socket.MSG_DONTWAIT))
+ except OSError as e:
+ print(e.errno == errno.EAGAIN)
+
+ # the above steps should not have taken any substantial time
+ elapsed = time.ticks_diff(time.ticks_ms(), begin)
+ print(True if elapsed < 50 else elapsed)
+
+ # Now instance1 will send us a UDP packet
+ multitest.broadcast("0-ready")
+ multitest.wait("1-sent")
+
+ for _ in range(10): # retry if necessary, to allow for network delay
+ time.sleep_ms(100)
+ try:
+ print(s.recv(8, socket.MSG_DONTWAIT))
+ break
+ except OSError as er:
+ if er.errno != errno.EAGAIN:
+ raise er
+ s.close()
+
+
+# Client
+def instance1():
+ multitest.next()
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
+ multitest.wait("0-ready")
+ print(s.send(b"abcdefgh"))
+ multitest.broadcast("1-sent")
+ s.close()
diff --git a/tests/multi_net/udp_recv_dontwait.py.exp b/tests/multi_net/udp_recv_dontwait.py.exp
new file mode 100644
index 0000000000..f61fd4bbe2
--- /dev/null
+++ b/tests/multi_net/udp_recv_dontwait.py.exp
@@ -0,0 +1,7 @@
+--- instance0 ---
+True
+True
+True
+b'abcdefgh'
+--- instance1 ---
+8
diff --git a/tests/multi_net/udp_recv_peek.py b/tests/multi_net/udp_recv_peek.py
new file mode 100644
index 0000000000..47897ce553
--- /dev/null
+++ b/tests/multi_net/udp_recv_peek.py
@@ -0,0 +1,36 @@
+# Test UDP recv and recvfrom with MSG_PEEK
+import random
+import socket
+import time
+
+
+# Server
+def instance0():
+ PORT = random.randrange(10000, 50000)
+ multitest.globals(IP=multitest.get_network_ip(), PORT=PORT)
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ s.bind(socket.getaddrinfo("0.0.0.0", PORT)[0][-1])
+ multitest.next()
+ peek_bytes, peek_addr = s.recvfrom(8, socket.MSG_PEEK)
+ print(peek_bytes)
+ real_bytes, real_addr = s.recvfrom(8)
+ print(real_bytes)
+ print(peek_addr == real_addr) # source addr should be the same for each
+ res = s.sendto(b"1234567890", peek_addr)
+ print(res)
+ print(s.recv(5, socket.MSG_PEEK))
+ print(s.recv(5, socket.MSG_PEEK))
+ s.close()
+
+
+# Client
+def instance1():
+ multitest.next()
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
+ s.send(b"abcdefgh")
+ s.send(b"klmnopqr")
+ print(s.recv(5, socket.MSG_PEEK))
+ print(s.recv(10))
+ s.close()
diff --git a/tests/ports/unix/extra_coverage.py.exp b/tests/ports/unix/extra_coverage.py.exp
index 5ff947e883..19b32f1c9d 100644
--- a/tests/ports/unix/extra_coverage.py.exp
+++ b/tests/ports/unix/extra_coverage.py.exp
@@ -122,6 +122,13 @@ unlocked
KeyboardInterrupt:
KeyboardInterrupt:
10
+loop
+scheduled function
+loop
+scheduled function
+loop
+scheduled function
+scheduled function
# ringbuf
99 0
98 1
diff --git a/tests/run-natmodtests.py b/tests/run-natmodtests.py
index b858989daa..073e0b053e 100755
--- a/tests/run-natmodtests.py
+++ b/tests/run-natmodtests.py
@@ -73,6 +73,7 @@ class __FS:
return __File()
vfs.mount(__FS(), '/__remote')
sys.path.insert(0, '/__remote')
+{import_prelude}
sys.modules['{}'] = __import__('__injected')
"""
@@ -133,6 +134,13 @@ def detect_architecture(target):
def run_tests(target_truth, target, args, stats, resolved_arch):
+ global injected_import_hook_code
+
+ prelude = ""
+ if args.begin:
+ prelude = args.begin.read()
+ injected_import_hook_code = injected_import_hook_code.replace("{import_prelude}", prelude)
+
for test_file in args.files:
# Find supported test
test_file_basename = os.path.basename(test_file)
@@ -212,6 +220,13 @@ def main():
cmd_parser.add_argument(
"-a", "--arch", choices=AVAILABLE_ARCHS, help="override native architecture of the target"
)
+ cmd_parser.add_argument(
+ "-b",
+ "--begin",
+ type=argparse.FileType("rt"),
+ default=None,
+ help="prologue python file to execute before module import",
+ )
cmd_parser.add_argument("files", nargs="*", help="input test files")
args = cmd_parser.parse_args()
diff --git a/tests/run-tests.py b/tests/run-tests.py
index 9294c7e636..ecd611aa5b 100755
--- a/tests/run-tests.py
+++ b/tests/run-tests.py
@@ -605,9 +605,7 @@ class PyboardNodeRunner:
def run_tests(pyb, tests, args, result_dir, num_threads=1):
test_count = ThreadSafeCounter()
testcase_count = ThreadSafeCounter()
- passed_tests = ThreadSafeCounter([])
- failed_tests = ThreadSafeCounter([])
- skipped_tests = ThreadSafeCounter([])
+ test_results = ThreadSafeCounter([])
skip_tests = set()
skip_native = False
@@ -896,7 +894,7 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
if skip_it:
print("skip ", test_file)
- skipped_tests.append((test_name, test_file))
+ test_results.append((test_name, test_file, "skip", ""))
return
# Run the test on the MicroPython target.
@@ -911,7 +909,7 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
# start-up code (eg boot.py) when preparing to run the next test.
pyb.read_until(1, b"raw REPL; CTRL-B to exit\r\n")
print("skip ", test_file)
- skipped_tests.append((test_name, test_file))
+ test_results.append((test_name, test_file, "skip", ""))
return
# Look at the output of the test to see if unittest was used.
@@ -994,7 +992,7 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
# Print test summary, update counters, and save .exp/.out files if needed.
if test_passed:
print("pass ", test_file, extra_info)
- passed_tests.append((test_name, test_file))
+ test_results.append((test_name, test_file, "pass", ""))
rm_f(filename_expected)
rm_f(filename_mupy)
else:
@@ -1006,7 +1004,7 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
rm_f(filename_expected) # in case left over from previous failed run
with open(filename_mupy, "wb") as f:
f.write(output_mupy)
- failed_tests.append((test_name, test_file))
+ test_results.append((test_name, test_file, "fail", ""))
test_count.increment()
@@ -1035,9 +1033,10 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
print(line)
sys.exit(1)
- passed_tests = sorted(passed_tests.value)
- skipped_tests = sorted(skipped_tests.value)
- failed_tests = sorted(failed_tests.value)
+ test_results = test_results.value
+ passed_tests = list(r for r in test_results if r[2] == "pass")
+ skipped_tests = list(r for r in test_results if r[2] == "skip")
+ failed_tests = list(r for r in test_results if r[2] == "fail")
print(
"{} tests performed ({} individual testcases)".format(
@@ -1069,9 +1068,11 @@ def run_tests(pyb, tests, args, result_dir, num_threads=1):
with open(os.path.join(result_dir, RESULTS_FILE), "w") as f:
json.dump(
{
+ # The arguments passed on the command-line.
"args": vars(args),
- "passed_tests": [test[1] for test in passed_tests],
- "skipped_tests": [test[1] for test in skipped_tests],
+ # A list of all results of the form [(test, result, reason), ...].
+ "results": list(test[1:] for test in test_results),
+ # A list of failed tests. This is deprecated, use the "results" above instead.
"failed_tests": [test[1] for test in failed_tests],
},
f,
@@ -1248,7 +1249,7 @@ the last matching regex is used:
results_file = os.path.join(args.result_dir, RESULTS_FILE)
if os.path.exists(results_file):
with open(results_file, "r") as f:
- tests = json.load(f)["failed_tests"]
+ tests = list(test[0] for test in json.load(f)["results"] if test[1] == "fail")
else:
tests = []
elif len(args.files) == 0:
generated by cgit v1.2.3 (git 2.39.1) at 2025-06-05 19:07:47 +0000