(self)
| 4063 | @skip_if_tsan_fork |
| 4064 | @support.requires_subprocess() |
| 4065 | def test_multiprocessing_queues(self): |
| 4066 | # See gh-119819 |
| 4067 | |
| 4068 | cd = copy.deepcopy(self.config_queue_handler) |
| 4069 | from multiprocessing import Queue as MQ, Manager as MM |
| 4070 | q1 = MQ() # this can't be pickled |
| 4071 | q2 = MM().Queue() # a proxy queue for use when pickling is needed |
| 4072 | q3 = MM().JoinableQueue() # a joinable proxy queue |
| 4073 | for qspec in (q1, q2, q3): |
| 4074 | fn = make_temp_file('.log', 'test_logging-cmpqh-') |
| 4075 | cd['handlers']['h1']['filename'] = fn |
| 4076 | cd['handlers']['ah']['queue'] = qspec |
| 4077 | qh = None |
| 4078 | try: |
| 4079 | self.apply_config(cd) |
| 4080 | qh = logging.getHandlerByName('ah') |
| 4081 | self.assertEqual(sorted(logging.getHandlerNames()), ['ah', 'h1']) |
| 4082 | self.assertIsNotNone(qh.listener) |
| 4083 | self.assertIs(qh.queue, qspec) |
| 4084 | self.assertIs(qh.listener.queue, qspec) |
| 4085 | finally: |
| 4086 | h = logging.getHandlerByName('h1') |
| 4087 | if h: |
| 4088 | self.addCleanup(closeFileHandler, h, fn) |
| 4089 | else: |
| 4090 | self.addCleanup(os.remove, fn) |
| 4091 | |
| 4092 | def test_90195(self): |
| 4093 | # See gh-90195 |
nothing calls this directly
no test coverage detected