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

Copied to Clipboard!

How to use parallel_suite method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py

1import munch2import slash3import os4import signal5import sys6from vintage import get_no_deprecations_context7from .utils.suite_writer import Suite8from slash.resuming import get_tests_from_previous_session9from slash.exceptions import InteractiveParallelNotAllowed, ParallelTimeout10from slash.parallel.server import ServerStates11from slash.parallel.parallel_manager import ParallelManager12from slash import Session13from slash.loader import Loader14import time15import tempfile16import pytest17if sys.platform.startswith("win"):18    pytest.skip("does not run on windows", allow_module_level=True)19@pytest.fixture(scope='module', autouse=True)20def no_parallel_user_config(request):21    tmpdir = tempfile.mkdtemp()22    path = os.path.join(tmpdir, 'slashrc')23    os.environ["SLASH_USER_SETTINGS"] = path24    @request.addfinalizer25    def cleanup():  # pylint: disable=unused-variable26        os.rmdir(tmpdir)27        del os.environ["SLASH_USER_SETTINGS"]28#basic features of parallel29def run_specific_workers_and_tests_num(workers_num, tests_num=10):30    suite = Suite(debug_info=False, is_parallel=True)31    suite.populate(num_tests=tests_num)32    summary = suite.run(num_workers=workers_num)33    assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num34    assert summary.session.results.get_num_successful() == tests_num35    assert summary.session.results.is_success()36    return summary37def test_one_worker():38    run_specific_workers_and_tests_num(workers_num=1)39def test_many_workers():40    run_specific_workers_and_tests_num(workers_num=3, tests_num=50)41def test_zero_workers(parallel_suite):42    summary = parallel_suite.run(num_workers=0) #should act like regular run of slash, not parallel43    assert summary.session.results.is_success()44    assert summary.session.parallel_manager is None45def test_test_causes_worker_exit(parallel_suite, config_override):46    config_override("parallel.communication_timeout_secs", 2)47    parallel_suite[0].append_line("import os")48    parallel_suite[0].append_line("os._exit(0)")49    parallel_suite[0].expect_interruption()50    workers_num = 151    summary = parallel_suite.run(num_workers=workers_num, verify=False)52    assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num53    assert summary.session.results.is_interrupted()54    test_results = summary.get_all_results_for_test(parallel_suite[0])55    if test_results:56        [result] = test_results57        assert result.is_interrupted()58def test_keepalive_works(parallel_suite, config_override):59    config_override("parallel.communication_timeout_secs", 2)60    parallel_suite[0].append_line("import time")61    parallel_suite[0].append_line("time.sleep(6)")62    workers_num = 163    summary = parallel_suite.run(num_workers=workers_num)64    assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num65    assert summary.session.results.is_success()66def test_disconnected_worker_not_considered_timed_out(parallel_suite, config_override):67    config_override("parallel.communication_timeout_secs", 2)68    parallel_suite[0].append_line("import time")69    parallel_suite[0].append_line("time.sleep(6)")70    workers_num = 271    summary = parallel_suite.run(num_workers=workers_num)72    assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num73    assert summary.session.results.is_success()74def test_server_fails(parallel_suite):75    @slash.hooks.worker_connected.register  # pylint: disable=no-member, unused-argument76    def simulate_ctrl_c(session_id):  # pylint: disable=unused-variable, unused-argument77        pid = os.getpid()78        os.kill(pid, signal.SIGINT)79    @slash.hooks.session_interrupt.register  # pylint: disable=no-member80    def check_workers_and_server_down():  # pylint: disable=unused-variable81        for worker in slash.context.session.parallel_manager.workers.values():82            assert not worker.is_active()83        assert slash.context.session.parallel_manager.server.interrupted84        assert not slash.context.session.parallel_manager.server.finished_tests85    for test in parallel_suite:86        test.expect_deselect()87    results = parallel_suite.run(expect_interruption=True).session.results88    assert not results.is_success(allow_skips=True)89#test slash features with parallel90def test_test_success(parallel_suite):91    results = parallel_suite.run().session.results92    assert results.is_success()93    assert results.get_num_successful() == len(parallel_suite)94def test_test_failure(parallel_suite):95    parallel_suite[0].when_run.fail()96    summary = parallel_suite.run()97    [result] = summary.get_all_results_for_test(parallel_suite[0])98    [failures] = result.get_failures()99    assert 'AssertionError' in str(failures)100    assert 'assert False' in str(failures)101def test_stop_on_error(parallel_suite, parallel_suite_test):102    parallel_suite_test.when_run.fail()103    summary = parallel_suite.run(additional_args=['-x'], verify=False)104    [result] = summary.get_all_results_for_test(parallel_suite_test)105    assert result.is_failure()106    found_failure = False107    for result in summary.session.results:108        if result.is_failure():109            found_failure = True110            continue111        if found_failure:112            assert result.is_not_run()113    assert found_failure114def test_pass_override_conf_flag(parallel_suite):115    summary = parallel_suite.run(additional_args=['-o', 'parallel.server_port=8001'])116    results = summary.session.results117    assert results.is_success()118    assert results.get_num_successful() == len(parallel_suite)119    assert summary.session.parallel_manager.server.port == 8001120def test_test_error(parallel_suite):121    parallel_suite[0].append_line('slash.add_error()')122    parallel_suite[0].expect_error()123    summary = parallel_suite.run()124    [result] = summary.get_all_results_for_test(parallel_suite[0])125    [err] = result.get_errors()126    assert 'RuntimeError' in str(err)127    assert 'add_error() must be called' in str(err)128def test_test_interruption_causes_communication_timeout(parallel_suite, config_override):129    config_override("parallel.communication_timeout_secs", 2)130    parallel_suite[0].when_run.interrupt()131    summary = parallel_suite.run(num_workers=1, verify=False)132    [interrupted_result] = summary.get_all_results_for_test(parallel_suite[0])133    assert interrupted_result.is_interrupted()134    for result in summary.session.results:135        if result != interrupted_result:136            assert result.is_success() or result.is_not_run()137def test_test_interruption_causes_no_requests(parallel_suite, config_override):138    config_override("parallel.no_request_timeout", 2)139    parallel_suite[0].when_run.interrupt()140    summary = parallel_suite.run(num_workers=1, verify=False)141    assert summary.get_all_results_for_test(parallel_suite[0]) == []142def test_test_skips(parallel_suite):143    parallel_suite[0].add_decorator('slash.skipped("reason")')144    parallel_suite[0].expect_skip()145    results = parallel_suite.run().session.results146    assert results.is_success(allow_skips=True)147    assert results.get_num_skipped() == 1148    for result in results:149        if result.is_skip():150            assert 'reason' in result.get_skips()151def test_session_warnings(parallel_suite):152    parallel_suite[0].append_line("import warnings")153    parallel_suite[0].append_line("warnings.warn('message')")154    session_results = parallel_suite.run().session155    assert len(session_results.warnings) == 1156    [w] = session_results.warnings157    assert w.message == 'message'158def test_child_session_errors(parallel_suite):159    parallel_suite[0].expect_failure()160    parallel_suite[0].append_line("import slash")161    parallel_suite[0].append_line("slash.context.session.results.global_result.add_error('bla')")162    session_results = parallel_suite.run(num_workers=1, verify=False).session163    assert not session_results.results.is_success()164    assert session_results.parallel_manager.server.worker_error_reported165def test_child_errors_in_cleanup_are_session_errors(parallel_suite):166    parallel_suite[0].expect_failure()167    parallel_suite[0].append_line("import slash")168    parallel_suite[0].append_line("def a():")169    parallel_suite[0].append_line("   def _cleanup():")170    parallel_suite[0].append_line("      slash.add_error('Session cleanup')")171    parallel_suite[0].append_line("   slash.add_cleanup(_cleanup, scope='session')")172    parallel_suite[0].append_line("a()")173    session_results = parallel_suite.run(num_workers=1, verify=False).session174    assert not session_results.results.is_success()175    assert session_results.parallel_manager.server.worker_error_reported176def test_child_fatal_error_terminates_session(parallel_suite):177    parallel_suite[0].expect_failure()178    parallel_suite[0].append_line("import slash")179    parallel_suite[0].append_line("slash.add_error('bla').mark_fatal()")180    session_results = parallel_suite.run(num_workers=1, verify=False).session181    first_result = session_results.results[0]182    assert first_result.is_error() and first_result.has_fatal_errors()183    assert session_results.results.get_num_not_run() == len(parallel_suite) - 1184def test_traceback_vars(parallel_suite):185    #code to be inserted:186        #     def test_traceback_frames():187        #     num = 0188        #     a()189        #190        # def a():191        #     x=1192        #     assert False193    parallel_suite[0].append_line("def a():")194    parallel_suite[0].append_line("   num = 0")195    parallel_suite[0].append_line("   b()")196    parallel_suite[0].append_line("def b():")197    parallel_suite[0].append_line("   x=1")198    parallel_suite[0].append_line("   assert False")199    parallel_suite[0].append_line("a()")200    parallel_suite[0].expect_failure()201    summary = parallel_suite.run(num_workers=1)202    results = summary.session.results203    found_failure = 0204    for result in results:205        if result.is_failure():206            found_failure += 1207            assert len(result.get_failures()) == 1208            assert len(result.get_failures()[0].traceback.frames) == 3209            with get_no_deprecations_context():210                assert 'x' in result.get_failures()[0].traceback.frames[2].locals211                assert 'num' in result.get_failures()[0].traceback.frames[1].locals212    assert found_failure == 1213def test_result_data_not_picklable(parallel_suite):214    parallel_suite[0].append_line("import socket")215    parallel_suite[0].append_line("s = socket.socket()")216    parallel_suite[0].append_line("slash.context.result.data.setdefault('socket', s)")217    summary = parallel_suite.run()218    [result] = summary.get_all_results_for_test(parallel_suite[0])219    assert result.data == {}220def test_result_data_is_picklable(parallel_suite):221    parallel_suite[0].append_line("slash.context.result.data.setdefault('num', 1)")222    summary = parallel_suite.run()223    [result] = summary.get_all_results_for_test(parallel_suite[0])224    assert 'num' in result.data225    assert result.data['num'] == 1226def test_result_details_not_picklable(parallel_suite):227    parallel_suite[0].append_line("import socket")228    parallel_suite[0].append_line("s = socket.socket()")229    parallel_suite[0].append_line("slash.context.result.details.append('socket', s)")230    summary = parallel_suite.run()231    [result] = summary.get_all_results_for_test(parallel_suite[0])232    assert result.details.all() == {}233def test_result_details_is_picklable(parallel_suite):234    parallel_suite[0].append_line("slash.context.result.details.append('num', 1)")235    summary = parallel_suite.run()236    [result] = summary.get_all_results_for_test(parallel_suite[0])237    details = result.details.all()238    assert 'num' in details239    assert details['num'] == [1]240def test_parameters(parallel_suite):241    parallel_suite[0].add_parameter(num_values=1)242    summary = parallel_suite.run()243    assert summary.session.results.is_success()244def test_requirements(parallel_suite):245    parallel_suite[0].add_decorator('slash.requires(False)')246    parallel_suite[0].expect_skip()247    results = parallel_suite.run().session.results248    assert results.get_num_skipped() == 1249    assert results.get_num_successful() == len(parallel_suite) - 1250    assert results.is_success(allow_skips=True)251def test_is_test_code(parallel_suite):252    parallel_suite[0].when_run.error()253    summary = parallel_suite.run()254    [result] = summary.get_all_results_for_test(parallel_suite[0])255    [err] = result.get_errors()256    assert err.traceback.frames[-1].is_in_test_code()257    error_json = err.traceback.to_list()258    assert error_json[-1]['is_in_test_code']259def test_parallel_resume(parallel_suite):260    parallel_suite[0].when_run.fail()261    result = parallel_suite.run()262    resumed = get_tests_from_previous_session(result.session.id)263    assert len(resumed) == 1264def test_parallel_symlinks(parallel_suite, logs_dir):   # pylint: disable=unused-argument265    files_dir = logs_dir.join("files")266    links_dir = logs_dir.join("links")267    session = parallel_suite.run(additional_args=['-l', str(files_dir)]).session268    session_log_file = files_dir.join(session.id, "session.log")269    assert session.results.is_success()270    assert session_log_file.check()271    assert links_dir.join("last-session").readlink() == session_log_file272    assert links_dir.join("last-session-dir").readlink() == session_log_file.dirname273    worker_session_ids = session.parallel_manager.server.worker_session_ids274    file_names = [x.basename for x in links_dir.join("last-session-dir").listdir()]275    assert 'worker_1' in file_names276    for file_name in links_dir.join("last-session-dir").listdir():277        if file_name.islink() and 'worker' in file_name.basename:278            last_token = file_name.readlink().split('/')[-1]279            assert last_token in worker_session_ids280            assert os.path.isdir(file_name.readlink())281def test_parallel_interactive_fails(parallel_suite):282    summary = parallel_suite.run(additional_args=['-i'], verify=False)283    results = list(summary.session.results.iter_all_results())284    assert len(results) == 1285    error = results[0].get_errors()[0]286    assert error.exception_type == InteractiveParallelNotAllowed287def test_children_session_ids(parallel_suite):288    summary = parallel_suite.run()289    assert summary.session.results.is_success()290    session_ids = summary.session.parallel_manager.server.worker_session_ids291    expected_session_ids = ["{}_1".format(summary.session.id.split('_')[0])]292    assert session_ids == expected_session_ids293def test_timeout_no_request_to_server(config_override, runnable_test_dir):294    config_override("parallel.no_request_timeout", 1)295    with Session() as session:296        runnables = Loader().get_runnables(str(runnable_test_dir))297        parallel_manager = ParallelManager([])298        session.parallel_manager = parallel_manager299        parallel_manager.start_server_in_thread(runnables)300        parallel_manager.try_connect()301        parallel_manager.server.state = ServerStates.SERVE_TESTS302        with slash.assert_raises(ParallelTimeout) as caught:303            parallel_manager.start()304        assert 'No request sent to server' in caught.exception.args[0]305def test_children_not_connected_timeout(runnable_test_dir, config_override):306    config_override("parallel.worker_connect_timeout", 0)307    config_override("parallel.num_workers", 1)308    with Session() as session:309        runnables = Loader().get_runnables(str(runnable_test_dir))310        parallel_manager = ParallelManager(munch.Munch(argv=[], cmd="run"))311        session.parallel_manager = parallel_manager312        parallel_manager.start_server_in_thread(runnables)313        time.sleep(0.1)314        with slash.assert_raises(ParallelTimeout) as caught:315            parallel_manager.wait_all_workers_to_connect()316        assert 'Not all clients connected' in caught.exception.args[0]317def test_worker_error_logs(parallel_suite, config_override):318    config_override("parallel.communication_timeout_secs", 2)319    parallel_suite[0].when_run.interrupt()320    summary = parallel_suite.run(num_workers=1, verify=False)321    [interrupted_result] = summary.get_all_results_for_test(parallel_suite[0])322    assert interrupted_result.is_interrupted()323    for result in summary.session.results:324        if result != interrupted_result:325            assert result.is_success() or result.is_not_run()326    file_path = os.path.join(summary.session.parallel_manager.workers_error_dircetory, 'errors-worker-1.log')327    assert os.path.isfile(file_path)328    with open(file_path) as error_file:329        line = error_file.readline()330        assert 'interrupted' in line331def test_shuffle(parallel_suite):332    @slash.hooks.tests_loaded.register   # pylint: disable=no-member333    def tests_loaded(tests):   # pylint: disable=unused-variable334        for index, test in enumerate(reversed(tests)):335            test.__slash__.set_sort_key(index)336    parallel_suite.run()337def test_server_hanging_dont_cause_worker_timeouts(config_override):338    config_override("parallel.no_request_timeout", 5)339    @slash.hooks.test_distributed.register   # pylint: disable=no-member340    def test_distributed(test_logical_id, worker_session_id):   # pylint: disable=unused-variable, unused-argument341        time.sleep(6)342    suite = Suite(debug_info=False, is_parallel=True)343    suite.populate(num_tests=2)344    suite.run(num_workers=1)345def test_force_worker(parallel_suite):346    for test in parallel_suite:347        test.append_line("from slash import config")348        test.append_line("slash.context.result.data.setdefault('worker_id', config.root.parallel.worker_id)")349    @slash.hooks.tests_loaded.register  # pylint: disable=no-member, unused-argument350    def tests_loaded(tests):  # pylint: disable=unused-variable, unused-argument351        if slash.utils.parallel_utils.is_parent_session():352            from slash import ctx353            workers = ctx.session.parallel_manager.workers354            for index in range(len(tests)):355                worker_id = '1' if index%2 == 0 else '2'356                workers[worker_id].force_test(index)357    summary = parallel_suite.run(num_workers=2)358    assert summary.session.results.is_success()359    for index, test in enumerate(parallel_suite):360        [result] = summary.get_all_results_for_test(test)361        assert result.data['worker_id'] if index%2 == 0 else '2'362def test_force_on_one_worker(parallel_suite):363    @slash.hooks.tests_loaded.register  # pylint: disable=no-member, unused-argument364    def tests_loaded(tests):  # pylint: disable=unused-variable, unused-argument365        if slash.utils.parallel_utils.is_parent_session():366            from slash import ctx367            worker = list(ctx.session.parallel_manager.workers.values())[0]368            for index in range(len(tests)):369                worker.force_test(index)370    summary = parallel_suite.run(num_workers=2)371    assert summary.session.results.is_success()372@pytest.mark.parametrize('num_workers', [1, 2])373def test_exclude_on_one_worker(parallel_suite, config_override, num_workers):374    config_override("parallel.no_request_timeout", 2)375    @slash.hooks.tests_loaded.register  # pylint: disable=no-member, unused-argument376    def tests_loaded(tests):  # pylint: disable=unused-variable, unused-argument377        if slash.utils.parallel_utils.is_parent_session():378            from slash import ctx379            worker = list(ctx.session.parallel_manager.workers.values())[0]380            for index in range(len(tests)):381                worker.exclude_test(index)382    if num_workers == 1:383        summary = parallel_suite.run(num_workers=num_workers, verify=False)384        assert summary.session.results.get_num_started() == 0385    else:386        summary = parallel_suite.run(num_workers=num_workers)387        assert summary.session.results.is_success()388@pytest.mark.parametrize('use_test_index', [True, False])389def test_exclude_test_on_all_workers_causes_timeout(parallel_suite, config_override, use_test_index):390    config_override("parallel.no_request_timeout", 2)391    @slash.hooks.tests_loaded.register  # pylint: disable=no-member, unused-argument392    def tests_loaded(tests):  # pylint: disable=unused-variable, unused-argument393        if slash.utils.parallel_utils.is_parent_session():394            from slash import ctx395            for worker in ctx.session.parallel_manager.workers.values():396                if use_test_index:397                    worker.exclude_test(0)398                else:399                    worker.exclude_test(tests[0])400    summary = parallel_suite.run(num_workers=2, verify=False)401    assert summary.session.results.get_num_started() == len(parallel_suite) - 1402    assert not summary.get_all_results_for_test(parallel_suite[0])403def test_exclude_and_force_on_same_worker_raises_runtime_err(parallel_suite):404    @slash.hooks.tests_loaded.register  # pylint: disable=no-member, unused-argument405    def tests_loaded(tests):  # pylint: disable=unused-variable, unused-argument406        if slash.utils.parallel_utils.is_parent_session():407            from slash import ctx408            worker = list(ctx.session.parallel_manager.workers.values())[0]409            worker.exclude_test(0)410            worker.force_test(0)411    summary = parallel_suite.run(num_workers=2, verify=False)...

