How to use get_num_successful method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py Github

copy

Full Screen

...25 suite = Suite(debug_info=False, is_parallel=True)26 suite.populate(num_tests=tests_num)27 summary = suite.run(num_workers=workers_num)28 assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num29 assert summary.session.results.get_num_successful() == tests_num30 assert summary.session.results.is_success()31 return summary32def test_one_worker():33 run_specific_workers_and_tests_num(workers_num=1)34def test_many_workers():35 run_specific_workers_and_tests_num(workers_num=3, tests_num=50)36def test_zero_workers(parallel_suite):37 summary = parallel_suite.run(num_workers=0) #should act like regular run of slash, not parallel38 assert summary.session.results.is_success()39 assert summary.session.parallel_manager is None40def test_test_causes_worker_exit(parallel_suite, config_override):41 config_override("parallel.communication_timeout_secs", 2)42 parallel_suite[0].append_line("import os")43 parallel_suite[0].append_line("os._exit(0)")44 parallel_suite[0].expect_interruption()45 workers_num = 146 summary = parallel_suite.run(num_workers=workers_num, verify=False)47 assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num48 [result] = summary.get_all_results_for_test(parallel_suite[0])49 assert result.is_interrupted()50def test_keepalive_works(parallel_suite, config_override):51 config_override("parallel.communication_timeout_secs", 2)52 parallel_suite[0].append_line("import time")53 parallel_suite[0].append_line("time.sleep(6)")54 workers_num = 155 summary = parallel_suite.run(num_workers=workers_num)56 assert len(summary.session.parallel_manager.server.worker_session_ids) == workers_num57 assert summary.session.results.is_success()58def test_server_fails(parallel_suite):59 @slash.hooks.worker_connected.register # pylint: disable=no-member, unused-argument60 def simulate_ctrl_c(session_id): # pylint: disable=unused-variable, unused-argument61 pid = os.getpid()62 os.kill(pid, signal.SIGINT)63 @slash.hooks.session_interrupt.register # pylint: disable=no-member64 def check_workers_and_server_down(): # pylint: disable=unused-variable65 for worker in slash.context.session.parallel_manager.workers.values():66 ret = worker.poll()67 assert not ret is None68 assert slash.context.session.parallel_manager.server.interrupted69 assert not slash.context.session.parallel_manager.server.finished_tests70 for test in parallel_suite:71 test.expect_deselect()72 parallel_suite.run(expect_interruption=True)73#test slash features with parallel74def test_test_success(parallel_suite):75 results = parallel_suite.run().session.results76 assert results.is_success()77 assert results.get_num_successful() == len(parallel_suite)78def test_test_failure(parallel_suite):79 parallel_suite[0].when_run.fail()80 summary = parallel_suite.run()81 [result] = summary.get_all_results_for_test(parallel_suite[0])82 [failures] = result.get_failures()83 assert 'AssertionError' in str(failures)84 assert 'assert False' in str(failures)85def test_stop_on_error(parallel_suite, parallel_suite_test):86 parallel_suite_test.when_run.fail()87 summary = parallel_suite.run(additional_args=['-x'], verify=False)88 [result] = summary.get_all_results_for_test(parallel_suite_test)89 assert result.is_failure()90 found_failure = False91 for result in summary.session.results:92 if result.is_failure():93 found_failure = True94 continue95 if found_failure:96 assert result.is_not_run()97 assert found_failure98def test_pass_override_conf_flag(parallel_suite):99 summary = parallel_suite.run(additional_args=['-o', 'parallel.server_port=8001'])100 results = summary.session.results101 assert results.is_success()102 assert results.get_num_successful() == len(parallel_suite)103 assert summary.session.parallel_manager.server.port == 8001104def test_test_error(parallel_suite):105 parallel_suite[0].append_line('slash.add_error()')106 parallel_suite[0].expect_error()107 summary = parallel_suite.run()108 [result] = summary.get_all_results_for_test(parallel_suite[0])109 [err] = result.get_errors()110 assert 'RuntimeError' in str(err)111 assert 'add_error() must be called' in str(err)112def test_test_interruption_causes_communication_timeout(parallel_suite, config_override):113 config_override("parallel.communication_timeout_secs", 2)114 parallel_suite[0].when_run.interrupt()115 summary = parallel_suite.run(num_workers=1, verify=False)116 [interrupted_result] = summary.get_all_results_for_test(parallel_suite[0])117 assert interrupted_result.is_interrupted()118 for result in summary.session.results:119 if result != interrupted_result:120 assert result.is_success() or result.is_not_run()121def test_test_interruption_causes_no_requests(parallel_suite, config_override):122 config_override("parallel.no_request_timeout", 2)123 parallel_suite[0].when_run.interrupt()124 summary = parallel_suite.run(num_workers=1, verify=False)125 assert summary.get_all_results_for_test(parallel_suite[0]) == []126def test_test_skips(parallel_suite):127 parallel_suite[0].add_decorator('slash.skipped("reason")')128 parallel_suite[0].expect_skip()129 results = parallel_suite.run().session.results130 assert results.is_success(allow_skips=True)131 assert results.get_num_skipped() == 1132 for result in results:133 if result.is_skip():134 assert 'reason' in result.get_skips()135def test_session_warnings(parallel_suite):136 parallel_suite[0].append_line("import warnings")137 parallel_suite[0].append_line("warnings.warn('message')")138 session_results = parallel_suite.run().session139 assert len(session_results.warnings) == 1140 [w] = session_results.warnings141 assert w.message == 'message'142def test_traceback_vars(parallel_suite):143 #code to be inserted:144 # def test_traceback_frames():145 # num = 0146 # a()147 #148 # def a():149 # x=1150 # assert False151 parallel_suite[0].append_line("def a():")152 parallel_suite[0].append_line(" num = 0")153 parallel_suite[0].append_line(" b()")154 parallel_suite[0].append_line("def b():")155 parallel_suite[0].append_line(" x=1")156 parallel_suite[0].append_line(" assert False")157 parallel_suite[0].append_line("a()")158 parallel_suite[0].expect_failure()159 summary = parallel_suite.run(num_workers=1)160 results = summary.session.results161 found_failure = 0162 for result in results:163 if result.is_failure():164 found_failure += 1165 assert len(result.get_failures()) == 1166 assert len(result.get_failures()[0].traceback.frames) == 3167 assert found_failure == 1168def test_result_data_not_picklable(parallel_suite):169 parallel_suite[0].append_line("import socket")170 parallel_suite[0].append_line("s = socket.socket()")171 parallel_suite[0].append_line("slash.context.result.data.setdefault('socket', s)")172 summary = parallel_suite.run()173 [result] = summary.get_all_results_for_test(parallel_suite[0])174 assert result.data == {}175def test_result_data_is_picklable(parallel_suite):176 parallel_suite[0].append_line("slash.context.result.data.setdefault('num', 1)")177 summary = parallel_suite.run()178 [result] = summary.get_all_results_for_test(parallel_suite[0])179 assert 'num' in result.data180 assert result.data['num'] == 1181def test_result_details_not_picklable(parallel_suite):182 parallel_suite[0].append_line("import socket")183 parallel_suite[0].append_line("s = socket.socket()")184 parallel_suite[0].append_line("slash.context.result.details.append('socket', s)")185 summary = parallel_suite.run()186 [result] = summary.get_all_results_for_test(parallel_suite[0])187 assert result.details.all() == {}188def test_result_details_is_picklable(parallel_suite):189 parallel_suite[0].append_line("slash.context.result.details.append('num', 1)")190 summary = parallel_suite.run()191 [result] = summary.get_all_results_for_test(parallel_suite[0])192 details = result.details.all()193 assert 'num' in details194 assert details['num'] == [1]195def test_parameters(parallel_suite):196 parallel_suite[0].add_parameter(num_values=1)197 summary = parallel_suite.run()198 assert summary.session.results.is_success()199def test_requirements(parallel_suite):200 parallel_suite[0].add_decorator('slash.requires(False)')201 parallel_suite[0].expect_skip()202 results = parallel_suite.run().session.results203 assert results.get_num_skipped() == 1204 assert results.get_num_successful() == len(parallel_suite) - 1205 assert results.is_success(allow_skips=True)206def test_is_test_code(parallel_suite):207 parallel_suite[0].when_run.error()208 summary = parallel_suite.run()209 [result] = summary.get_all_results_for_test(parallel_suite[0])210 [err] = result.get_errors()211 assert err.traceback.frames[-1].is_in_test_code()212 error_json = err.traceback.to_list()213 assert error_json[-1]['is_in_test_code']214def test_parallel_resume(parallel_suite):215 parallel_suite[0].when_run.fail()216 result = parallel_suite.run()217 resumed = get_tests_to_resume(result.session.id)218 assert len(resumed) == 1...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful