Best Python code snippet using autotest_python
bkr_proxy.py
Source:bkr_proxy.py
...32class BkrProxyException(Exception):33 def __init__(self, text):34 Exception.__init__(self, text)35'''Hard coded internal paths'''36def make_path_cmdlog(r):37 """38 Converts a recipe id into an internal path for logging purposes39 :param r: recipe id40 :return: a path to the internal command log41 """42 path = AUTOTEST_CACHE_DIR + '/recipes/' + r43 if not os.path.exists(path):44 os.makedirs(path)45 if not os.path.isdir(path):46 raise BkrProxyException("Path(%s) exists and is not a directory" % path)47 return path + '/cmd_log'48def make_path_bkrcache(r):49 """50 Converts a recipe id into an internal path for cache'ing recipe51 :param r: recipe id52 :return: a path to the internal recipe cache file53 """54 return AUTOTEST_CACHE_DIR + '/recipes/' + r + '/beaker_recipe.cache'55'''End Hard coded internal paths'''56"""57Hard coded paths as described in the Beaker Server API58http://beaker-project.org/dev/proposals/harness-api.html59"""60def make_path_recipe(r):61 """62 Converts a recipe id into a beaker path63 :param r: recipe id64 :return: a beaker path to the recipe id65 """66 return '/recipes/' + r67def make_path_watchdog(r):68 """69 Converts a recipe id into a beaker path for the watchdog70 :param r: recipe id71 :return: a beaker path of the recipe's watchdog file72 """73 return '/recipes/' + r + '/watchdog'74def make_path_status(r, t=None):75 """76 Converts id into a beaker path to status file77 Given a recipe id and/or a task id, translate them into78 the proper beaker path to the status file. Recipe only, returns79 the path to the recipe's status, whereas including a task returns80 the path to the task's status.81 :param r: recipe id82 :param t: task id83 :return: a beaker path of the recipe's/task's status file84 """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 as 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)...
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!!