KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use expect_interruption method in Slash
Best Python code snippet using slash
test_interruptions.py
Source:test_interruptions.py 
...37 summary = interrupted_suite.run(expect_interruption=True)38 assert cleanup not in summary.events39def test_sigterm_interrupt(suite, suite_test):40 suite_test.append_line('raise slash.exceptions.TerminatedException()')41 suite_test.expect_interruption()42 for test in suite.iter_all_after(suite_test):43 test.expect_deselect()44 suite.run(expect_interruption=True)45@pytest.mark.parametrize('hook_name', ['session_start', 'test_start'])46def test_sigterm_on_hook(suite, hook_name):47 @gossip.register('slash.{}'.format(hook_name))48 def session_start(): # pylint: disable=unused-variable49 raise slash.exceptions.TerminatedException('Terminated by signal')50 assert suite51 for index, test in enumerate(suite):52 if index == 0 and hook_name == 'test_start':53 # first test should be interrupted...54 test.expect_interruption()55 else:56 test.expect_deselect()57 result = suite.run(expect_interruption=True)58def test_test_end_called_for_interrupted_test(interrupted_suite, interrupted_test):59 ended = []60 @gossip.register('slash.test_end')61 def test_end():62 ended.append(slash.context.test.__slash__.id)63 s = interrupted_suite.run(expect_interruption=True)64 result = s[interrupted_test]65 assert result.test_metadata.id in ended66def test_ayalas(interrupted_suite, interrupted_test, interrupted_index, config_override, tmpdir):67 config_override('log.format', 'file: {record.message}')68 config_override('log.console_format', 'console: {record.message}')69 config_override('log.root', str(tmpdir))70 callback = Checkpoint()71 slash.hooks.log_file_closed.register(callback) # pylint: disable=no-member72 result = interrupted_suite.run(expect_interruption=True)73 num_closed_log_files = interrupted_index + 2 # One for each test that run (the index is zero based) + session log74 assert callback.called_count == num_closed_log_files75def test_session_interruption_in_start(suite, suite_test, session_interrupt):76 @suite.slashconf.append_body77 def __code__():78 @slash.hooks.session_start.register # pylint: disable=no-member79 def session_cleanup():80 raise KeyboardInterrupt()81 for test in suite:82 test.expect_deselect()83 suite.run(expect_interruption=True)84 assert session_interrupt.called_count == 185def test_interrupt_hooks_should_be_called_once(suite, suite_test, is_last_test, session_interrupt, test_interrupt_callback):86 @suite_test.append_body87 def __code__():88 @slash.add_critical_cleanup89 def cleanup():90 raise KeyboardInterrupt('A')91 raise KeyboardInterrupt('B')92 suite_test.expect_interruption()93 for t in suite.iter_all_after(suite_test, assert_has_more=not is_last_test):94 t.expect_deselect()95 result = suite.run(expect_interruption=True)96 assert test_interrupt_callback.called_count == 197 assert session_interrupt.called_count == 198 assert result.session.results.global_result.is_interrupted()99def test_interrupted_with_custom_exception(suite, suite_test, request):100 import test101 class CustomException(Exception):102 pass103 test.__interruption_exception__ = CustomException104 prev_interruption_exceptions = slash.exceptions.INTERRUPTION_EXCEPTIONS105 slash.exceptions.INTERRUPTION_EXCEPTIONS += (CustomException,)106 @request.addfinalizer107 def cleanup():108 del test.__interruption_exception__109 slash.exceptions.INTERRUPTION_EXCEPTIONS = prev_interruption_exceptions110 suite_test.append_line('import test')111 suite_test.append_line('raise test.__interruption_exception__()')112 suite_test.expect_interruption()113 for t in suite.iter_all_after(suite_test):114 t.expect_deselect()115 results = suite.run(expect_interruption=True)116def test_test_interrupt_hook_exception(suite_builder):117 # pylint: disable=reimported,redefined-outer-name118 @suite_builder.first_file.add_code119 def __code__():120 import slash121 @slash.hooks.test_interrupt.register # pylint: disable=no-member122 def test_interrupt(**_):123 1/0 # pylint: disable=pointless-statement124 def test_1():125 raise KeyboardInterrupt()126 def test_2():127 pass128 [res] = suite_builder.build().run().assert_results(1)129 assert res.is_interrupted()130@pytest.mark.parametrize('hook_name', ['before_session_cleanup', 'session_start', 'before_session_start'])131def test_session_scope_interruption(hook_name, suite, checkpoint):132 @gossip.register('slash.{}'.format(hook_name))133 def hook(*_, **__):134 raise KeyboardInterrupt()135 @gossip.register('slash.session_interrupt')136 def interrupt(*_, **__):137 checkpoint()138 if 'session_start' in hook_name:139 for test in suite:140 test.expect_deselect()141 else:142 assert hook_name == 'before_session_cleanup'143 suite[-1].expect_interruption()144 results = suite.run(expect_interruption=True)145 assert results.session.results.global_result.is_interrupted()146 assert checkpoint.called147@pytest.fixture148def session_interrupt():149 callback = Checkpoint()150 slash.hooks.session_interrupt.register(callback) # pylint: disable=no-member151 return callback152@pytest.fixture153def test_interrupt_callback():154 callback = Checkpoint()155 slash.hooks.test_interrupt.register(callback) # pylint: disable=no-member156 return callback157@pytest.fixture158def interrupted_suite(suite, interrupted_index):159 for index, test in enumerate(suite):160 if index == interrupted_index:161 test.append_line('raise KeyboardInterrupt()')162 test.expect_interruption()163 elif index > interrupted_index:164 test.expect_deselect()165 return suite166@pytest.fixture167def interrupted_test(interrupted_suite, interrupted_index):168 return interrupted_suite[interrupted_index]169@pytest.fixture170def interrupted_index(suite):...
suite.py
Source:suite.py
1import warnings2warnings.filterwarnings("ignore", category=DeprecationWarning)3import munch4import itertools5import os6import sys7import tempfile8from contextlib import contextmanager9import gossip10from io import StringIO11from slash.frontend.slash_run import slash_run12from ..code_formatter import CodeFormatter13from .file import File14from .slash_run_result import SlashRunResult15from .suite_strategy import BalancedStrategy16from .validation import validate_run, get_test_id_from_test_address17from .utils import get_temporary_slashrc_context18class Suite(object):19 def __init__(self, strategy=BalancedStrategy(), path=None, debug_info=True, is_parallel=False):20 super(Suite, self).__init__()21 self._path = path22 self._last_committed_path = None23 self.strategy = strategy24 self.debug_info = debug_info25 self.is_parallel = is_parallel26 self.clear()27 def disable_debug_info(self):28 self.debug_info = False29 def deselect_all(self, exclude=()):30 for test in self:31 if test in exclude:32 continue33 test.expect_deselect()34 def populate(self, num_tests=10):35 for _ in range(num_tests):36 self.add_test()37 def clear(self):38 self._files = []39 self._notified = []40 self._num_method_tests = self._num_function_tests = 041 self._slashconf = self._slashrc = None42 def iter_all_after(self, test, assert_has_more=False):43 found = had_more = False44 for t in self:45 if t == test:46 found = True47 elif found:48 had_more = True49 yield t50 if assert_has_more:51 assert had_more52 assert found53 @property54 def classes(self):55 return [cls for file in self._files for cls in file.classes]56 @property57 def files(self):58 return list(self._files)59 @property60 def slashconf(self):61 if self._slashconf is None:62 self._slashconf = File(self, relpath='slashconf.py')63 return self._slashconf64 @property65 def slashrc(self):66 if self._slashrc is None:67 self._slashrc = File(self, relpath='.slashrc')68 return self._slashrc69 def add_test(self, type=None, file=None): # pylint: disable=unused-argument70 if type is None:71 type = self.strategy.get_test_type()72 if type == 'function':73 returned = self.add_function_test()74 elif type == 'method':75 returned = self.add_method_test()76 else:77 raise NotImplementedError('Unknown test type {!r}'.format(type)) # pragma: no cover78 assert returned in self._notified79 return returned80 def add_method_test(self):81 cls = self.strategy.get_class_for_test(82 self.strategy.get_file_for_test(self))83 return cls.add_method_test()84 def add_function_test(self):85 return self.strategy.get_file_for_test(self).add_function_test()86 def notify_test_added(self, test):87 self._notified.append(test)88 if test.is_method_test():89 self._num_method_tests += 190 else:91 self._num_function_tests += 192 def add_file(self):93 returned = File(self)94 self._files.append(returned)95 return returned96 def get_last_file(self):97 if not self._files:98 return None99 return self._files[-1]100 def __len__(self):101 return len(self._notified)102 def __getitem__(self, idx):103 return self._notified[idx]104 def run(self, verify=True, expect_interruption=False, additional_args=(), args=None, commit=True, sort=True, num_workers=1,105 expect_session_errors=False):106 if commit:107 self.commit()108 path = self._last_committed_path109 assert path is not None110 report_stream = StringIO()111 returned = SlashRunResult(report_stream=report_stream)112 captured = []113 if args is None:114 args = [path]115 args.extend(additional_args)116 if self.is_parallel:117 args.extend(['--parallel', str(num_workers), '-vvvvv', '--parallel-addr', 'localhost'])118 with self._capture_events(returned), self._custom_sorting(sort):119 with self._custom_slashrc(path):120 app = slash_run(121 munch.Munch(argv=args, cmd="run"), report_stream=report_stream,122 app_callback=captured.append,123 )124 returned.exit_code = app.exit_code125 if app.interrupted:126 assert expect_interruption, 'Unexpectedly interrupted'127 else:128 assert not expect_interruption, 'Session was not interrupted as expected'129 if captured:130 assert len(captured) == 1131 returned.session = captured[0].session132 assert not returned.session.has_internal_errors(), 'Session has internal errors!'133 if verify:134 validate_run(self, returned, expect_interruption=expect_interruption, expect_session_errors=expect_session_errors)135 return returned136 @contextmanager137 def _custom_sorting(self, do_sort):138 @gossip.register('slash.tests_loaded')139 def tests_loaded(tests):140 if do_sort:141 for test in tests:142 if not test.__slash__.is_interactive():143 test.__slash__.set_sort_key(int(self._get_test_id_from_runnable(test)))144 try:145 yield146 finally:147 tests_loaded.gossip.unregister()148 def _get_test_id_from_runnable(self, test):149 return get_test_id_from_test_address(test.__slash__.address)150 @contextmanager151 def _custom_slashrc(self, path):152 if self._slashrc is not None:153 slashrc_path = os.path.join(path, self._slashrc.get_relative_path())154 else:155 slashrc_path = None156 with get_temporary_slashrc_context(slashrc_path):157 yield158 @contextmanager159 def _capture_events(self, summary):160 sys.modules['__ut__'] = summary.tracker161 try:162 yield163 finally:164 sys.modules.pop('__ut__')165 def commit(self):166 path = self._path167 if path is None:168 path = tempfile.mkdtemp()169 elif not os.path.isdir(path):170 os.makedirs(path)171 files = self._files172 if self._slashconf is not None:173 files = itertools.chain(files, [self._slashconf])174 if self._slashrc is not None:175 files = itertools.chain(files, [self._slashrc])176 # TODO: clean up paths # pylint: disable=fixme177 for file in files:178 with open(os.path.join(path, file.get_relative_path()), 'w') as f:179 formatter = CodeFormatter(f)180 file.write(formatter)181 self._last_committed_path = path182 return path183 # Shortcuts184 @property185 def num_method_tests(self):186 return self._num_method_tests187 @property188 def num_function_tests(self):189 return self._num_function_tests190 @property191 def method_tests(self):192 return [test for test in self._notified if test.is_method_test()]193 @property194 def function_tests(self):...
validation.py
Source:validation.py
1import itertools2import os3import re4import logbook5from .generator_fixture import GeneratorFixture6_logger = logbook.Logger(__name__)7def validate_run(suite, run_result, expect_interruption, expect_session_errors):8 if expect_interruption:9 assert run_result.session.results.global_result.is_interrupted(), \10 'Session global result is not marked as interrupted, even though interruption was expected'11 if expect_interruption or not run_result.session.results.is_success(allow_skips=True):12 assert run_result.exit_code != 0, '`slash run` unexpectedly returned 0'13 else:14 assert run_result.exit_code == 0, '`slash run` unexpectedly returned {}. Output: {}'.format(15 run_result.exit_code, run_result.get_console_output())16 global_result = run_result.session.results.global_result17 errors = global_result.get_errors() + global_result.get_failures()18 if expect_session_errors:19 assert errors, 'Expected session errors but found none'20 else:21 assert not errors, 'Sessions errors were not expected (Got {})'.format(errors)22 for test, results in _group_results_by_test_id(suite, run_result).items():23 _validate_single_test(test, results)24def _validate_single_test(test, results):25 param_names = {p.id: p.name for p in _find_all_parameters(test)}26 for param_values in _iter_param_value_sets(test):27 is_excluded = any((param_names[param_id], value) in test.excluded_param_values for param_id, value in param_values.items())28 for repetition in range(test.get_num_expected_repetitions()): # pylint: disable=unused-variable29 for index, result in enumerate(results):30 if _result_matches(result, param_values):31 results.pop(index)32 if is_excluded:33 assert result.is_skip()34 else:35 _validate_single_test_result(test, result)36 break37 else:38 assert False, 'Could not find parameter set {}'.format(39 param_values)40 assert not results, 'Unmatched results exist'41def _iter_param_value_sets(test):42 params = _find_all_parameters(test)43 param_ids = [p.id for p in params]44 for combination in itertools.product(*(param.values for param in params)):45 yield dict(zip(param_ids, combination))46def _find_all_parameters(func):47 params = []48 stack = [func]49 while stack:50 f = stack.pop()51 for subfixture in f.get_fixtures():52 if isinstance(subfixture, GeneratorFixture):53 params.append(subfixture)54 continue55 else:56 stack.append(subfixture)57 params.extend(f.get_parameters())58 # This function returns a list of parameters (type: Parameter) with unique ID59 # Therefore, returning list(params) is not engouth60 return list({p.id: p for p in params}.values())61def _result_matches(result, param_values):62 values = result.test_metadata.variation.values.copy()63 for param_name in list(values):64 # handle the case of a fixture with a single param, which is logically a parameter by itself65 if re.match(r'^fx_\d+.param$', param_name):66 values_name = param_name.split('_')[1].split('.')[0]67 else:68 values_name = param_name.rsplit('_', 1)[-1]69 values[values_name] = values.pop(param_name)70 return values == param_values71def _validate_single_test_result(test, result):72 expected = test.get_expected_result()73 if expected == 'ERROR':74 assert result.is_error(), 'Test did not issue error as expected'75 elif expected == 'FAIL':76 assert result.is_failure(), 'Test did not fail as expected'77 elif expected == 'SUCCESS':78 assert result.is_success(), 'Test {} unexpectedly unsuccessful:\n{}'.format(79 test.id, list(itertools.chain(result.get_errors(), result.get_failures())))80 elif expected == 'INTERRUPT':81 assert result.is_interrupted(), 'Test did not get interrupted as expected'82 elif expected == 'SKIP':83 assert result.is_skip()84 elif expected == 'NOT_RUN':85 assert result.is_not_run()86 else:87 raise NotImplementedError(88 'Unknown expected result: {!r}'.format(expected)) # pragma: no cover89def _group_results_by_test_id(suite, run_result):90 tests_by_id = dict((t.id, t) for t in suite)91 unseen = tests_by_id.copy()92 groups = {}93 for result in run_result.session.results:94 if result.test_metadata.is_interactive():95 continue96 test_id = get_test_id_from_test_address(result.test_metadata.address)97 assert tests_by_id[test_id].is_selected(), 'Test {} appears in results, although not expected!'.format(test_id)98 groups.setdefault(tests_by_id[test_id], []).append(result)99 unseen.pop(test_id, None)100 for test_id, test in list(unseen.items()):101 if not test.is_selected():102 unseen.pop(test_id, None)103 assert not unseen, 'Expected results not found ({})'.format(unseen)104 return groups105def get_test_id_from_test_address(addr):106 _, addr = os.path.splitdrive(addr)107 _, addr = addr.split(':', 1)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
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!!
Was this article helpful?
