Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.