KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use runnable_test_dir method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py

1import slash2import os3import signal4from .utils.suite_writer import Suite5from slash.resuming import get_tests_to_resume6from slash.exceptions import InteractiveParallelNotAllowed, ParallelTimeout7from slash.parallel.server import ServerStates8from slash.parallel.parallel_manager import ParallelManager9from slash import Session10from slash.loader import Loader11import time12import tempfile13import pytest14@pytest.fixture(scope='module', autouse=True)15def no_parallel_user_config(request):16    tmpdir = tempfile.mkdtemp()17    path = os.path.join(tmpdir, 'slashrc')18    os.environ["SLASH_USER_SETTINGS"] = path19    @request.addfinalizer20    def cleanup():  # pylint: disable=unused-variable21        os.rmdir(tmpdir)22        del os.environ["SLASH_USER_SETTINGS"]23#basic features of parallel24def run_specific_workers_and_tests_num(workers_num, tests_num=10):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) == 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()...

conftest.py

Source:conftest.py

...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.fixture143def test_loader():144    return Loader()145@pytest.fixture146def active_slash_session(request):147    returned = slash.Session()...

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.