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

Copied to Clipboard!

How to use get_num_successful method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py

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

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.