How to use before_session_cleanup method in Slash

Best Python code snippet using slash

test_interruptions.py

Source:test_interruptions.py

1# pylint: disable=unused-argument,unused-variable,redefined-outer-name2import gossip3import pytest4import slash5import slash.hooks6from .conftest import Checkpoint7def test_interruption(interrupted_suite, interrupted_index):8    interrupted_suite.run(expect_interruption=True)9def test_interruption_added_to_result(interrupted_suite, interrupted_index):10    caught = []11    @gossip.register('slash.interruption_added')12    def interruption_added(result, exception):13        caught.append(exception)14    summary = interrupted_suite.run(expect_interruption=True)15    assert len(caught) == 116    [err] = caught              # pylint: disable=unbalanced-tuple-unpacking17    assert err.exception_type is KeyboardInterrupt18def test_interruption_triggers_gossip(request, interrupted_suite, interrupted_test):19    test_id = {'value': None}20    @gossip.register('slash.test_interrupt')21    def skip():22        test_id['value'] = slash.test.__slash__.id23    @request.addfinalizer24    def cleanup():25        skip.gossip.unregister()26    summary = interrupted_suite.run(expect_interruption=True)27    assert test_id['value'] is not None28    for result in summary.get_all_results_for_test(interrupted_test):29        assert result.test_metadata.id == test_id['value']30def test_critical_cleanups_called(interrupted_suite, interrupted_test):31    cleanup = interrupted_test.add_deferred_event(32        'slash.add_critical_cleanup', 'critical_cleanup')33    summary = interrupted_suite.run(expect_interruption=True)34    assert cleanup in summary.events35def test_non_critical_cleanups_not_called(interrupted_suite, interrupted_test):36    cleanup = interrupted_test.add_cleanup()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):...

session.py

Source:session.py

...126        if not self.scope_manager.has_active_scopes():127            return128        try:129            with handling_exceptions(swallow=True):130                hooks.before_session_cleanup()  # pylint: disable=no-member131        finally:132            with handling_exceptions(swallow=True):133                self.scope_manager.flush_remaining_scopes()134    def mark_complete(self):135        self._complete = True136    def is_complete(self):137        return self._complete138    _total_num_tests = 0139    def get_total_num_tests(self):140        """Returns the total number of tests expected to run in this session141        """142        return self._total_num_tests143    def increment_total_num_tests(self, increment):144        self._total_num_tests += increment

test_scoped_cleanups.py

Source:test_scoped_cleanups.py

...18        @slash.add_cleanup19        def cleanup():20            events.append(('test_cleanup', test_id))21    @slash.hooks.register22    def before_session_cleanup():23        events.append('before_session_cleanup')24    @slash.hooks.register25    def session_start():26        slash.add_cleanup(events.append, args=(('session_cleanup', 1),))27        slash.add_cleanup(events.append, args=(('session_cleanup', 2),))28    @slash.hooks.register29    def after_session_end():30        events.append('after_session_end')31    @slash.hooks.register32    def session_end():33        events.append('session_end')34    suite.run()35    expected_session_cleanup = [36        'before_session_cleanup',...

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.