How to use mark_exception method in Slash

Best Python code snippet using slash

gfworkflow.py

Source:gfworkflow.py Github

copy

Full Screen

...521 exc_type, exc_value, tb = sys.exc_info()522 self._execute_on_error_listeners(ctx,523 exc_type, exc_value, tb,524 listener_objects)525 ctrl.mark_exception(exc_info=(exc_type, exc_value, tb))526 return BadRequestEnd(msg=unicode(e))527 except Exception:528 self._logger.exception(u"系统错误,工作流被迫中止")529 exc_type, exc_value, tb = sys.exc_info()530 self._execute_on_error_listeners(ctx,531 exc_type, exc_value, tb,532 listener_objects)533 ctrl.mark_exception(exc_info=(exc_type, exc_value, tb))534 return ErrorEnd(exc_type, exc_value, tb) # 返回错误的结果535 else:536 # 执行完成事件537 self._execute_listeners("on_unit_finish",538 ctx, listener_objects)539 self._logger.info(u"工作单元 {} [{}] 执行完毕".format(540 current_unit.name, current_unit.unittype))541 if goto == "end":542 self._execute_listeners("on_finish", ctx, listener_objects)543 self._logger.info(u"工作流成功执行完毕")544 ctrl.mark_finished(result=last_result)545 return OkEnd(result=last_result) # 将最后一个单元的结果作为默认返回的结果546 else:547 current_unit = self._units[goto] # 处理下一个单元548 def _execute_listeners(self, event_name, context, listener_objects):549 """遍历执行监听器550 """551 for idx, listener in enumerate(self._listeners):552 if isinstance(listener, AbstractListener):553 getattr(listener, event_name)(context)554 elif issubclass(listener, AbstractListener):555 listener_obj = self.__create_listener_obj(556 idx, listener, listener_objects)557 getattr(listener_obj, event_name)(context)558 def _execute_on_error_listeners(self, context,559 exc_type, exc_value, tb, listener_objects):560 """遍历执行错误监听器561 """562 for idx, listener in enumerate(self._listeners):563 if isinstance(listener, AbstractListener):564 listener.on_error(context, exc_type, exc_value, tb)565 elif issubclass(listener, AbstractListener):566 listener_obj = self.__create_listener_obj(567 idx, listener, listener_objects)568 listener_obj.on_error(569 context, exc_type, exc_value, tb)570 def __create_listener_obj(self, idx, listener_class, listener_objects):571 listener_obj = listener_objects.get(idx)572 if not listener_obj:573 listener_obj = listener_class()574 listener_objects[idx] = listener_obj575 return listener_obj576class SessionCtrl(AbstractSessionCtrl):577 def __init__(self):578 super(SessionCtrl, self).__init__()579 self._session_id = uuid.uuid1().hex580 self._result = None581 self._exc_info = tuple()582 self._current_unit = None583 self._stop_mark = False584 self._stop_on = None585 @property586 def session_id(self):587 return self._session_id588 @property589 def status(self):590 return self._status591 @status.setter592 def status(self, status):593 self._status = status594 @property595 def result(self):596 if self._status != AbstractSessionCtrl.STATUS_FINISHED:597 raise InvalidStatusException(u"会话没有完成,无法获取结果")598 return self._result599 @property600 def exc_info(self):601 if self._status != AbstractSessionCtrl.STATUS_EXCEPTION:602 raise InvalidStatusException(u"会话未因错误终止,无法获取异常信息")603 return self._exc_info604 @property605 def current_unit(self):606 return self._current_unit607 @current_unit.setter608 def current_unit(self, current_unit):609 self._current_unit = current_unit610 @property611 def stop_mark(self):612 return self._stop_mark613 @property614 def stop_on(self):615 return self._stop_on616 def stop(self, stop_on=None):617 self._stop_mark = True618 self._stop_on = stop_on619 def mark_finished(self, result=None):620 """该方法用于标记工作流正常结束621 """622 self._result = result623 self._status = AbstractSessionCtrl.STATUS_FINISHED624 def mark_exception(self, exc_info=None):625 """该方法用于标记异常终止626 """627 self._exc_info = exc_info628 self._status = AbstractSessionCtrl.STATUS_EXCEPTION629class WorkflowUnitExistedException(GirlFriendBizException):630 def __init__(self, unit_name):631 super(WorkflowUnitExistedException, self).__init__(632 u"工作单元 '{}' 已经存在".format(unit_name))633class WorkflowStoppedException(GirlFriendBizException):634 def __init__(self, stop_on):635 self._stop_on = stop_on636 super(WorkflowStoppedException, self).__init__(637 u"工作流已终止在 '{}' 单元上".format(stop_on))638 @property...

