(self)
| 364 | |
| 365 | class UnixSocketStreamTest(unittest.TestCase): |
| 366 | def setUp(self): |
| 367 | socket_dir = tempfile.mkdtemp() |
| 368 | self.build_context = tempfile.mkdtemp() |
| 369 | self.addCleanup(shutil.rmtree, socket_dir) |
| 370 | self.addCleanup(shutil.rmtree, self.build_context) |
| 371 | self.socket_file = os.path.join(socket_dir, 'test_sock.sock') |
| 372 | self.server_socket = self._setup_socket() |
| 373 | self.stop_server = False |
| 374 | server_thread = threading.Thread(target=self.run_server) |
| 375 | server_thread.daemon = True |
| 376 | server_thread.start() |
| 377 | self.response = None |
| 378 | self.request_handler = None |
| 379 | self.addCleanup(server_thread.join) |
| 380 | self.addCleanup(self.stop) |
| 381 | |
| 382 | def stop(self): |
| 383 | self.stop_server = True |
nothing calls this directly
no test coverage detected