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