|
1 | 1 | import os |
2 | 2 | import socket |
| 3 | +import tempfile |
| 4 | +import threading |
3 | 5 | import unittest |
| 6 | +from unittest.mock import patch |
4 | 7 |
|
5 | 8 | import server |
6 | 9 |
|
@@ -99,5 +102,83 @@ def test_response_headers_include_length_and_status(self): |
99 | 102 | self.assertIn("Connection: close", response) |
100 | 103 |
|
101 | 104 |
|
| 105 | +class StreamingAndConcurrencyTests(unittest.TestCase): |
| 106 | + def test_streams_a_deterministic_multi_chunk_fixture(self): |
| 107 | + payload = b"0123456789abcdef" * 700 |
| 108 | + sender, receiver = socket.socketpair() |
| 109 | + result = {} |
| 110 | + |
| 111 | + try: |
| 112 | + with tempfile.NamedTemporaryFile(suffix=".txt") as fixture: |
| 113 | + fixture.write(payload) |
| 114 | + fixture.flush() |
| 115 | + headers, error = server.get_content_headers_for_path(fixture.name) |
| 116 | + |
| 117 | + self.assertIsNone(error) |
| 118 | + |
| 119 | + def stream_fixture(): |
| 120 | + try: |
| 121 | + result["bytes_sent"] = server.send_file(sender, fixture.name, headers) |
| 122 | + sender.shutdown(socket.SHUT_WR) |
| 123 | + except Exception as exc: # Propagate worker failures below. |
| 124 | + result["error"] = exc |
| 125 | + |
| 126 | + stream_thread = threading.Thread(target=stream_fixture) |
| 127 | + stream_thread.start() |
| 128 | + |
| 129 | + response = bytearray() |
| 130 | + while chunk := receiver.recv(4096): |
| 131 | + response.extend(chunk) |
| 132 | + |
| 133 | + stream_thread.join(timeout=2) |
| 134 | + self.assertFalse(stream_thread.is_alive()) |
| 135 | + finally: |
| 136 | + sender.close() |
| 137 | + receiver.close() |
| 138 | + |
| 139 | + if "error" in result: |
| 140 | + raise result["error"] |
| 141 | + |
| 142 | + _, response_body = bytes(response).split(b"\r\n\r\n", 1) |
| 143 | + self.assertEqual(result["bytes_sent"], len(payload)) |
| 144 | + self.assertEqual(response_body, payload) |
| 145 | + |
| 146 | + def test_bounded_pool_rejects_work_when_its_queue_is_full(self): |
| 147 | + started = threading.Event() |
| 148 | + release = threading.Event() |
| 149 | + server_sockets = [] |
| 150 | + peer_sockets = [] |
| 151 | + |
| 152 | + def blocking_handler(client_socket, _client_address): |
| 153 | + started.set() |
| 154 | + release.wait(timeout=2) |
| 155 | + client_socket.close() |
| 156 | + |
| 157 | + try: |
| 158 | + with patch.object(server, "handle_client", side_effect=blocking_handler): |
| 159 | + pool = server.ThreadPool(max_workers=1, queue_size=1) |
| 160 | + |
| 161 | + for _ in range(3): |
| 162 | + server_socket, peer_socket = socket.socketpair() |
| 163 | + server_sockets.append(server_socket) |
| 164 | + peer_sockets.append(peer_socket) |
| 165 | + |
| 166 | + self.assertTrue(pool.submit(server_sockets[0], ("local", 1))) |
| 167 | + self.assertTrue(started.wait(timeout=1)) |
| 168 | + self.assertTrue(pool.submit(server_sockets[1], ("local", 2))) |
| 169 | + self.assertFalse(pool.submit(server_sockets[2], ("local", 3))) |
| 170 | + |
| 171 | + server_sockets[2].close() |
| 172 | + release.set() |
| 173 | + pool.tasks.join() |
| 174 | + finally: |
| 175 | + release.set() |
| 176 | + for sock in server_sockets + peer_sockets: |
| 177 | + try: |
| 178 | + sock.close() |
| 179 | + except OSError: |
| 180 | + pass |
| 181 | + |
| 182 | + |
102 | 183 | if __name__ == "__main__": |
103 | 184 | unittest.main() |
0 commit comments