Best Python code snippet using slash
result.py
Source:result.py
...42 self._log_path = None43 self._extra_logs = []44 def _fact_set_callback(self, fact_name, fact_value):45 gossip.trigger('slash.fact_set', name=fact_name, value=fact_value)46 def is_global_result(self):47 return False48 def serialize(self):49 serialized_object = {}50 for key, value in vars(self).items():51 if key not in Result.pickle_key_blacklist:52 try:53 serialized_object[key] = pickle.dumps(value)54 except (pickle.PicklingError, TypeError):55 _logger.debug('Failed serializing result, skipping this value. key = {}', key)56 capture_sentry_exception()57 return serialized_object58 def deserialize(self, result_dict):59 for key, value in result_dict.items():60 try:61 self.__dict__[key] = unpickle(value)62 except TypeError:63 _logger.error('Error when deserialize reult, skipping this value. key = {}', key,64 extra={'capture': False})65 for failure in self._failures:66 self.add_failure(failure, append=False)67 for error in self._errors:68 self.add_error(error, append=False)69 for skip in self._skips:70 self.add_skip(skip, append=False)71 def add_exception(self, exc_info=None):72 """Adds the currently active exception, assuming it wasn't already added to a result73 """74 if exc_info is None:75 exc_info = sys.exc_info()76 _, exc_value, _ = exc_info # pylint: disable=unpacking-non-sequence77 if _ADDED_TO_RESULT.is_exception_marked(exc_value):78 return79 _ADDED_TO_RESULT.mark_exception(exc_value)80 if isinstance(exc_value, exceptions.FAILURE_EXCEPTION_TYPES):81 self.add_failure()82 elif isinstance(exc_value, context.session.get_skip_exception_types()):83 self.add_skip(getattr(exc_value, 'reason', str(exc_value)))84 elif isinstance(exc_value, exceptions.INTERRUPTION_EXCEPTIONS):85 err = Error.capture_exception(exc_info=exc_info)86 hooks.interruption_added(result=self, exception=err) # pylint: disable=no-member87 session_result = context.session.results.global_result88 interrupted_test = self.is_interrupted()89 interrupted_session = session_result.is_interrupted()90 if not self.is_global_result():91 self.mark_interrupted()92 if not interrupted_test and not context.session.has_children():93 with notify_if_slow_context(message="Cleaning up test due to interrupt. Please wait..."),\94 handling_exceptions(swallow=True):95 hooks.test_interrupt() # pylint: disable=no-member96 if not interrupted_session:97 session_result.mark_interrupted()98 elif not isinstance(exc_value, GeneratorExit):99 #skip keyboardinterrupt and system exit100 self.add_error(exc_info=exc_info)101 else:102 _logger.trace('Ignoring GeneratorExit exception')103 def has_errors_or_failures(self):104 return bool(self._failures or self._errors)105 def get_log_path(self):106 """Returns log path107 """108 return self._log_path109 def set_log_path(self, path):110 """Set log path111 """112 self._log_path = path113 def get_log_dir(self):114 """Returns log's directory.115 """116 if self._log_path is None:117 return None118 return os.path.dirname(self._log_path)119 def add_extra_log_path(self, path):120 """Add additional log path. This path will be added to the list returns by get_log_paths121 """122 self._extra_logs.append(path)123 def get_log_paths(self):124 """Returns a list of all log paths125 """126 logs = []127 if self._log_path:128 logs.append(self._log_path)129 return logs + list(self._extra_logs)130 def is_started(self):131 return self._started132 def is_not_run(self):133 return not self.is_started() and not self.has_errors_or_failures()134 def mark_started(self):135 self._start_time = datetime.now()136 self._started = True137 def is_error(self):138 return bool(self._errors)139 @property140 def test_id(self):141 return self.test_metadata.id142 def is_failure(self):143 return bool(self._failures)144 def is_just_failure(self):145 """Indicates this is a pure failure, without errors involved"""146 return self.is_failure() and not self.is_error()147 def is_skip(self):148 return bool(self._skips)149 def is_run_and_skip(self):150 return self.is_started() and self.is_skip()151 def is_success(self, allow_skips=False):152 if self._errors or self._failures or self._interrupted:153 return False154 if self._skips:155 return allow_skips156 return self.is_started()157 def is_success_finished(self):158 return self.is_success() and self.is_finished()159 def is_finished(self):160 return self._finished161 def mark_finished(self):162 self._end_time = datetime.now()163 self._finished = True164 def mark_interrupted(self):165 self._interrupted = True166 def is_interrupted(self):167 return self._interrupted168 def add_error(self, e=None, frame_correction=0, exc_info=None, append=True):169 """Adds a failure to the result170 """171 err = self._add_error(self._errors, e, frame_correction=frame_correction + 1, exc_info=exc_info, append=append)172 context.reporter.report_test_error_added(context.test, err)173 return err174 def add_failure(self, e=None, frame_correction=0, exc_info=None, append=True):175 """Adds a failure to the result176 """177 err = self._add_error(self._failures, e, frame_correction=frame_correction + 1, exc_info=exc_info, is_failure=True, append=append)178 context.reporter.report_test_failure_added(context.test, err)179 return err180 def set_test_detail(self, key, value):181 """Adds a generic detail to this test result, which can be later inspected or used182 """183 self.details.set(key, value)184 def _add_error(self, error_list, error=None, frame_correction=0, exc_info=None, is_failure=False, append=True):185 try:186 if error is None:187 error = Error.capture_exception(exc_info=exc_info)188 if error is None:189 raise RuntimeError('add_error() must be called with either an argument or in an active exception')190 if not isinstance(error, Error):191 error = Error(error, frame_correction=frame_correction + 1, exc_info=exc_info)192 if is_failure:193 # force the error object to be marked as failure194 error.mark_as_failure()195 error.log_added()196 if append:197 error_list.append(error)198 #report children errors or parent session errors only199 if not context.session or not context.session.has_children() or self.is_global_result():200 hooks.error_added(result=self, error=error) # pylint: disable=no-member201 error.forget_exc_info()202 return error203 except Exception:204 _logger.error("Failed to add error to result", exc_info=True, extra={'capture': False})205 raise206 def add_skip(self, reason, append=True):207 if append:208 self._skips.append(reason)209 context.reporter.report_test_skip_added(context.test, reason)210 def get_duration(self):211 """Returns the test duration time as timedelta object212 :return: timedelta213 """214 if self._end_time is None or self._start_time is None:215 return timedelta()216 return self._end_time - self._start_time217 def get_errors(self):218 """Returns the list of errors recorded for this result219 :return: a list of :class:`slash.core.error.Error` objects220 """221 return self._errors222 def get_failures(self):223 """Returns the list of failures recorded for this result224 :return: a list of :class:`slash.core.error.Error` objects225 """226 return self._failures227 @deprecated('Use result.details.all()', since='0.20.0')228 def get_additional_details(self):229 return self.details.all()230 def get_skips(self):231 return self._skips232 def has_skips(self):233 return bool(self._skips)234 def has_fatal_errors(self):235 return any(e.is_fatal() for e in236 itertools.chain(self._errors, self._failures))237 has_fatal_exception = has_fatal_errors238 def __repr__(self):239 return "<{} ({})>".format(240 self.__class__.__name__,241 ", ".join(242 attr243 for attr in ("success", "error", "failure", "skip", "finished", "interrupted")244 if getattr(self, "is_{}".format(attr))()245 )246 )247class GlobalResult(Result):248 def __init__(self, session_results=None):249 super(GlobalResult, self).__init__()250 self._session_results = session_results251 def is_global_result(self):252 return True253 def is_success(self, allow_skips=False):254 if not super(GlobalResult, self).is_success(allow_skips=allow_skips):255 return False256 if self._session_results is None:257 return True258 if self._session_results.session.has_children() and self._session_results.session.parallel_manager.server.worker_error_reported:259 return False260 return all(result.is_success(allow_skips=allow_skips) for result in self._session_results.iter_test_results())261class SessionResults(object):262 def __init__(self, session):263 super(SessionResults, self).__init__()264 self.session = session265 self.global_result = GlobalResult(self)...
log.py
Source:log.py
...145 if config.root.log.cleanup.enabled and self._should_delete_log():146 os.remove(path)147 def _should_delete_log(self):148 return (not config.root.log.cleanup.keep_failed) or \149 (not self.session.results.current.is_global_result() and self.session.results.current.is_success(allow_skips=True)) or \150 (self.session.results.current.is_global_result() and self.session.results.is_success(allow_skips=True))151 def _get_error_logging_context(self):152 path = config.root.log.errors_subpath153 if path:154 warn_deprecation('log.errors_subpath configuration is deprecated since 1.5.0. '155 'Please use log.highlights_subpath instead')156 else:157 path = config.root.log.highlights_subpath158 def _error_added_filter(record, handler): # pylint: disable=unused-argument159 return record.extra.get('highlight')160 handler, log_path = self._get_file_log_handler(path, symlink=None, bubble=True, filter=_error_added_filter)161 if log_path and self.session.results.current is self.session.results.global_result:162 self.session.results.global_result.add_extra_log_path(log_path)163 return handler.applicationbound()164 def _get_silenced_logs_context(self):...
test_result.py
Source:test_result.py
...108 suite_test.expect_error()109 [result] = suite.run().get_all_results_for_test(suite_test)110 [err] = result.get_errors()111 assert err.traceback112def test_is_global_result(suite, suite_test):113 suite_test.append_line('assert not slash.context.result.is_global_result()')114 result = suite.run()115 assert result.session.results.global_result.is_global_result()116def test_global_result_is_success(suite, suite_test):117 suite_test.when_run.fail()118 assert not suite.run().session.results.global_result.is_success()119def test_global_result_error_without_started_context():120 with slash.Session() as session:121 with handling_exceptions(swallow=True):122 1/0 # pylint: disable=pointless-statement123 assert not session.results.is_success()124def test_session_cleanups_under_global_result(suite, suite_test):125 @suite_test.append_body126 def __code__(): # pylint: disable=unused-variable127 def cleanup():128 slash.context.result.data['ok'] = True129 slash.add_cleanup(cleanup, scope='session')...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!