KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
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?
