How to use _get_consistent_execution_path method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...45 """46 hqe = models.HostQueueEntry.objects.get(id=hqe_id)47 hqes = hqe.job.hostqueueentry_set.all()48 try:49 _get_consistent_execution_path(hqes)50 except _ExecutionPathError:51 return True52 return False53# TODO( This is temporary to enable toggling54# lucifer rollouts with an option.55def spawn_starting_job_handler(manager, job):56 """Spawn job_reporter to handle a job.57 Pass all arguments by keyword.58 @param manager: scheduler.drone_manager.DroneManager instance59 @param job: Job instance60 @returns: Drone instance61 """62 raise NotImplementedError63# TODO( This is temporary to enable toggling64# lucifer rollouts with an option.65def spawn_gathering_job_handler(manager, job, autoserv_exit, pidfile_id=None):66 """Spawn job_reporter to handle a job.67 Pass all arguments by keyword.68 @param manager: scheduler.drone_manager.DroneManager instance69 @param job: Job instance70 @param autoserv_exit: autoserv exit status71 @param pidfile_id: PidfileId instance72 @returns: Drone instance73 """74 manager = _DroneManager(manager)75 if pidfile_id is None:76 drone = manager.pick_drone_to_use()77 else:78 drone = manager.get_drone_for_pidfile(pidfile_id)79 results_dir = _results_dir(manager, job)80 num_tests_failed = manager.get_num_tests_failed(pidfile_id)81 args = [82 _JOB_REPORTER_PATH,83 # General configuration84 '--jobdir', _get_jobdir(),85 '--run-job-path', _get_run_job_path(),86 '--watcher-path', _get_watcher_path(),87 # Job specific88 '--job-id', str(,89 '--lucifer-level', 'GATHERING',90 '--autoserv-exit', str(autoserv_exit),91 '--need-gather',92 '--num-tests-failed', str(num_tests_failed),93 '--results-dir', results_dir,94 ]95 if _get_gcp_creds():96 args = [97 'GOOGLE_APPLICATION_CREDENTIALS=%s'98 % pipes.quote(_get_gcp_creds()),99 ] + args100 output_file = os.path.join(results_dir, 'job_reporter_output.log')101 drone.spawn(_ENV, args, output_file=output_file)102 return drone103# TODO( This is temporary to enable toggling104# lucifer rollouts with an option.105def spawn_parsing_job_handler(manager, job, autoserv_exit, pidfile_id=None):106 """Spawn job_reporter to handle a job.107 Pass all arguments by keyword.108 @param manager: scheduler.drone_manager.DroneManager instance109 @param job: Job instance110 @param autoserv_exit: autoserv exit status111 @param pidfile_id: PidfileId instance112 @returns: Drone instance113 """114 manager = _DroneManager(manager)115 if pidfile_id is None:116 drone = manager.pick_drone_to_use()117 else:118 drone = manager.get_drone_for_pidfile(pidfile_id)119 results_dir = _results_dir(manager, job)120 args = [121 _JOB_REPORTER_PATH,122 # General configuration123 '--jobdir', _get_jobdir(),124 '--run-job-path', _get_run_job_path(),125 '--watcher-path', _get_watcher_path(),126 # Job specific127 '--job-id', str(,128 '--lucifer-level', 'GATHERING',129 '--autoserv-exit', str(autoserv_exit),130 '--results-dir', results_dir,131 ]132 if _get_gcp_creds():133 args = [134 'GOOGLE_APPLICATION_CREDENTIALS=%s'135 % pipes.quote(_get_gcp_creds()),136 ] + args137 output_file = os.path.join(results_dir, 'job_reporter_output.log')138 drone.spawn(_ENV, args, output_file=output_file)139 return drone140def _get_jobdir():141 return _config.get_config_value(_SECTION, 'jobdir')142def _get_run_job_path():143 return os.path.join(_get_binaries_path(), 'lucifer_run_job')144def _get_watcher_path():145 return os.path.join(_get_binaries_path(), 'lucifer_watcher')146def _get_binaries_path():147 """Get binaries dir path from config.."""148 return _config.get_config_value(_SECTION, 'binaries_path')149def _get_gcp_creds():150 """Return path to GCP service account credentials.151 This is the empty string by default, if no credentials will be used.152 """153 return _config.get_config_value(_SECTION, 'gcp_creds', default='')154class _DroneManager(object):155 """Simplified drone API."""156 def __init__(self, old_manager):157 """Initialize instance.158 @param old_manager: old style DroneManager159 """160 self._manager = old_manager161 def get_num_tests_failed(self, pidfile_id):162 """Return the number of tests failed for autoserv by pidfile.163 @param pidfile_id: PidfileId instance.164 @returns: int (-1 if missing)165 """166 state = self._manager.get_pidfile_contents(pidfile_id)167 if state.num_tests_failed is None:168 return -1169 return state.num_tests_failed170 def get_drone_for_pidfile(self, pidfile_id):171 """Return a drone to use from a pidfile.172 @param pidfile_id: PidfileId instance.173 """174 return _wrap_drone(self._manager.get_drone_for_pidfile_id(pidfile_id))175 def pick_drone_to_use(self, num_processes=1, prefer_ssp=False):176 """Return a drone to use.177 Various options can be passed to optimize drone selection.178 @param num_processes: number of processes the drone is intended179 to run180 @param prefer_ssp: indicates whether drones supporting181 server-side packaging should be preferred. The returned182 drone is not guaranteed to support it.183 """184 old_drone = self._manager.pick_drone_to_use(185 num_processes=num_processes,186 prefer_ssp=prefer_ssp,187 )188 return _wrap_drone(old_drone)189 def absolute_path(self, path):190 """Return absolute path for drone results.191 The returned path might be remote.192 """193 return self._manager.absolute_path(path)194def _wrap_drone(old_drone):195 """Wrap an old style drone."""196 host = old_drone._host197 if isinstance(host, local_host.LocalHost):198 return LocalDrone()199 elif isinstance(host, ssh_host.SSHHost):200 return RemoteDrone(host)201 else:202 raise TypeError('Drone has an unknown host type')203def _results_dir(manager, job):204 """Return results dir for a job.205 Path may be on a remote host.206 """207 return manager.absolute_path(_working_directory(job))208def _working_directory(job):209 return _get_consistent_execution_path(job.hostqueueentry_set.all())210def _get_consistent_execution_path(execution_entries):211 first_execution_path = execution_entries[0].execution_path()212 for execution_entry in execution_entries[1:]:213 if execution_entry.execution_path() != first_execution_path:214 raise _ExecutionPathError(215 '%s (%s) != %s (%s)'216 % (execution_entry.execution_path(),217 execution_entry,218 first_execution_path,219 execution_entries[0]))220 return first_execution_path221class _ExecutionPathError(Exception):222 """Raised by _get_consistent_execution_path()."""223class Drone(object):224 """Simplified drone API."""225 def hostname(self):226 """Return the hostname of the drone."""227 def spawn(self, path, args, output_file):228 """Spawn an independent process.229 path must be an absolute path. path may be on a remote machine.230 args is a list of arguments.231 The process is spawned in its own session. It should not try to232 obtain a controlling terminal.233 The new process will have stdin opened to /dev/null and stdout,234 stderr opened to output_file.235 output_file is a pathname, but how it is interpreted is236 implementation defined, e.g., it may be a remote file....

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:


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

Next-Gen App & Browser Testing Cloud

Was this article helpful?