KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use make_path_result method in autotest

Best Python code snippet using autotest_python

bkr_proxy.py

Source:bkr_proxy.py

...84    """85    rpath = '/recipes/' + r86    tpath = t and '/tasks/' + t or ''87    return rpath + tpath + '/status'88def make_path_result(r, t):89    """90    Converts task id into a beaker path to result file91    Given a recipe id and a task id, translate them into92    the proper beaker path to the result file.93    :param r: recipe id94    :param t: task id95    :return: a beaker path of the task's result file96    """97    rpath = '/recipes/' + r98    tpath = '/tasks/' + t99    return rpath + tpath + '/results/'100def make_path_log(r, t=None, i=None):101    """102    Converts id into a beaker path to log file103    Given a recipe id, a task id, and/or result id, translate104    them into the proper beaker path to the log file.  Depending105    on which log file is needed, provide the appropriate params.106    Note the dependency, a result id needs a task id and recipe id,107    while a task id needs a recipe id.108    :param r: recipe id109    :param t: task id110    :param i: result id111    :return: a beaker path of the task's result file112    """113    rpath = '/recipes/' + r114    tpath = t and '/tasks/' + t or ''115    ipath = i and '/results/' + i or ''116    return rpath + tpath + ipath + '/logs'117'''End Hard coded paths'''118def copy_remote(data, dest, use_put=None):119    """120    Copy data to a remote server using http calls POST or PUT121    Using http POST and PUT methods, copy data over http.  To use122    PUT method, provide a dictionary of values to be populated in123    the Content-Range and Content-Length headers.  Otherwise default124    is to use POST method.125    Traps on HTTPError 500 and 400126    :param data: encoded data string to copy remotely127    :param dest: remote server URL128    :param use_put: dictionary of items if using PUT method129    :return: html header info for post processing130    """131    ret = None132    req = urllib2.Request(dest, data=data)133    if use_put:134        req.add_header('Content-Type', 'application/octet-stream')135        end = use_put['start'] + use_put['size'] - 1136        req.add_header('Content-Range', 'bytes %s-%s/%s' % (use_put['start'],137                                                            end, use_put['total']))138        req.add_header('Content-Length', '%s' % use_put['size'])139        req.get_method = lambda: 'PUT'140    try:141        res = utils.urlopen(req)142        ret = res.info()143        res.close()144    except urllib2.HTTPError, e:145        if e.code == 500:146            # the server aborted this recipe DIE DIE DIE147            raise BkrProxyException("We have been aborted!!!")148        elif e.code == 400 and use_put:149            log.error("Error(%s) failed to upload file %s" % (e.code, dest))150    return ret151def copy_local(data, dest, use_put=None):152    """153    Copy data locally to a file154    To aid in debugging, copy a file locally to verify the contents.155    Attempts to write the same data that would otherwise be sent156    remotely.157    :param data: encoded data string to copy locally158    :param dest: local file path159    :param use_put: chooses to write in binary or text160    :return: nothing161    """162    dpath = os.path.dirname(dest)163    if not os.path.isdir(dpath):164        os.makedirs(dpath)165    if use_put:166        open(dest, 'ab').write(data)167    else:168        open(dest, 'a').write("%s %s\n" % (time.time(), data))169def copy_data(data, dest, header=None, use_put=None):170    """171    Copy data to a destination172    To aid in debugging, copy a file locally to verify the contents.173    Attempts to write the same data that would otherwise be sent174    remotely.175    :param data: data string to copy176    :param dest: destination path177    :param header: header info item to return178    :param use_put: dictionary of items for PUT method179    :return: nothing or header info if requested180    """181    ret = None182    # PUT uses a filename instead of a list like POST183    if use_put:184        udata = data185    else:186        udata = urllib.urlencode(data)187    if utils.is_url(dest):188        ret = copy_remote(udata, dest, use_put)189        if header:190            return ret[header]191    else:192        if header:193            ret = dest + str(time.time())  # should be unique194            dest = ret + "/_task_result"195        copy_local(udata, dest, use_put)196    return ret197class BkrProxy(object):198    def __init__(self, recipe_id, labc_url=None):199        # labc_url determines local or remote functionality200        self.labc_url = labc_url or AUTOTEST_CACHE_DIR201        self.recipe_id = recipe_id202        if not labc_url:203            path = self.labc_url + make_path_recipe(self.recipe_id)204            log.info('Writing offline files to %s' % path)205        path = make_path_cmdlog(self.recipe_id)206        self.cmd_log = open(path, 'a', 0)207    def _upload_file(self, lf, rp, r, t=None, i=None):208        if not os.path.isfile(lf):209            raise BkrProxyException("Bad file - %s" % lf)210        lfile = os.path.basename(lf)211        path = self.labc_url + make_path_log(r, t, i) + rp + '/' + lfile212        # copy in chunks213        chunksize = 262144214        start = 0215        total = os.path.getsize(lf)216        use_put = {'total': total}217        f = open(lf, 'r')218        def readchunk():219            return f.read(chunksize)220        for d in iter(readchunk, ''):221            use_put['start'] = start222            use_put['size'] = len(d)223            copy_data(d, path, use_put=use_put)224            start += len(d)225        if total == 0:226            use_put['start'] = 0227            use_put['size'] = 0228            copy_data('', path, use_put=use_put)229        f.close()230    def recipe_upload_file(self, localfile, remotepath=''):231        self.cmd_log.write('recipe_upload_file: localfile=%s, remotepath=%s\n' %232                           (localfile, remotepath))233        self._upload_file(localfile, remotepath, self.recipe_id)234    def task_upload_file(self, task_id, localfile, remotepath=''):235        self.cmd_log.write('task_upload_file: task_id(%s) localfile=%s, remotepath=%s\n' %236                           (task_id, localfile, remotepath))237        self._upload_file(localfile, remotepath, self.recipe_id, task_id)238    def result_upload_file(self, task_id, result_id, localfile, remotepath=''):239        self.cmd_log.write('result_upload_file: task_id(%s), result_id(%s)'240                           ' localfile=%s, remotepath=%s\n' %241                           (task_id, result_id, localfile, remotepath))242        self._upload_file(localfile, remotepath, self.recipe_id, task_id, result_id)243    def get_recipe(self):244        self.cmd_log.write('get_recipe: GET %s\n' % self.recipe_id)245        path = make_path_bkrcache(self.recipe_id)246        try:247            rpath = self.labc_url + make_path_recipe(self.recipe_id)248            utils.get_file(rpath, path)249        except Exception:250            # local will fall through to here251            if not os.path.isfile(path):252                raise BkrProxyException("No remote or cached recipe %s" % self.recipe_id)253        return open(path, 'r').read()254    def task_result(self, task_id, result_type, result_path,255                    result_score, result_summary):256        self.cmd_log.write('task_result: task_id(%s) result: %s, score: %s, summary: %s\n'257                           'path: %s\n' % (task_id, result_type, result_score, result_summary,258                                           result_path))259        data = {'result': result_type, 'path': result_path,260                'score': result_score, 'message': result_summary}261        path = self.labc_url + make_path_result(self.recipe_id, task_id)262        ret = copy_data(data, path, header='Location')263        # strip the path and return just the id264        return re.sub(path, "", ret)265    def task_start(self, task_id, kill_time=0):266        self.cmd_log.write('task_start: task_id(%s) kill_time(%s) RUNNING\n' % (task_id, kill_time))267        data = {'status': 'Running'}268        self.update_watchdog(task_id, kill_time)269        path = self.labc_url + make_path_status(self.recipe_id, task_id)270        copy_data(data, path)271    def task_stop(self, task_id):272        self.cmd_log.write('task_stop: task_id(%s) COMPLETED\n' % task_id)273        data = {'status': 'Completed'}274        path = self.labc_url + make_path_status(self.recipe_id, task_id)275        copy_data(data, path)...

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.