KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use pidfile_from_path method in autotest
Best Python code snippet using autotest_python
monitor_db_functional_test.py
Source:monitor_db_functional_test.py 
...92 def finish_process(self, pidfile_type, exit_status=0):93 pidfile_id = self._last_pidfile_id[pidfile_type]94 self._set_pidfile_exit_status(pidfile_id, exit_status)95 def finish_specific_process(self, working_directory, pidfile_name):96 pidfile_id = self.pidfile_from_path(working_directory, pidfile_name)97 self._set_pidfile_exit_status(pidfile_id, 0)98 def finish_active_process_on_host(self, host_id):99 match = 'hosts/host%d/' % host_id100 for pidfile_id in self.nonfinished_pidfile_ids():101 if pidfile_id._working_directory.startswith(match):102 self._set_pidfile_exit_status(pidfile_id, 0)103 break104 else:105 raise KeyError('No active process matched %s' % match)106 def _set_pidfile_exit_status(self, pidfile_id, exit_status):107 assert pidfile_id is not None108 contents = self._pidfiles[pidfile_id]109 contents.exit_status = exit_status110 contents.num_tests_failed = 0111 def was_last_process_killed(self, pidfile_type, sigs):112 pidfile_id = self._last_pidfile_id[pidfile_type]113 return sigs == self._pids_to_signals_received[pidfile_id]114 def nonfinished_pidfile_ids(self):115 return [pidfile_id for pidfile_id, pidfile_contents116 in self._pidfiles.iteritems()117 if pidfile_contents.exit_status is None]118 def running_pidfile_ids(self):119 return [pidfile_id for pidfile_id in self.nonfinished_pidfile_ids()120 if self._pidfiles[pidfile_id].process is not None]121 def pidfile_from_path(self, working_directory, pidfile_name):122 return self._pidfile_index[(working_directory, pidfile_name)]123 def attached_files(self, working_directory):124 """125 Return dict mapping path to contents for attached files with specified126 paths.127 """128 return dict((path, contents) for path, contents129 in self._attached_files.get(working_directory, [])130 if path is not None)131 # DroneManager emulation APIs for use by monitor_db132 def get_orphaned_autoserv_processes(self):133 return set()134 def total_running_processes(self):135 return sum(pidfile_id._num_processes136 for pidfile_id in self.nonfinished_pidfile_ids())137 def max_runnable_processes(self, username, drone_hostnames_allowed):138 return self.process_capacity - self.total_running_processes()139 def refresh(self):140 for pidfile_id in self._unregistered_pidfiles:141 # intentionally handle non-registered pidfiles silently142 self._pidfiles.pop(pidfile_id, None)143 self._unregistered_pidfiles = set()144 def execute_actions(self):145 # executing an "execute_command" causes a pidfile to be created146 for pidfile_id in self._future_pidfiles:147 # Process objects are opaque to monitor_db148 process = object()149 self._pidfiles[pidfile_id].process = process150 self._process_index[process] = pidfile_id151 self._future_pidfiles = []152 for pidfile_id in self._set_pidfile_exit_status_queue:153 self._set_pidfile_exit_status(pidfile_id, 271)154 self._set_pidfile_exit_status_queue = []155 def attach_file_to_execution(self, result_dir, file_contents,156 file_path=None):157 self._attached_files.setdefault(result_dir, set()).add((file_path,158 file_contents))159 return 'attach_path'160 def _initialize_pidfile(self, pidfile_id):161 if pidfile_id not in self._pidfiles:162 assert pidfile_id.key() not in self._pidfile_index163 self._pidfiles[pidfile_id] = drone_manager.PidfileContents()164 self._pidfile_index[pidfile_id.key()] = pidfile_id165 def _set_last_pidfile(self, pidfile_id, working_directory, pidfile_name):166 if working_directory.startswith('hosts/'):167 # such paths look like hosts/host1/1-verify, we'll grab the end168 type_string = working_directory.rsplit('-', 1)[1]169 pidfile_type = _PidfileType.get_value(type_string)170 else:171 pidfile_type = _PIDFILE_TO_PIDFILE_TYPE[pidfile_name]172 self._last_pidfile_id[pidfile_type] = pidfile_id173 def execute_command(self, command, working_directory, pidfile_name,174 num_processes, log_file=None, paired_with_pidfile=None,175 username=None, drone_hostnames_allowed=None):176 logging.debug('Executing %s in %s', command, working_directory)177 pidfile_id = self._DummyPidfileId(working_directory, pidfile_name)178 if pidfile_id.key() in self._pidfile_index:179 pidfile_id = self._pidfile_index[pidfile_id.key()]180 pidfile_id._num_processes = num_processes181 pidfile_id._paired_with_pidfile = paired_with_pidfile182 self._future_pidfiles.append(pidfile_id)183 self._initialize_pidfile(pidfile_id)184 self._pidfile_index[(working_directory, pidfile_name)] = pidfile_id185 self._set_last_pidfile(pidfile_id, working_directory, pidfile_name)186 return pidfile_id187 def get_pidfile_contents(self, pidfile_id, use_second_read=False):188 if pidfile_id not in self._pidfiles:189 logging.debug('Request for nonexistent pidfile %s', pidfile_id)190 return self._pidfiles.get(pidfile_id, drone_manager.PidfileContents())191 def is_process_running(self, process):192 return True193 def register_pidfile(self, pidfile_id):194 self._initialize_pidfile(pidfile_id)195 def unregister_pidfile(self, pidfile_id):196 self._unregistered_pidfiles.add(pidfile_id)197 def declare_process_count(self, pidfile_id, num_processes):198 pidfile_id.num_processes = num_processes199 def absolute_path(self, path):200 return 'absolute/' + path201 def write_lines_to_file(self, file_path, lines, paired_with_process=None):202 # TODO: record this203 pass204 def get_pidfile_id_from(self, execution_tag, pidfile_name):205 default_pidfile = self._DummyPidfileId(execution_tag, pidfile_name,206 num_processes=0)207 return self._pidfile_index.get((execution_tag, pidfile_name),208 default_pidfile)209 def kill_process(self, process, sig=signal.SIGKILL):210 pidfile_id = self._process_index[process]211 if pidfile_id not in self._pids_to_signals_received:212 self._pids_to_signals_received[pidfile_id] = set()213 self._pids_to_signals_received[pidfile_id].add(sig)214 if signal.SIGKILL == sig:215 self._set_pidfile_exit_status_queue.append(pidfile_id)216class MockEmailManager(NullMethodObject):217 _NULL_METHODS = ('send_queued_emails', 'send_email')218 def enqueue_notify_email(self, subject, message):219 logging.warning('enqueue_notify_email: %s', subject)220 logging.warning(message)221class SchedulerFunctionalTest(unittest.TestCase,222 frontend_test_utils.FrontendTestMixin):223 # some number of ticks after which the scheduler is presumed to have224 # stabilized, given no external changes225 _A_LOT_OF_TICKS = 10226 def setUp(self):227 self._frontend_common_setup()228 self._set_stubs()229 self._set_global_config_values()230 self._create_dispatcher()231 logging.basicConfig(level=logging.DEBUG)232 def _create_dispatcher(self):233 self.dispatcher = monitor_db.Dispatcher()234 def tearDown(self):235 self._database.disconnect()236 self._frontend_common_teardown()237 def _set_stubs(self):238 self.mock_config = global_config.FakeGlobalConfig()239 self.god.stub_with(global_config, 'global_config', self.mock_config)240 self.mock_drone_manager = MockDroneManager()241 drone_manager._set_instance(self.mock_drone_manager)242 self.mock_email_manager = MockEmailManager()243 self.god.stub_with(email_manager, 'manager', self.mock_email_manager)244 self._database = (245 database_connection.TranslatingDatabase.get_test_database(246 translators=scheduler_lib._DB_TRANSLATORS))247 self._database.connect(db_type='django')248 self.god.stub_with(monitor_db, '_db', self._database)249 self.god.stub_with(scheduler_models, '_db', self._database)250 MockConnectionManager.db = self._database251 scheduler_lib.ConnectionManager = MockConnectionManager252 monitor_db.initialize_globals()253 scheduler_models.initialize_globals()254 patcher = mock.patch(255 'autotest_lib.scheduler.luciferlib.is_lucifer_enabled',256 lambda: False)257 patcher.start()258 self.addCleanup(patcher.stop)259 def _set_global_config_values(self):260 self.mock_config.set_config_value('SCHEDULER', 'pidfile_timeout_mins',261 1)262 self.mock_config.set_config_value('SCHEDULER', 'gc_stats_interval_mins',263 999999)264 self.mock_config.set_config_value('SCHEDULER', 'enable_archiving', True)265 self.mock_config.set_config_value('SCHEDULER',266 'clean_interval_minutes', 60)267 self.mock_config.set_config_value('SCHEDULER',268 'max_parse_processes', 50)269 self.mock_config.set_config_value('SCHEDULER',270 'max_transfer_processes', 50)271 self.mock_config.set_config_value('SCHEDULER',272 'clean_interval_minutes', 50)273 self.mock_config.set_config_value('SCHEDULER',274 'max_provision_retries', 1)275 self.mock_config.set_config_value('SCHEDULER', 'max_repair_limit', 1)276 self.mock_config.set_config_value(277 'SCHEDULER', 'secs_to_wait_for_atomic_group_hosts', 600)278 self.mock_config.set_config_value(279 'SCHEDULER', 'inline_host_acquisition', True)280 scheduler_config.config.read_config()281 def _initialize_test(self):282 self.dispatcher.initialize()283 def _run_dispatcher(self):284 for _ in xrange(self._A_LOT_OF_TICKS):285 self.dispatcher.tick()286 def test_idle(self):287 self._initialize_test()288 self._run_dispatcher()289 def _assert_process_executed(self, working_directory, pidfile_name):290 process_was_executed = self.mock_drone_manager.was_process_executed(291 'hosts/host1/1-verify', drone_manager.AUTOSERV_PID_FILE)292 self.assert_(process_was_executed,293 '%s/%s not executed' % (working_directory, pidfile_name))294 def _update_instance(self, model_instance):295 return type(model_instance).objects.get(pk=model_instance.pk)296 def _check_statuses(self, queue_entry, queue_entry_status,297 host_status=None):298 self._check_entry_status(queue_entry, queue_entry_status)299 if host_status:300 self._check_host_status(queue_entry.host, host_status)301 def _check_entry_status(self, queue_entry, status):302 # update from DB303 queue_entry = self._update_instance(queue_entry)304 self.assertEquals(queue_entry.status, status)305 def _check_host_status(self, host, status):306 # update from DB307 host = self._update_instance(host)308 self.assertEquals(host.status, status)309 def _run_pre_job_verify(self, queue_entry):310 self._run_dispatcher() # launches verify311 self._check_statuses(queue_entry, HqeStatus.VERIFYING,312 HostStatus.VERIFYING)313 self.mock_drone_manager.finish_process(_PidfileType.VERIFY)314 def test_simple_job(self):315 self._initialize_test()316 job, queue_entry = self._make_job_and_queue_entry()317 self._run_pre_job_verify(queue_entry)318 self._run_dispatcher() # launches job319 self._check_statuses(queue_entry, HqeStatus.RUNNING, HostStatus.RUNNING)320 self._finish_job(queue_entry)321 self._check_statuses(queue_entry, HqeStatus.COMPLETED, HostStatus.READY)322 self._assert_nothing_is_running()323 def _setup_for_pre_job_reset(self):324 self._initialize_test()325 job, queue_entry = self._make_job_and_queue_entry()326 job.reboot_before = model_attributes.RebootBefore.ALWAYS327 job.save()328 return queue_entry329 def _run_pre_job_reset_job(self, queue_entry):330 self._run_dispatcher() # reset331 self._check_statuses(queue_entry, HqeStatus.RESETTING,332 HostStatus.RESETTING)333 self.mock_drone_manager.finish_process(_PidfileType.RESET)334 self._run_dispatcher() # job335 self._finish_job(queue_entry)336 def test_pre_job_reset(self):337 queue_entry = self._setup_for_pre_job_reset()338 self._run_pre_job_reset_job(queue_entry)339 def _run_pre_job_reset_one_failure(self):340 queue_entry = self._setup_for_pre_job_reset()341 self._run_dispatcher() # reset342 self.mock_drone_manager.finish_process(_PidfileType.RESET,343 exit_status=256)344 self._run_dispatcher() # repair345 self._check_statuses(queue_entry, HqeStatus.QUEUED,346 HostStatus.REPAIRING)347 self.mock_drone_manager.finish_process(_PidfileType.REPAIR)348 return queue_entry349 def test_pre_job_reset_failure(self):350 queue_entry = self._run_pre_job_reset_one_failure()351 # from here the job should run as normal352 self._run_pre_job_reset_job(queue_entry)353 def test_pre_job_reset_double_failure(self):354 # TODO (showard): this test isn't perfect. in reality, when the second355 # reset fails, it copies its results over to the job directory using356 # copy_results_on_drone() and then parses them. since we don't handle357 # that, there appear to be no results at the job directory. the358 # scheduler handles this gracefully, parsing gets effectively skipped,359 # and this test passes as is. but we ought to properly test that360 # behavior.361 queue_entry = self._run_pre_job_reset_one_failure()362 self._run_dispatcher() # second reset363 self.mock_drone_manager.finish_process(_PidfileType.RESET,364 exit_status=256)365 self._run_dispatcher()366 self._check_statuses(queue_entry, HqeStatus.FAILED,367 HostStatus.REPAIR_FAILED)368 # nothing else should run369 self._assert_nothing_is_running()370 def _assert_nothing_is_running(self):371 self.assertEquals(self.mock_drone_manager.running_pidfile_ids(), [])372 def _setup_for_post_job_cleanup(self):373 self._initialize_test()374 job, queue_entry = self._make_job_and_queue_entry()375 job.reboot_after = model_attributes.RebootAfter.ALWAYS376 job.save()377 return queue_entry378 def _run_post_job_cleanup_failure_up_to_repair(self, queue_entry,379 include_verify=True):380 if include_verify:381 self._run_pre_job_verify(queue_entry)382 self._run_dispatcher() # job383 self.mock_drone_manager.finish_process(_PidfileType.JOB)384 self._run_dispatcher() # parsing + cleanup385 self.mock_drone_manager.finish_process(_PidfileType.PARSE)386 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP,387 exit_status=256)388 self._run_dispatcher() # repair, HQE unaffected389 return queue_entry390 def test_post_job_cleanup_failure(self):391 queue_entry = self._setup_for_post_job_cleanup()392 self._run_post_job_cleanup_failure_up_to_repair(queue_entry)393 self._check_statuses(queue_entry, HqeStatus.COMPLETED,394 HostStatus.REPAIRING)395 self.mock_drone_manager.finish_process(_PidfileType.REPAIR)396 self._run_dispatcher()397 self._check_statuses(queue_entry, HqeStatus.COMPLETED, HostStatus.READY)398 def test_post_job_cleanup_failure_repair_failure(self):399 queue_entry = self._setup_for_post_job_cleanup()400 self._run_post_job_cleanup_failure_up_to_repair(queue_entry)401 self.mock_drone_manager.finish_process(_PidfileType.REPAIR,402 exit_status=256)403 self._run_dispatcher()404 self._check_statuses(queue_entry, HqeStatus.COMPLETED,405 HostStatus.REPAIR_FAILED)406 def _ensure_post_job_process_is_paired(self, queue_entry, pidfile_type):407 pidfile_name = _PIDFILE_TYPE_TO_PIDFILE[pidfile_type]408 queue_entry = self._update_instance(queue_entry)409 pidfile_id = self.mock_drone_manager.pidfile_from_path(410 queue_entry.execution_path(), pidfile_name)411 self.assert_(pidfile_id._paired_with_pidfile)412 def _finish_job(self, queue_entry):413 self._check_statuses(queue_entry, HqeStatus.RUNNING)414 self.mock_drone_manager.finish_process(_PidfileType.JOB)415 self._run_dispatcher() # launches parsing416 self._check_statuses(queue_entry, HqeStatus.PARSING)417 self._ensure_post_job_process_is_paired(queue_entry, _PidfileType.PARSE)418 self._finish_parsing()419 def _finish_parsing(self):420 self.mock_drone_manager.finish_process(_PidfileType.PARSE)421 self._run_dispatcher()422 def _create_reverify_request(self):423 host = self.hosts[0]...
monitor_db_functional_unittest.py
Source:monitor_db_functional_unittest.py
...104 def finish_process(self, pidfile_type, exit_status=0):105 pidfile_id = self._last_pidfile_id[pidfile_type]106 self._set_pidfile_exit_status(pidfile_id, exit_status)107 def finish_specific_process(self, working_directory, pidfile_name):108 pidfile_id = self.pidfile_from_path(working_directory, pidfile_name)109 self._set_pidfile_exit_status(pidfile_id, 0)110 def _set_pidfile_exit_status(self, pidfile_id, exit_status):111 assert pidfile_id is not None112 contents = self._pidfiles[pidfile_id]113 contents.exit_status = exit_status114 contents.num_tests_failed = 0115 def was_last_process_killed(self, pidfile_type):116 pidfile_id = self._last_pidfile_id[pidfile_type]117 return pidfile_id in self._killed_pidfiles118 def nonfinished_pidfile_ids(self):119 return [pidfile_id for pidfile_id, pidfile_contents120 in self._pidfiles.iteritems()121 if pidfile_contents.exit_status is None]122 def running_pidfile_ids(self):123 return [pidfile_id for pidfile_id in self.nonfinished_pidfile_ids()124 if self._pidfiles[pidfile_id].process is not None]125 def pidfile_from_path(self, working_directory, pidfile_name):126 return self._pidfile_index[(working_directory, pidfile_name)]127 def attached_files(self, working_directory):128 """129 Return dict mapping path to contents for attached files with specified130 paths.131 """132 return dict((path, contents) for path, contents133 in self._attached_files.get(working_directory, [])134 if path is not None)135 # DroneManager emulation APIs for use by monitor_db136 def get_orphaned_autoserv_processes(self):137 return set()138 def total_running_processes(self):139 return sum(pidfile_id._num_processes140 for pidfile_id in self.nonfinished_pidfile_ids())141 def max_runnable_processes(self, username, drone_hostnames_allowed):142 return self.process_capacity - self.total_running_processes()143 def refresh(self):144 for pidfile_id in self._unregistered_pidfiles:145 # intentionally handle non-registered pidfiles silently146 self._pidfiles.pop(pidfile_id, None)147 self._unregistered_pidfiles = set()148 def execute_actions(self):149 # executing an "execute_command" causes a pidfile to be created150 for pidfile_id in self._future_pidfiles:151 # Process objects are opaque to monitor_db152 process = object()153 self._pidfiles[pidfile_id].process = process154 self._process_index[process] = pidfile_id155 self._future_pidfiles = []156 def attach_file_to_execution(self, result_dir, file_contents,157 file_path=None):158 self._attached_files.setdefault(result_dir, set()).add((file_path,159 file_contents))160 return 'attach_path'161 def _initialize_pidfile(self, pidfile_id):162 if pidfile_id not in self._pidfiles:163 assert pidfile_id.key() not in self._pidfile_index164 self._pidfiles[pidfile_id] = drone_manager.PidfileContents()165 self._pidfile_index[pidfile_id.key()] = pidfile_id166 def _set_last_pidfile(self, pidfile_id, working_directory, pidfile_name):167 if working_directory.startswith('hosts/'):168 # such paths look like hosts/host1/1-verify, we'll grab the end169 type_string = working_directory.rsplit('-', 1)[1]170 pidfile_type = _PidfileType.get_value(type_string)171 else:172 pidfile_type = _PIDFILE_TO_PIDFILE_TYPE[pidfile_name]173 self._last_pidfile_id[pidfile_type] = pidfile_id174 def execute_command(self, command, working_directory, pidfile_name,175 num_processes, log_file=None, paired_with_pidfile=None,176 username=None, drone_hostnames_allowed=None):177 logging.debug('Executing %s in %s', command, working_directory)178 pidfile_id = self._DummyPidfileId(working_directory, pidfile_name)179 if pidfile_id.key() in self._pidfile_index:180 pidfile_id = self._pidfile_index[pidfile_id.key()]181 pidfile_id._num_processes = num_processes182 pidfile_id._paired_with_pidfile = paired_with_pidfile183 self._future_pidfiles.append(pidfile_id)184 self._initialize_pidfile(pidfile_id)185 self._pidfile_index[(working_directory, pidfile_name)] = pidfile_id186 self._set_last_pidfile(pidfile_id, working_directory, pidfile_name)187 return pidfile_id188 def get_pidfile_contents(self, pidfile_id, use_second_read=False):189 if pidfile_id not in self._pidfiles:190 logging.debug('Request for nonexistent pidfile %s' % pidfile_id)191 return self._pidfiles.get(pidfile_id, drone_manager.PidfileContents())192 def is_process_running(self, process):193 return True194 def register_pidfile(self, pidfile_id):195 self._initialize_pidfile(pidfile_id)196 def unregister_pidfile(self, pidfile_id):197 self._unregistered_pidfiles.add(pidfile_id)198 def declare_process_count(self, pidfile_id, num_processes):199 pidfile_id.num_processes = num_processes200 def absolute_path(self, path):201 return 'absolute/' + path202 def write_lines_to_file(self, file_path, lines, paired_with_process=None):203 # TODO: record this204 pass205 def get_pidfile_id_from(self, execution_tag, pidfile_name):206 default_pidfile = self._DummyPidfileId(execution_tag, pidfile_name,207 num_processes=0)208 return self._pidfile_index.get((execution_tag, pidfile_name),209 default_pidfile)210 def kill_process(self, process):211 pidfile_id = self._process_index[process]212 self._killed_pidfiles.add(pidfile_id)213 self._set_pidfile_exit_status(pidfile_id, 271)214class MockEmailManager(NullMethodObject):215 _NULL_METHODS = ('send_queued_admin', 'send')216 def enqueue_admin(self, subject, message):217 logging.warn('enqueue_notify_email: %s', subject)218 logging.warn(message)219class SchedulerFunctionalTest(unittest.TestCase,220 test_utils.FrontendTestMixin):221 # some number of ticks after which the scheduler is presumed to have222 # stabilized, given no external changes223 _A_LOT_OF_TICKS = 10224 def setUp(self):225 self._frontend_common_setup()226 self._set_stubs()227 self._set_settings_values()228 self._create_dispatcher()229 logging.basicConfig(level=logging.DEBUG)230 def _create_dispatcher(self):231 self.dispatcher = monitor_db.Dispatcher()232 def tearDown(self):233 self._database.disconnect()234 self._frontend_common_teardown()235 def _set_stubs(self):236 self.mock_config = MockGlobalConfig()237 self.god.stub_with(settings, 'settings', self.mock_config)238 self.mock_drone_manager = MockDroneManager()239 drone_manager._set_instance(self.mock_drone_manager)240 self.mock_email_manager = MockEmailManager()241 self.god.stub_with(mail, "manager", self.mock_email_manager)242 self._database = (243 database_connection.TranslatingDatabase.get_test_database(244 translators=_DB_TRANSLATORS))245 self._database.connect(db_type='django')246 self.god.stub_with(monitor_db, '_db', self._database)247 self.god.stub_with(scheduler_models, '_db', self._database)248 monitor_db.initialize_globals()249 scheduler_models.initialize_globals()250 def _set_settings_values(self):251 self.mock_config.set_config_value('SCHEDULER', 'pidfile_timeout_mins',252 1)253 self.mock_config.set_config_value('SCHEDULER', 'gc_stats_interval_mins',254 999999)255 def _initialize_test(self):256 self.dispatcher.initialize()257 def _run_dispatcher(self):258 for _ in xrange(self._A_LOT_OF_TICKS):259 self.dispatcher.tick()260 def test_idle(self):261 self._initialize_test()262 self._run_dispatcher()263 def _assert_process_executed(self, working_directory, pidfile_name):264 process_was_executed = self.mock_drone_manager.was_process_executed(265 'hosts/host1/1-verify', drone_manager.AUTOSERV_PID_FILE)266 self.assert_(process_was_executed,267 '%s/%s not executed' % (working_directory, pidfile_name))268 def _update_instance(self, model_instance):269 return type(model_instance).objects.get(pk=model_instance.pk)270 def _check_statuses(self, queue_entry, queue_entry_status,271 host_status=None):272 self._check_entry_status(queue_entry, queue_entry_status)273 if host_status:274 self._check_host_status(queue_entry.host, host_status)275 def _check_entry_status(self, queue_entry, status):276 # update from DB277 queue_entry = self._update_instance(queue_entry)278 self.assertEquals(queue_entry.status, status)279 def _check_host_status(self, host, status):280 # update from DB281 host = self._update_instance(host)282 self.assertEquals(host.status, status)283 def _run_pre_job_verify(self, queue_entry):284 self._run_dispatcher() # launches verify285 self._check_statuses(queue_entry, HqeStatus.VERIFYING,286 HostStatus.VERIFYING)287 self.mock_drone_manager.finish_process(_PidfileType.VERIFY)288 def test_simple_job(self):289 self._initialize_test()290 job, queue_entry = self._make_job_and_queue_entry()291 self._run_pre_job_verify(queue_entry)292 self._run_dispatcher() # launches job293 self._check_statuses(queue_entry, HqeStatus.RUNNING, HostStatus.RUNNING)294 self._finish_job(queue_entry)295 self._check_statuses(queue_entry, HqeStatus.COMPLETED, HostStatus.READY)296 self._assert_nothing_is_running()297 def _setup_for_pre_job_cleanup(self):298 self._initialize_test()299 job, queue_entry = self._make_job_and_queue_entry()300 job.reboot_before = model_attributes.RebootBefore.ALWAYS301 job.save()302 return queue_entry303 def _run_pre_job_cleanup_job(self, queue_entry):304 self._run_dispatcher() # cleanup305 self._check_statuses(queue_entry, HqeStatus.VERIFYING,306 HostStatus.CLEANING)307 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP)308 self._run_dispatcher() # verify309 self.mock_drone_manager.finish_process(_PidfileType.VERIFY)310 self._run_dispatcher() # job311 self._finish_job(queue_entry)312 def test_pre_job_cleanup(self):313 queue_entry = self._setup_for_pre_job_cleanup()314 self._run_pre_job_cleanup_job(queue_entry)315 def _run_pre_job_cleanup_one_failure(self):316 queue_entry = self._setup_for_pre_job_cleanup()317 self._run_dispatcher() # cleanup318 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP,319 exit_status=256)320 self._run_dispatcher() # repair321 self._check_statuses(queue_entry, HqeStatus.QUEUED,322 HostStatus.REPAIRING)323 self.mock_drone_manager.finish_process(_PidfileType.REPAIR)324 return queue_entry325 def test_pre_job_cleanup_failure(self):326 queue_entry = self._run_pre_job_cleanup_one_failure()327 # from here the job should run as normal328 self._run_pre_job_cleanup_job(queue_entry)329 def test_pre_job_cleanup_double_failure(self):330 # TODO (showard): this test isn't perfect. in reality, when the second331 # cleanup fails, it copies its results over to the job directory using332 # copy_results_on_drone() and then parses them. since we don't handle333 # that, there appear to be no results at the job directory. the334 # scheduler handles this gracefully, parsing gets effectively skipped,335 # and this test passes as is. but we ought to properly test that336 # behavior.337 queue_entry = self._run_pre_job_cleanup_one_failure()338 self._run_dispatcher() # second cleanup339 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP,340 exit_status=256)341 self._run_dispatcher()342 self._check_statuses(queue_entry, HqeStatus.FAILED,343 HostStatus.REPAIR_FAILED)344 # nothing else should run345 self._assert_nothing_is_running()346 def _assert_nothing_is_running(self):347 self.assertEquals(self.mock_drone_manager.running_pidfile_ids(), [])348 def _setup_for_post_job_cleanup(self):349 self._initialize_test()350 job, queue_entry = self._make_job_and_queue_entry()351 job.reboot_after = model_attributes.RebootAfter.ALWAYS352 job.save()353 return queue_entry354 def _run_post_job_cleanup_failure_up_to_repair(self, queue_entry,355 include_verify=True):356 if include_verify:357 self._run_pre_job_verify(queue_entry)358 self._run_dispatcher() # job359 self.mock_drone_manager.finish_process(_PidfileType.JOB)360 self._run_dispatcher() # parsing + cleanup361 self.mock_drone_manager.finish_process(_PidfileType.PARSE)362 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP,363 exit_status=256)364 self._run_dispatcher() # repair, HQE unaffected365 self.mock_drone_manager.finish_process(_PidfileType.ARCHIVE)366 self._run_dispatcher()367 return queue_entry368 def test_post_job_cleanup_failure(self):369 queue_entry = self._setup_for_post_job_cleanup()370 self._run_post_job_cleanup_failure_up_to_repair(queue_entry)371 self._check_statuses(queue_entry, HqeStatus.COMPLETED,372 HostStatus.REPAIRING)373 self.mock_drone_manager.finish_process(_PidfileType.REPAIR)374 self._run_dispatcher()375 self._check_statuses(queue_entry, HqeStatus.COMPLETED, HostStatus.READY)376 def test_post_job_cleanup_failure_repair_failure(self):377 queue_entry = self._setup_for_post_job_cleanup()378 self._run_post_job_cleanup_failure_up_to_repair(queue_entry)379 self.mock_drone_manager.finish_process(_PidfileType.REPAIR,380 exit_status=256)381 self._run_dispatcher()382 self._check_statuses(queue_entry, HqeStatus.COMPLETED,383 HostStatus.REPAIR_FAILED)384 def _ensure_post_job_process_is_paired(self, queue_entry, pidfile_type):385 pidfile_name = _PIDFILE_TYPE_TO_PIDFILE[pidfile_type]386 queue_entry = self._update_instance(queue_entry)387 pidfile_id = self.mock_drone_manager.pidfile_from_path(388 queue_entry.execution_path(), pidfile_name)389 self.assert_(pidfile_id._paired_with_pidfile)390 def _finish_job(self, queue_entry):391 self.mock_drone_manager.finish_process(_PidfileType.JOB)392 self._run_dispatcher() # launches parsing + cleanup393 self._check_statuses(queue_entry, HqeStatus.PARSING,394 HostStatus.CLEANING)395 self._ensure_post_job_process_is_paired(queue_entry, _PidfileType.PARSE)396 self._finish_parsing_and_cleanup(queue_entry)397 def _finish_parsing_and_cleanup(self, queue_entry):398 self.mock_drone_manager.finish_process(_PidfileType.CLEANUP)399 self.mock_drone_manager.finish_process(_PidfileType.PARSE)400 self._run_dispatcher()401 self._check_entry_status(queue_entry, HqeStatus.ARCHIVING)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest 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!!
Was this article helpful?