runner.py

Source:runner.py

1# -*- coding=utf-8 -*-2import re3import sys4import unittest5from django.conf import settings6from django.test.runner import reorder_suite, DiscoverRunner7from datetime import datetime8WITH_HTML_REPORT = getattr(settings, 'TEST_HTML_REPORT', False)9if WITH_HTML_REPORT:10    try:11        from pyunitreport import HTMLTestRunner12        from ttoolly.html_report.report import CustomHtmlTestResult13    except ImportError:14        raise Exception('For html reports you should install pyunitreport:\n    pip install PyUnitReport')15if getattr(settings, 'TEST_RUNNER_PARENT', '') == 'xmlrunner.extra.djangotestrunner.XMLTestRunner':16    from xmlrunner.result import _XMLTestResult, safe_unicode, _TestInfo17    from xmlrunner.runner import XMLTestRunner18    original_report_testcase = _XMLTestResult._report_testcase19    @staticmethod20    def _report_testcase(test_result, xml_testsuite, xml_document):21        original_report_testcase(test_result, xml_testsuite, xml_document)22        testcase = xml_testsuite.childNodes[-1]23        description = xml_document.createElement('description')24        testcase.appendChild(description)25        description_text = safe_unicode(test_result.test_description)26        _XMLTestResult._createCDATAsections(xml_document, description, description_text)27        tags = xml_document.createElement('tags')28        for tag_name in test_result.tags:29            tag = xml_document.createElement('tag')30            tag.appendChild(xml_document.createTextNode(tag_name))31            tags.appendChild(tag)32        testcase.appendChild(tags)33    _XMLTestResult._report_testcase = _report_testcase34    class XMLInfoClass(_TestInfo):35        def __init__(self, test_result, test_method, *args, **kwargs):36            super().__init__(test_result, test_method, *args, **kwargs)37            tags = set(getattr(test_method, 'tags', set()))38            test_fn_name = getattr(test_method, '_testMethodName', str(test_method))39            test_fn = getattr(test_method, test_fn_name, test_method)40            test_fn_tags = set(getattr(test_fn, 'tags', set()))41            self.tags = tags.union(test_fn_tags)42    class CustomXMLTestRunner(XMLTestRunner):43        def _make_result(self):44            return self.resultclass(self.stream, self.descriptions, self.verbosity, self.elapsed_times,45                                    infoclass=XMLInfoClass)46def get_runner():47    test_runner_class = getattr(settings, 'TEST_RUNNER_PARENT', None)48    if not test_runner_class:49        return DiscoverRunner50    test_path = test_runner_class.split('.')51    test_module = __import__('.'.join(test_path[:-1]), {}, {}, str(test_path[-1]))52    test_runner = getattr(test_module, test_path[-1])53    return test_runner54ParentRunner = get_runner()55def filter_suite_by_decorators(suite, verbosity=1):56    new_suite = unittest.TestSuite()57    for el in suite:58        need_skip = False59        fn = getattr(el, el._testMethodName)60        if getattr(fn, '__unittest_skip__', False):61            need_skip = True62            skip_text = fn.__unittest_skip_why__63        else:64            for decorator in reversed(getattr(fn, 'decorators', ())):65                check = getattr(decorator, 'check', None)66                if check:67                    need_skip = not(check(el))68                    if need_skip:69                        skip_text = decorator.skip_text70                        break71        if not need_skip:72            new_suite.addTest(el)73        elif verbosity > 1:74            st = unittest.runner._WritelnDecorator(sys.stderr)75            st.write('Skip {test_name}: {skip_text}\n'.format(test_name='.'.join([el.__class__.__module__,76                                                                                  el.__class__.__name__,77                                                                                  el._testMethodName]),78                                                              skip_text=skip_text))79    return new_suite80class RegexpTestSuiteRunner(ParentRunner):81    parallel = 182    def get_test_runner(self):83        if WITH_HTML_REPORT:84            return HTMLTestRunner85        if getattr(settings, 'TEST_RUNNER_PARENT', '') == 'xmlrunner.extra.djangotestrunner.XMLTestRunner':86            return CustomXMLTestRunner87        return ParentRunner.test_runner88    mro_names = [m.__name__ for m in ParentRunner.__mro__]89    def __init__(self, *args, **kwargs):90        super(RegexpTestSuiteRunner, self).__init__(*args, **kwargs)91        self.tags_rule = kwargs['tags_rule']92        if self.tags_rule:93            self.tags = []94            self.exclude_tags = []95        self.parallelism = [int(el) for el in kwargs['parallelism'].split('/')] if kwargs['parallelism'] else None96        self.test_runner = self.get_test_runner()97    @classmethod98    def add_arguments(cls, parser):99        super(RegexpTestSuiteRunner, cls).add_arguments(parser)100        parser.add_argument(101            '--tags', action='store', dest='tags_rule',102            help='Tags boolean rule. Example: "low AND middle AND NOT high"',103        )104        parser.add_argument(105            '--parallelism', dest='parallelism', default=None,106            help='Part of tests (if parallel by ci). For example 2/5 - second part of five. Will be ignored if parallel > 1',107        )108    def convert_by_parallel(self, suite):109        if self.parallel > 1 or not self.parallelism or self.parallelism[1] == 1:110            return suite111        def get_chunk(count, chunk_n):112            if chunk_n >= count:113                return []114            length = len(suite._tests)115            chunk_len, additional = divmod(length, count)116            return suite._tests[chunk_n * chunk_len:(chunk_n + 1) * chunk_len + (additional if chunk_n == count - 1 else 0)]117        return unittest.TestSuite(get_chunk(self.parallelism[1], self.parallelism[0] - 1))118    def get_resultclass(self):119        if WITH_HTML_REPORT:120            return CustomHtmlTestResult121        return super(RegexpTestSuiteRunner, self).get_resultclass()122    def build_suite(self, test_labels, extra_tests=None, **kwargs):123        real_parallel = self.parallel124        self.parallel = 1125        labels_for_suite = []126        for label in test_labels:127            if re.findall(r'(^[\w\d_]+(?:\.[\w\d_]+)*$)', label) == [label]:128                labels_for_suite.append(label)129            else:130                while label:131                    label = '.'.join(label.split('.')[:-1])132                    if re.findall(r'(^[\w\d_]+(?:\.[\w\d_]+)*$)', label) == [label]:133                        labels_for_suite.append(label)134                        break135        full_suite = super(RegexpTestSuiteRunner, self).build_suite(labels_for_suite, extra_tests=None, **kwargs)136        my_suite = unittest.TestSuite()137        labels_for_suite = []138        if test_labels:139            full_re = []140            for label in test_labels:141                if re.findall(r'(^[\w\d_]+(?:\.[\w\d_]+)*$)', label) == [label]:142                    labels_for_suite.append(label)143                    continue144                text_for_re = label.replace('.', '\.').replace('*', '[^\.]+?')145                if 'DiscoverRunner' in self.mro_names:146                    if len(label.split('.')) > 3:147                        text_for_re += '$'148                    else:149                        text_for_re += '\..+$'150                full_re.append(text_for_re)151            full_re = '(^' + ')|(^'.join(full_re) + ')' if full_re else ''152            for el in full_suite._tests:153                module_name = el.__module__154                full_name = [module_name, el.__class__.__name__, el._testMethodName]155                full_name = '.'.join(full_name)156                if (full_re and re.findall(r'%s' % full_re, full_name)):157                    my_suite.addTest(el)158        else:159            my_suite = full_suite160        if labels_for_suite:161            my_suite.addTests(ParentRunner.build_suite(self, labels_for_suite, extra_tests=None, **kwargs))162        if self.tags_rule:163            from .for_runner import algebra164            parsed = algebra.parse(self.tags_rule).simplify()165            my_suite = filter_tests_by_tags_rule(my_suite, parsed)166        if getattr(settings, 'TEST_SKIP_SILENT', False):167            my_suite = filter_suite_by_decorators(my_suite, self.verbosity)168        suite = reorder_suite(my_suite, (unittest.TestCase,))169        self.parallel = real_parallel170        if self.parallel > 1:171            parallel_suite = self.parallel_test_suite(suite, self.parallel, self.failfast)172            # Since tests are distributed across processes on a per-TestCase173            # basis, there's no need for more processes than TestCases.174            parallel_units = len(parallel_suite.subsuites)175            if self.parallel > parallel_units:176                self.parallel = parallel_units177            # If there's only one TestCase, parallelization isn't needed.178            if self.parallel > 1:179                suite = parallel_suite180        return self.convert_by_parallel(suite)181    def run_suite(self, suite, **kwargs):182        if WITH_HTML_REPORT:183            resultclass = self.get_resultclass()184            result = self.test_runner(185                output=getattr(settings, 'TEST_REPORT_OUTPUT_DIR', datetime.now().strftime('%Y-%m-%d %H-%M-%S')),186                verbosity=self.verbosity,187                failfast=self.failfast,188                resultclass=resultclass,189            ).run(suite)190        else:191            result = super(RegexpTestSuiteRunner, self).run_suite(suite, **kwargs)192        if self.verbosity > 2 and (result.errors or result.failures):193            st = unittest.runner._WritelnDecorator(sys.stderr)194            st.write('\n' + '*' * 29 + ' Run failed ' + '*' * 29 + '\n\n')195            st.write('python manage.py test %s' % ' '.join(196                ['.'.join([test.__class__.__module__, test.__class__.__name__, test._testMethodName]) for test, _ in197                 result.errors + result.failures if hasattr(test, '_testMethodName')]) + '\n\n')198            st.write('*' * 70 + '\n\n')199        return result200def filter_tests_by_tags_rule(suite, parsed_rule):201    suite_class = type(suite)202    filtered_suite = suite_class()203    for test in suite:204        if isinstance(test, suite_class):205            filtered_suite.addTests(filter_tests_by_tags_rule(test, parsed_rule))206        else:207            test_tags = set(getattr(test, 'tags', set()))208            test_fn_name = getattr(test, '_testMethodName', str(test))209            test_fn = getattr(test, test_fn_name, test)210            test_fn_tags = set(getattr(test_fn, 'tags', set()))211            all_tags = test_tags.union(test_fn_tags)212            if parsed_rule.__bool__(all_tags):213                filtered_suite.addTest(test)...

