How to use make_path_result method in autotest

Best Python code snippet using autotest_python

bkr_proxy.py

Source:bkr_proxy.py Github

copy

Full Screen

...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)...

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:

YouTube

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?

Helpful

NotHelpful