Best Python code snippet using autotest_python
profilers.py
Source:profilers.py
...99 os.makedirs(local_dir)100 else:101 local_dir = test.profdir102 return local_dir103 def _get_failure_logs(self, autodir, test, host):104 """105 Collect the client logs from a profiler run and put them in a106 file named failure-*.log.107 """108 try:109 fd, path = tempfile.mkstemp(suffix='.log', prefix='failure-',110 dir=self._get_local_profilers_dir(test, host.hostname))111 os.close(fd)112 host.get_file(get_profiler_log_path(autodir), path)113 # try to collect any partial profiler logs114 self._get_profiler_logs(autodir, test, host)115 except (error.AutotestError, error.AutoservError):116 logging.exception('Profiler failure log collection failed')117 # swallow the exception so that we don't override an existing118 # exception being thrown119 def _get_all_failure_logs(self, test, hosts):120 for host, at, autodir in hosts:121 self._get_failure_logs(autodir, test, host)122 def _get_profiler_logs(self, autodir, test, host):123 results_dir = get_profiler_results_dir(autodir)124 local_dir = self._get_local_profilers_dir(test, host.hostname)125 self.job.remove_client_log(host.hostname, results_dir, local_dir)126 tempdir = tempfile.mkdtemp(dir=self.job.tmpdir)127 try:128 host.get_file(results_dir + '/', tempdir)129 except error.AutoservRunError:130 pass # no files to pull back, nothing we can do131 utils.merge_trees(tempdir, local_dir)132 shutil.rmtree(tempdir, ignore_errors=True)133 def _run_clients(self, test, hosts):134 """135 We initialize the profilers just before start because only then we136 know all the hosts involved.137 """138 hostnames = [host_info[0].hostname for host_info in hosts]139 profilers_args = [(p.name, p.args, p.dargs)140 for p in self.list]141 for host, at, autodir in hosts:142 control_script = standalone_profiler.generate_test(hostnames,143 host.hostname,144 profilers_args,145 180, None)146 try:147 at.run(control_script, background=True)148 except Exception:149 self._get_failure_logs(autodir, test, host)150 raise151 remote_results_dir = get_profiler_results_dir(autodir)152 local_results_dir = self._get_local_profilers_dir(test,153 host.hostname)154 self.job.add_client_log(host.hostname, remote_results_dir,155 local_results_dir)156 try:157 # wait for the profilers to be added158 standalone_profiler.wait_for_profilers(hostnames)159 except Exception:160 self._get_all_failure_logs(test, hosts)161 raise162 def before_start(self, test, host=None):163 # create host objects and install the needed clients...
profiler.py
Source:profiler.py
...116 elif host in self.installed_hosts:117 return {host: self.installed_hosts[host]}118 else:119 return {}120 def _get_failure_logs(self, autodir, test, host):121 """Collect the client logs from a profiler run and put them in a122 file named failure-*.log."""123 try:124 profdir = os.path.join(test.profdir, host.hostname)125 if not os.path.exists(profdir):126 os.makedirs(profdir)127 fd, path = tempfile.mkstemp(suffix=".log", prefix="failure-",128 dir=os.path.join(test.profdir,129 host.hostname))130 os.close(fd)131 host.get_file(get_profiler_log_path(autodir), path)132 except (error.AutotestError, error.AutoservError):133 logging.exception("Profiler failure log collection failed")134 # swallow the exception so that we don't override an existing135 # exception being thrown136 def start(self, test, host=None):137 self._install()138 encoded_args = encode_args(self.name, self.args, self.dargs)139 control_script = run_profiler_control % (encoded_args, self.name)140 for host, (at, autodir) in self._get_hosts(host).iteritems():141 fifo_pattern = os.path.join(autodir, "profiler.*")142 host.run("rm -f %s" % fifo_pattern)143 host.run("mkfifo %s" % os.path.join(autodir, "profiler.ready"))144 try:145 at.run(control_script, background=True)146 self._wait_on_client(host, autodir, "ready")147 self._signal_client(host, autodir, "start")148 remote_results_dir = get_profiler_results_dir(autodir)149 local_results_dir = os.path.join(test.profdir, host.hostname)150 self.job.add_client_log(host.hostname, remote_results_dir,151 local_results_dir)152 except:153 self._get_failure_logs(autodir, test, host)154 raise155 self.current_test = test156 def stop(self, test, host=None):157 assert self.current_test == test158 for host, (at, autodir) in self._get_hosts(host).iteritems():159 try:160 self._signal_client(host, autodir, "stop")161 except:162 self._get_failure_logs(autodir, test, host)163 raise164 def report(self, test, host=None, wait_on_client=True):165 assert self.current_test == test166 self.current_test = None167 # signal to all the clients that they should report168 if wait_on_client:169 for host, (at, autodir) in self._get_hosts(host).iteritems():170 try:171 self._signal_client(host, autodir, "report")172 except:173 self._get_failure_logs(autodir, test, host)174 raise175 # pull back all the results176 for host, (at, autodir) in self._get_hosts(host).iteritems():177 if wait_on_client:178 self._wait_on_client(host, autodir, "finished")179 results_dir = get_profiler_results_dir(autodir)180 local_dir = os.path.join(test.profdir, host.hostname)181 if not os.path.exists(local_dir):182 os.makedirs(local_dir)183 self.job.remove_client_log(host.hostname, results_dir, local_dir)184 tempdir = tempfile.mkdtemp(dir=self.job.tmpdir)185 try:186 host.get_file(results_dir + "/", tempdir)187 except error.AutoservRunError:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!