Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_all_results_for_test method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py

...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) == 1219def test_parallel_symlinks(parallel_suite, logs_dir):   # pylint: disable=unused-argument220    files_dir = logs_dir.join("files")221    links_dir = logs_dir.join("links")222    session = parallel_suite.run(additional_args=['-l', str(files_dir)]).session223    session_log_file = files_dir.join(session.id, "session.log")224    assert session.results.is_success()225    assert session_log_file.check()226    assert links_dir.join("last-session").readlink() == session_log_file227    assert links_dir.join("last-session-dir").readlink() == session_log_file.dirname228    worker_session_ids = session.parallel_manager.server.worker_session_ids229    file_names = [x.basename for x in links_dir.join("last-session-dir").listdir()]230    assert 'worker_1' in file_names231    for file_name in links_dir.join("last-session-dir").listdir():232        if file_name.islink() and 'worker' in file_name.basename:233            last_token = file_name.readlink().split('/')[-1]234            assert last_token in worker_session_ids235            assert os.path.isdir(file_name.readlink())236def test_parallel_interactive_fails(parallel_suite):237    summary = parallel_suite.run(additional_args=['-i'], verify=False)238    results = list(summary.session.results.iter_all_results())239    assert len(results) == 1240    error = results[0].get_errors()[0]241    assert error.exception_type == InteractiveParallelNotAllowed242def test_children_session_ids(parallel_suite):243    summary = parallel_suite.run()244    assert summary.session.results.is_success()245    session_ids = summary.session.parallel_manager.server.worker_session_ids246    expected_session_ids = ["{}_1".format(summary.session.id.split('_')[0])]247    assert session_ids == expected_session_ids248def test_timeout_no_request_to_server(config_override, runnable_test_dir):249    config_override("parallel.no_request_timeout", 1)250    with Session():251        runnables = Loader().get_runnables(str(runnable_test_dir))252        parallel_manager = ParallelManager([])253        parallel_manager.start_server_in_thread(runnables)254        parallel_manager.server.state = ServerStates.SERVE_TESTS255        with slash.assert_raises(ParallelTimeout) as caught:256            parallel_manager.start()257        assert 'No request sent to server' in caught.exception.args[0]258def test_children_not_connected_timeout(runnable_test_dir, config_override):259    config_override("parallel.worker_connect_timeout", 0)260    config_override("parallel.num_workers", 1)261    with Session():262        runnables = Loader().get_runnables(str(runnable_test_dir))263        parallel_manager = ParallelManager([])264        parallel_manager.start_server_in_thread(runnables)265        time.sleep(0.1)266        with slash.assert_raises(ParallelTimeout) as caught:267            parallel_manager.wait_all_workers_to_connect()268        assert caught.exception.args[0] == 'Not all clients connected'269def test_worker_error_logs(parallel_suite, config_override):270    config_override("parallel.communication_timeout_secs", 2)271    parallel_suite[0].when_run.interrupt()272    summary = parallel_suite.run(num_workers=1, verify=False)273    [interrupted_result] = summary.get_all_results_for_test(parallel_suite[0])274    assert interrupted_result.is_interrupted()275    for result in summary.session.results:276        if result != interrupted_result:277            assert result.is_success() or result.is_not_run()278    file_path = os.path.join(summary.session.parallel_manager.workers_error_dircetory, 'errors-worker-1.log')279    assert os.path.isfile(file_path)280    with open(file_path) as error_file:281        line = error_file.readline()...

test_ci_links_plugin.py

Source:test_ci_links_plugin.py

...16    manager.activate('ci links')17    test = suite.add_test()18    test.when_run.fail()19    summary = suite.run()20    result = summary.get_all_results_for_test(test)[0]21    log_link = result.details.all().get('log_link')22    assert log_link is None23def test_build_url_defined(suite, build_url):24    manager.install(Plugin(), is_internal=False)25    manager.activate('ci links')26    test = suite.add_test()27    test.when_run.fail()28    summary = suite.run()29    result = summary.get_all_results_for_test(test)[0]30    log_link = result.details.all().get('log_link')31    expected_log_link = '/'.join(32        (33            build_url.rstrip('/'),34            'artifact',35            *result.get_log_path().split(os.sep)36        )37    )38    assert log_link == expected_log_link39def test_adds_link_for_passing_test(suite, build_url):40    manager.install(Plugin(), is_internal=False)41    manager.activate('ci links')42    summary = suite.run()43    for test in suite:44        result = summary.get_all_results_for_test(test)[0]45        log_link = result.details.all().get('log_link')46        expected_log_link = '/'.join(47            (48                build_url.rstrip('/'),49                'artifact',50                *result.get_log_path().split(os.sep)51            )52        )53        assert log_link == expected_log_link54def test_nondefault_build_url_environment_variable(55        suite, config_override, build_url_env_var):56    env_var_name, env_var_value = build_url_env_var57    manager.install(Plugin(), is_internal=False)58    config_override(59        'plugin_config.ci_links.build_url_environment_variable',60        env_var_name61    )62    manager.activate('ci links')63    test = suite.add_test()64    test.when_run.fail()65    summary = suite.run()66    result = summary.get_all_results_for_test(test)[0]67    log_link = result.details.all().get('log_link')68    expected_log_link = '/'.join(69        (70            env_var_value.rstrip('/'),71            'artifact',72            *result.get_log_path().split(os.sep)73        )74    )75    assert log_link == expected_log_link76@pytest.mark.parametrize('template', LOG_PATH_TEMPLATES)77def test_nondefault_link_template(78        suite, config_override, build_url, template):79    manager.install(Plugin(), is_internal=False)80    config_override('plugin_config.ci_links.link_template', template)81    manager.activate('ci links')82    test = suite.add_test()83    test.when_run.fail()84    summary = suite.run()85    result = summary.get_all_results_for_test(test)[0]86    log_link = result.details.all().get('log_link')87    local_log_path = '/'.join(result.get_log_path().split(os.sep))88    expected_log_link = template % {'build_url': build_url.rstrip('/'),89                                    'log_path': local_log_path}90    assert log_link == expected_log_link91class PluginWithOverriddenBuildURLGetter(Plugin):92    build_url = 'https://overridden/build/url'93    def _get_build_url(self):94        return self.build_url95def test_overridden_build_url_getter(suite):96    plugin = PluginWithOverriddenBuildURLGetter()97    manager.install(plugin, is_internal=False)98    manager.activate('ci links')99    test = suite.add_test()100    test.when_run.fail()101    summary = suite.run()102    result = summary.get_all_results_for_test(test)[0]103    log_link = result.details.all().get('log_link')104    local_log_path = '/'.join(result.get_log_path().split(os.sep))105    expected_log_link = '%s/artifact/%s' % (plugin.build_url, local_log_path)106    assert log_link == expected_log_link107@pytest.fixture(params=BUILD_URLS)108def build_url(request):109    os.environ['BUILD_URL'] = request.param110    @request.addfinalizer111    def fin():  # pylint: disable=unused-variable112        os.environ.pop('BUILD_URL')113    return request.param114@pytest.fixture(params=itertools.product(BUILD_URLS, ENV_VAR_NAMES))115def build_url_env_var(request):116    url, env_var_name = request.param...

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.