KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use call_exception_handler method in fMBT

Best Python code snippet using fMBT_python

test_exceptionhandler.py

Source:test_exceptionhandler.py

...16        def handler(loop, context):17            handled.append((loop, context))18        self.loop.set_exception_handler(handler)19        context = {"a": "b"}20        self.loop.call_exception_handler(context)21        self.assertEqual(handled, [(self.loop, context)])22        with utils.captured_log() as stream:23            # Make sure exceptions are swallowed24            handled = None25            self.loop.call_exception_handler(context)26            self.assertIs(handled, None)27        contents = stream.getvalue()28        self.assertIn("Unhandled error in exception handler", contents)29        self.assertIn("AttributeError", contents)30    def test_default_handler(self):31        with utils.captured_log() as stream:32            context = {"attribute": "value", "dogs": "cats"}33            self.loop.call_exception_handler(context)34        contents = stream.getvalue()35        self.assertIn("Unhandled exception in event loop", contents)36        self.assertIn("attribute", contents)37        self.assertIn("value", contents)38        self.assertIn("dogs", contents)39        self.assertIn("cats", contents)40    def test_default_handler_with_traceback(self):41        with utils.captured_log() as stream:42            for key, needle in [43                ("source_traceback", "Object"),44                ("handle_traceback", "Handle"),45            ]:46                with self.subTest(key=key):47                    stream.seek(0)48                    stream.truncate()49                    tb = traceback.extract_stack()50                    context = {"attribute": "value", key: tb}51                    self.loop.call_exception_handler(context)52                    contents = stream.getvalue()53                    self.assertIn("Unhandled exception in event loop", contents)54                    self.assertIn("attribute", contents)55                    self.assertIn("value", contents)56                    self.assertIn(f"{needle} created at (most", contents)57    def test_default_handler_with_handle_traceback(self):58        with utils.captured_log() as stream:59            tb = traceback.extract_stack()60            context = {"attribute": "value"}61            self.loop._current_handle = Attr()62            self.loop._current_handle._source_traceback = tb63            self.loop.call_exception_handler(context)64            contents = stream.getvalue()65            self.assertIn("Unhandled exception in event loop", contents)66            self.assertIn("attribute", contents)67            self.assertIn("value", contents)68            self.assertIn("Handle created at (most", contents)69    def test_exit_in_exception_handler(self):70        with utils.captured_log() as stream:71            def handler(loop, context):72                raise exception73            self.loop.set_exception_handler(handler)74            for exception in (SystemExit, KeyboardInterrupt):75                with self.subTest(exception=exception):76                    stream.seek(0)77                    stream.truncate()78                    async def main():79                        context = {"a": "b"}80                        self.loop.call_exception_handler(context)81                    with self.assertRaises(exception):82                        self.loop.run_until_complete(main())83                    self.assertEqual(stream.getvalue(), "")84    def test_exit_in_default_handler(self):85        with utils.captured_log() as stream:86            def handler(loop, context):87                raise ValueError("dummy")88            def default_handler(context):89                raise exception90            self.loop.default_exception_handler = default_handler91            for handler in (None, handler):92                self.loop.set_exception_handler(handler)93                for exception in (SystemExit, KeyboardInterrupt):94                    with self.subTest(handler=handler, exception=exception):95                        stream.seek(0)96                        stream.truncate()97                        async def main():98                            context = {"a": "b"}99                            self.loop.call_exception_handler(context)100                        with self.assertRaises(exception):101                            self.loop.run_until_complete(main())102                        self.assertEqual(stream.getvalue(), "")103    def test_error_in_default_handler(self):104        with utils.captured_log() as stream:105            class MyException(Exception):106                pass107            def handler(loop, context):108                raise ValueError("dummy")109            def default_handler(context):110                raise exception111            self.loop.default_exception_handler = default_handler112            for handler in (None, handler):113                self.loop.set_exception_handler(handler)114                for exception in (MyException,):115                    with self.subTest(handler=handler, exception=exception):116                        stream.seek(0)117                        stream.truncate()118                        context = {"a": "b"}119                        self.loop.call_exception_handler(context)120                        self.assertIn(121                            "Exception in default exception handler", stream.getvalue()122                        )123                        if handler is not None:124                            self.assertIn(125                                "while handling an unexpected error in custom "126                                "exception handler",127                                stream.getvalue(),...

notifier.py

Source:notifier.py

...41                exc.errno != errno.EWOULDBLOCK42                and exc.errno != errno.EAGAIN43            ):44                message = 'Exception ignored when trying to write to the signal wakeup fd'45                self.loop.call_exception_handler({'message': message, 'exception': exc})46    def _wakeup_send(self, data: bytes) -> None:47        assert self._wakeup_sock is not None48        try:49            self._wakeup_sock.send(data)50        except OSError as exc:51            if (52                exc.errno != errno.EWOULDBLOCK53                and exc.errno != errno.EAGAIN54            ):55                message = 'Exception ignored when trying to send to the signal wakeup fd'56                self.loop.call_exception_handler({'message': message, 'exception': exc})57    async def _read_loop(self) -> None:58        while True:59            signums = await self.loop.sock_recv(self._csock, 4096)60            if self._wakeup_fd != -1:61                self._wakeup_write(signums)62            elif self._wakeup_sock is not None:63                self._wakeup_send(signums)64            for signum in signums:65                try:66                    self.notify(signal.Signals(signum))67                except ValueError as exc:68                    message = 'Notifier received invalid signal in read loop'69                    self.loop.call_exception_handler({'message': message, 'exception': exc})70    def _set_wakeup_fd(self, wakeup_fd: int) -> None:71        try:72            self._wakeup_sock = socket.socket(fileno=wakeup_fd)73            self._wakeup_sock.setblocking(False)74        except OSError as exc:75            if exc.errno != WSAENOTSOCK:76                raise77            self._wakeup_fd = wakeup_fd78    def start_notifying(self) -> None:79        for signum in signal.valid_signals():80            try:81                signal.signal(signum, self.signal_handler)82            except OSError as exc:83                if exc.errno != errno.EINVAL:...

error.py

Source:error.py

...5Set handler as the new event loop exception handler.6    If handler is None, the default exception handler will be set.7    Otherwise, handler must be a callable with the signature matching (loop, context),8    where loop is a reference to the active event loop, and context is a dict object9    containing the details of the exception (see call_exception_handler() documentation10                                             for details about context).11loop.get_exception_handler()12    Return the current exception handler, or None if no custom exception handler was set.13    New in version 3.5.2.14loop.default_exception_handler(context)15Default exception handler.16    This is called when an exception occurs and no exception handler is set.17    This can be called by a custom exception handler that wants to defer to the default handler behavior.18    context parameter has the same meaning as in call_exception_handler().19loop.call_exception_handler(context)20Call the current event loop exception handler.21context is a dict object containing the following keys (new keys may be introduced in future Python versions):22    âmessageâ: Error message;23    âexceptionâ (optional): Exception object;24    âfutureâ (optional): asyncio.Future instance;25    âhandleâ (optional): asyncio.Handle instance;26    âprotocolâ (optional): Protocol instance;27    âtransportâ (optional): Transport instance;28    âsocketâ (optional): socket.socket instance.29    **Note This method should not be overloaded in subclassed event loops.30    **For custom exception handling, use the set_exception_handler() method.31"""32#DEBUGGING33"""...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.