Sets up imap_handler for tests. imap_handler should inherit from either: - SimpleIMAPHandler - for testing IMAP commands, - socketserver.StreamRequestHandler - if raw access to stream is needed. Returns (client, server).
(self, imap_handler, connect=True)
| 272 | client = None |
| 273 | |
| 274 | def _setup(self, imap_handler, connect=True): |
| 275 | """ |
| 276 | Sets up imap_handler for tests. imap_handler should inherit from either: |
| 277 | - SimpleIMAPHandler - for testing IMAP commands, |
| 278 | - socketserver.StreamRequestHandler - if raw access to stream is needed. |
| 279 | Returns (client, server). |
| 280 | """ |
| 281 | class TestTCPServer(self.server_class): |
| 282 | def handle_error(self, request, client_address): |
| 283 | """ |
| 284 | End request and raise the error if one occurs. |
| 285 | """ |
| 286 | self.close_request(request) |
| 287 | self.server_close() |
| 288 | raise |
| 289 | |
| 290 | self.addCleanup(self._cleanup) |
| 291 | self.server = self.server_class((socket_helper.HOST, 0), imap_handler) |
| 292 | self.thread = threading.Thread( |
| 293 | name=self._testMethodName+'-server', |
| 294 | target=self.server.serve_forever, |
| 295 | # Short poll interval to make the test finish quickly. |
| 296 | # Time between requests is short enough that we won't wake |
| 297 | # up spuriously too many times. |
| 298 | kwargs={'poll_interval': 0.01}) |
| 299 | self.thread.daemon = True # In case this function raises. |
| 300 | self.thread.start() |
| 301 | |
| 302 | if connect: |
| 303 | self.client = self.imap_class(*self.server.server_address) |
| 304 | |
| 305 | return self.client, self.server |
| 306 | |
| 307 | def _cleanup(self): |
| 308 | """ |
no test coverage detected