Full Screen

Full Screen

cli.py

Source:cli.py Github

copy

Full Screen

...249 prompter.print(runtime.lrepr(result))250 repl_module.mark_repl_result(result)251 except reader.SyntaxError as e:252 traceback.print_exception(reader.SyntaxError, e, e.__traceback__)253 repl_module.mark_exception(e)254 continue255 except compiler.CompilerException as e:256 traceback.print_exception(257 compiler.CompilerException, e, e.__traceback__258 )259 repl_module.mark_exception(e)260 continue261 except Exception as e:262 traceback.print_exception(Exception, e, e.__traceback__)263 repl_module.mark_exception(e)264 continue265@_subcommand(266 "repl",267 help="start the Basilisp REPL",268 description="Start a Basilisp REPL.",269 handler=repl,270)271def _add_repl_subcommand(parser: argparse.ArgumentParser) -> None:272 parser.add_argument(273 "--default-ns",274 default=runtime.REPL_DEFAULT_NS,275 help="default namespace to use for the REPL",276 )277 _add_compiler_arg_group(parser)...

Full Screen

Full Screen

exception_handling.py

Source:exception_handling.py Github

copy

Full Screen

...124 mark_exception_handled(exc_info[1])125 for handler in _EXCEPTION_HANDLERS:126 handler(exc_info)127def mark_exception_handled(e):128 return mark_exception(e, "handled", True)129def is_exception_handled(e):130 """131 Checks if the exception ``e`` already passed through the exception handling logic132 """133 return bool(get_exception_mark(e, "handled", False))134@contextmanager135def get_exception_swallowing_context(report_to_sentry=True):136 """137 Returns a context under which all exceptions are swallowed (ignored)138 """139 try:140 yield141 except:142 if not get_exception_mark(sys.exc_info()[1], "swallow", True):143 raise144 if report_to_sentry:145 capture_sentry_exception()146 _logger.debug("Ignoring exception", exc_info=sys.exc_info())147def noswallow(exception):148 """149 Marks an exception to prevent swallowing by :func:`slash.exception_handling.get_exception_swallowing_context`,150 and returns it151 """152 mark_exception(exception, "swallow", False)153 return exception154def mark_exception_fatal(exception):155 """156 Causes this exception to halt the execution of the entire run.157 This is useful when detecting errors that need careful examination, thus preventing further tests from158 altering the test subject's state159 """160 mark_exception(exception, "fatal", True)161 return exception162def mark_exception_frame_correction(exception, correction=+1):163 current_correction = get_exception_frame_correction(exception)164 return mark_exception(exception, 'frame_correction', current_correction + correction)165def get_exception_frame_correction(exception):166 return get_exception_mark(exception, 'frame_correction', 0)167def is_exception_fatal(exception):168 return bool(get_exception_mark(exception, "fatal", False))169def inhibit_unhandled_exception_traceback(exception):170 """171 Causes this exception to inhibit console tracback172 """173 mark_exception(exception, "inhibit_console_tb", True)174 return exception175def should_inhibit_unhandled_exception_traceback(exception):176 return bool(get_exception_mark(exception, "inhibit_console_tb", False))177def disable_exception_swallowing(func_or_exception):178 """179 Marks an exception to prevent swallowing. Can also be used as a decorator around a function to mark all escaped180 exceptions181 """182 if isinstance(func_or_exception, BaseException):183 return noswallow(func_or_exception)184 @functools.wraps(func_or_exception)185 def func(*args, **kwargs):186 try:187 return func_or_exception(*args, **kwargs)...

Full Screen

Full Screen

main.py

Source:main.py Github

copy

Full Screen

...15 if writer.get_url():16 if writer.scrap_data():17 writer.set_data()18 else:19 editor.mark_exception('no_meta_url')20 else:21 writer = MetadataWriter(editor)22 writer.adjust_subpage()23 editor.mark_exception('fill_manually')24 def quit(self):25 pass26class MetadataWriter:27 def __init__(self, editor: ReadingPageEditor):28 self.editor = editor29 self.page = editor.page30 self.titles = editor.titles31 self.write_contents = ContentsWriter()32 self.subpage = self._get_subpage()33 self.url = ''34 self.data = {}35 def set_data(self):36 self.page.root.disable_overwrite = not self.editor.enable_overwrite37 if true_name := self.data.get('name'):...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Slash automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful