(self)
| 4129 | @skip_if_tsan_fork |
| 4130 | @support.requires_subprocess() |
| 4131 | def test_multiprocessing_queues(self): |
| 4132 | # See gh-119819 |
| 4133 | |
| 4134 | cd = copy.deepcopy(self.config_queue_handler) |
| 4135 | from multiprocessing import Queue as MQ, Manager as MM |
| 4136 | q1 = MQ() # this can't be pickled |
| 4137 | q2 = MM().Queue() # a proxy queue for use when pickling is needed |
| 4138 | q3 = MM().JoinableQueue() # a joinable proxy queue |
| 4139 | for qspec in (q1, q2, q3): |
| 4140 | fn = make_temp_file('.log', 'test_logging-cmpqh-') |
| 4141 | cd['handlers']['h1']['filename'] = fn |
| 4142 | cd['handlers']['ah']['queue'] = qspec |
| 4143 | qh = None |
| 4144 | try: |
| 4145 | self.apply_config(cd) |
| 4146 | qh = logging.getHandlerByName('ah') |
| 4147 | self.assertEqual(sorted(logging.getHandlerNames()), ['ah', 'h1']) |
| 4148 | self.assertIsNotNone(qh.listener) |
| 4149 | self.assertIs(qh.queue, qspec) |
| 4150 | self.assertIs(qh.listener.queue, qspec) |
| 4151 | finally: |
| 4152 | h = logging.getHandlerByName('h1') |
| 4153 | if h: |
| 4154 | self.addCleanup(closeFileHandler, h, fn) |
| 4155 | else: |
| 4156 | self.addCleanup(os.remove, fn) |
| 4157 | |
| 4158 | def test_90195(self): |
| 4159 | # See gh-90195 |
nothing calls this directly
no test coverage detected