1# Test uasyncio stream readinto() method using TCP server/client
2
3try:
4    import uasyncio as asyncio
5except ImportError:
6    try:
7        import asyncio
8    except ImportError:
9        print("SKIP")
10        raise SystemExit
11
12try:
13    import uarray as array
14except ImportError:
15    try:
16        import array
17    except ImportError:
18        print("SKIP")
19        raise SystemExit
20
21PORT = 8000
22
23
24async def handle_connection(reader, writer):
25    writer.write(b"ab")
26    await writer.drain()
27
28    writer.write(b"c")
29    await writer.drain()
30
31    print("close")
32    writer.close()
33    await writer.wait_closed()
34
35    print("done")
36    ev.set()
37
38
39async def tcp_server():
40    global ev
41    ev = asyncio.Event()
42    server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT)
43    print("server running")
44    multitest.next()
45    async with server:
46        await asyncio.wait_for(ev.wait(), 10)
47
48
49async def tcp_client():
50    reader, writer = await asyncio.open_connection(IP, PORT)
51
52    ba = bytearray(2)
53    n = await reader.readinto(ba)
54    print(n)
55    print(ba[:n])
56
57    a = array.array("b", [0, 0])
58    n = await reader.readinto(a)
59    print(n)
60    print(a[:n])
61
62    try:
63        n = await reader.readinto(5)
64    except TypeError as er:
65        print("TypeError")
66
67    try:
68        n = await reader.readinto()
69    except TypeError as er:
70        print("TypeError")
71
72
73def instance0():
74    multitest.globals(IP=multitest.get_network_ip())
75    asyncio.run(tcp_server())
76
77
78def instance1():
79    multitest.next()
80    asyncio.run(tcp_client())
81