conftest.py

Source:conftest.py

...124    if not is_last_test:125        _ = parallel_suite.add_test(type=test_type)126    return returned127@pytest.fixture128def parallel_suite():129    returned = Suite(debug_info=False, is_parallel=True)130    returned.populate()131    return returned132@pytest.fixture133def runnable_test_dir(tmpdir):134    tests_dir = tmpdir.join(str(uuid4()))135    filename = str(uuid4()).replace('-', '') + '.py'136    with tests_dir.join(filename).open('w', ensure=True) as f:137        f.write('def test_something():\n    pass')138    return tests_dir139@pytest.fixture140def slash_session():141    return slash.Session()142@pytest.fixture...

test_xunit_plugin.py

Source:test_xunit_plugin.py

...104                raise NotImplementedError()  # pragma: no cover105    return func106@pytest.mark.skipif(sys.platform == 'win32', reason="does not run on windows")107@pytest.mark.parametrize('action', ['skipped', 'error', 'failure', 'success'])108def test_xunit_parallel_suite(parallel_suite, xunit_filename, action):109    test = parallel_suite[3]110    if action == 'skipped':111        test.when_run.skip()112    elif action == 'failure':113        test.when_run.fail()114    elif action == 'error':115        test.when_run.raise_exception()116    else:117        assert action == 'success', 'Unsupported action'118    parallel_suite.run()119    root = validate_xml(xunit_filename, suite=parallel_suite)120    if action != 'success':121        for child in root:122            if len(child) > 0: # pylint: disable=len-as-condition...

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.