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