How to use get_beaker_code method in autotest

Best Python code snippet using autotest_python

harness_beaker.py

Source:harness_beaker.py

...370                if 'BEAKER_TASK_ID' not in os.environ:371                    raise error.HarnessError("No BEAKER_TASK_ID set")372                task_id = os.environ['BEAKER_TASK_ID']373                task_upload = True374            bkr_status = get_beaker_code(code)375            try:376                resultid = self.bkr_proxy.task_result(task_id, bkr_status,377                                                      subdir, 1, '')378                if task_upload:379                    self.upload_result_files(task_id, resultid, subdir)380            except Exception:381                logging.critical('ERROR: Failed to process test results')382    def tear_down(self):383        '''called from complete and abort.  clean up and shutdown'''384        self.kill_watchdog()385        if self.recipe_id != '0':386            self.upload_recipe_files()387            self.bkr_proxy.recipe_stop()388        os.remove(self.state_file)389    def start_watchdog(self, heartbeat):390        logging.debug('harness: Starting watchdog process, heartbeat: %d' % heartbeat)391        try:392            pid = os.fork()393            if pid == 0:394                self.watchdog_loop(heartbeat)395            else:396                self.watchdog_pid = pid397                logging.debug('harness: Watchdog process started, pid: %d', self.watchdog_pid)398        except OSError as e:399            logging.error('harness: fork in start_watchdog failed: %d (%s)\n' % (e.errno, e.strerror))400    def kill_watchdog(self):401        logging.debug('harness: Killing watchdog, pid: %d', self.watchdog_pid)402        utils.nuke_pid(self.watchdog_pid)403        self.watchdog_pid = None404    def watchdog_loop(self, heartbeat):405        while True:406            time.sleep(heartbeat)407            logging.info('[-- MARK -- %s]' % time.asctime(time.localtime(time.time())))408        sys.exit()409    def get_processed_tests(self):410        tests = {}411        if not os.path.isfile(self.state_file):412            return tests413        f = open(self.state_file, 'r')414        lines = f.readlines()415        f.close()416        for line in lines:417            subdir, t_id = line.strip().split()418            # duplicates result from multiple writers419            # once during the conversion and then again420            # during an update of a test run421            # former has task ids, latter will not422            if subdir not in tests:423                tests[subdir] = t_id424        return tests425    def write_processed_tests(self, subdir, t_id='0'):426        f = open(self.state_file, 'a')427        f.write(subdir + ' ' + t_id + '\n')428        f.close()429    def upload_recipe_files(self):430        path = self.job.resultdir431        # refresh latest executed tests432        tests = self.get_processed_tests()433        logging.debug("Recipe filtering following tests: %s" % tests)434        for root, dirnames, files in os.walk(path):435            '''do not upload previously uploaded results files'''436            for d in dirnames:437                if d in tests:438                    dirnames.remove(d)439            for name in files:440                # strip full path441                remotepath = re.sub(path, "", root)442                # The localfile has the full path443                localfile = os.path.join(root, name)444                if os.path.getsize(localfile) == 0:445                    continue  # skip empty files446                # Upload the file447                self.bkr_proxy.recipe_upload_file(localfile, remotepath)448    def upload_task_files(self, task_id, subdir):449        path = os.path.join(self.job.resultdir, subdir)450        for root, _, files in os.walk(path):451            for name in files:452                # strip full path453                remotepath = re.sub(path, "", root)454                # The localfile has the full path455                localfile = os.path.join(root, name)456                if os.path.getsize(localfile) == 0:457                    continue  # skip empty files458                # Upload the file459                self.bkr_proxy.task_upload_file(task_id, localfile,460                                                remotepath)461    def upload_result_files(self, task_id, resultid, subdir):462        path = os.path.join(self.job.resultdir, subdir)463        for root, _, files in os.walk(path):464            for name in files:465                # strip full path466                remotepath = re.sub(path, "", root)467                # The localfile has the full path468                localfile = os.path.join(root, name)469                if os.path.getsize(localfile) == 0:470                    continue  # skip empty files471                # Upload the file472                self.bkr_proxy.result_upload_file(task_id, resultid, localfile,473                                                  remotepath)474def get_beaker_code(at_code):475    bkr_status = 'Warn'476    if at_code == 'GOOD':477        bkr_status = 'Pass'478    if at_code in ['WARN', 'FAIL', 'ERROR', 'ABORT', 'TEST_NA']:479        bkr_status = 'Fail'480    return bkr_status481if __name__ == '__main__':...

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.