RIOT-OS/RIOT
RIOT-OS/RIOT
1
0
Fork 0
mirror of https://github.com/RIOT-OS/RIOT.git synced 2024-12-29 04:50:03 +01:00
Code Releases Activity

tests/lwip: Add a test for partial TCP reads.

Browse Source 
Add a new test to check the behavior of `sock_tcp_read` when more data
is available in the connection than the buffer passed. This test checks
issue #16124 as well as reading from multiple small network packets into
a single buffer.
This commit is contained in:
iosabi 2021-04-10 18:56:12 +02:00
parent e469f2dea4
commit 9974803f43
1 changed files with 51 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
tests/lwip/tests/01-run.py
View File

@ -275,6 +275,55 @@ def test_tcpv6_send(board_group, application, env=None):
client.expect_exact(u"could not send")
def test_tcpv6_large_send(board_group, application, env=None):
"""Test that the TCP server can receive a large packet in multiple reads"""
if any(b.name != "native" for b in board_group.boards):
# run test only with native
print("SKIP_TEST INFO found non-native board")
return
env_server = os.environ.copy()
if env is not None:
env_server.update(env)
env_server.update(board_group.boards[1].to_env())
with pexpect.spawnu(MAKE, ["-C", application, "term"], env=env_server,
timeout=DEFAULT_TIMEOUT) as server:
port = random.randint(0x0000, 0xffff)
server_ip = get_ipv6_address(server)
try:
connect_addr = socket.getaddrinfo(
"%s%%tapbr0" % server_ip, port)[0][4]
except socket.gaierror as e:
print("SKIP_TEST INFO", e)
return
server.sendline(u"tcp server start %d" % port)
# wait for neighbor discovery to be done
time.sleep(5)
# Send large amount of data to verify multiple reads.
with socket.socket(socket.AF_INET6) as sock:
sock.connect(connect_addr)
server.expect(u"TCP client \\[[0-9a-f:]+\\]:[0-9]+ connected")
# Default read size in tcp.c SOCK_INBUF_SIZE (256 bytes), so this
# is at least 3 reads.
data = bytearray(random.randint(0, 255) for i in range(608))
sock.send(data)
def data_line(i):
return u' '.join(u'%.2X' % data[i + j] for j in range(16))
# We expect three consecutive reads of 256 byte buffer to print the
# data at positions 0, 256 and 512.
server.expect(u"00000000 %s" % data_line(0))
server.expect(u"00000000 %s" % data_line(256))
server.expect(u"00000000 %s" % data_line(512))
# Last line of the message should be on the third read call,
# together with the rest of the third read.
server.expect(u"00000050 %s" % data_line(len(data) - 16))
sock.close()
server.expect(u"TCP connection to \\[[0-9a-f:]+\\]:[0-9]+ reset")
def test_tcpv6_multiconnect(board_group, application, env=None):
if any(b.name != "native" for b in board_group.boards):
# run test only with native
@ -366,4 +415,5 @@ if __name__ == "__main__":
TestStrategy().execute([BoardGroup((Board("native", "tap0"),
Board("native", "tap1")))],
[test_ipv6_send, test_udpv6_send, test_tcpv6_send,
test_tcpv6_multiconnect, test_triple_send])
test_tcpv6_large_send, test_tcpv6_multiconnect,
test_triple_send])