(self)
| 3993 | @threading_helper.requires_working_threading() |
| 3994 | @support.requires_subprocess() |
| 3995 | def test_config_queue_handler(self): |
| 3996 | qs = [CustomQueue(), CustomQueueProtocol()] |
| 3997 | dqs = [{'()': f'{__name__}.{cls}', 'maxsize': 10} |
| 3998 | for cls in ['CustomQueue', 'CustomQueueProtocol']] |
| 3999 | dl = { |
| 4000 | '()': __name__ + '.listenerMaker', |
| 4001 | 'arg1': None, |
| 4002 | 'arg2': None, |
| 4003 | 'respect_handler_level': True |
| 4004 | } |
| 4005 | qvalues = (None, __name__ + '.queueMaker', __name__ + '.CustomQueue', *dqs, *qs) |
| 4006 | lvalues = (None, __name__ + '.CustomListener', dl, CustomListener) |
| 4007 | for qspec, lspec in itertools.product(qvalues, lvalues): |
| 4008 | self.do_queuehandler_configuration(qspec, lspec) |
| 4009 | |
| 4010 | # Some failure cases |
| 4011 | qvalues = (None, 4, int, '', 'foo') |
| 4012 | lvalues = (None, 4, int, '', 'bar') |
| 4013 | for qspec, lspec in itertools.product(qvalues, lvalues): |
| 4014 | if lspec is None and qspec is None: |
| 4015 | continue |
| 4016 | with self.assertRaises(ValueError) as ctx: |
| 4017 | self.do_queuehandler_configuration(qspec, lspec) |
| 4018 | msg = str(ctx.exception) |
| 4019 | self.assertEqual(msg, "Unable to configure handler 'ah'") |
| 4020 | |
| 4021 | def _apply_simple_queue_listener_configuration(self, qspec): |
| 4022 | self.apply_config({ |
nothing calls this directly
no test coverage detected