Test transferring larger amounts of data.
(selenium_nodesock)
| 124 | |
| 125 | |
| 126 | def test_socket_large_data_transfer(selenium_nodesock): |
| 127 | """Test transferring larger amounts of data.""" |
| 128 | DATA_SIZE = 64 * 1024 # 64KB |
| 129 | |
| 130 | server_received = [] |
| 131 | |
| 132 | def handler(conn, _addr): |
| 133 | received = b"" |
| 134 | while len(received) < DATA_SIZE: |
| 135 | chunk = conn.recv(8192) |
| 136 | if not chunk: |
| 137 | break |
| 138 | received += chunk |
| 139 | server_received.append(received) |
| 140 | conn.sendall(f"Received {len(received)} bytes".encode()) |
| 141 | |
| 142 | @run_in_pyodide |
| 143 | def run(selenium, host, port, data_size): |
| 144 | import socket |
| 145 | |
| 146 | s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 147 | s.connect((host, port)) |
| 148 | |
| 149 | data = b"X" * data_size |
| 150 | s.sendall(data) |
| 151 | |
| 152 | response = s.recv(1024) |
| 153 | s.close() |
| 154 | return response.decode() |
| 155 | |
| 156 | with tcp_server(handler, timeout=10.0) as (host, port): |
| 157 | result = run(selenium_nodesock, host, port, DATA_SIZE) |
| 158 | assert len(server_received) == 1 |
| 159 | assert len(server_received[0]) == DATA_SIZE |
| 160 | assert result == f"Received {DATA_SIZE} bytes" |
| 161 | |
| 162 | |
| 163 | def test_socket_getpeername(selenium_nodesock): |
nothing calls this directly
no test coverage detected
searching dependent graphs…