How to use runnable_test_dir method in Slash

Best Python code snippet using slash

test_parallel.py

Source:test_parallel.py Github

copy

Full Screen

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

Full Screen

Full Screen

conftest.py

Source:conftest.py Github

copy

Full Screen

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

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