Best Python code snippet using localstack_python
manifest.py
Source:manifest.py
...114 if resp.status_code != 200:115 return ManifestResource.error_response(116 resp.status_code, resp.error_message)117 # found it! replace hostname, adjust other stuff118 fixed_manif_string = self.fix_placeholders(119 resp.manifest_str,120 service_info,121 )122 # save it back to resp obj to jsonify123 resp.manifest_str = fixed_manif_string124 # save in filesys cache125 self.save_to_filecache_as_string(manifest_id, resp.manifest_str)126 # return manifest127 return resp.manifest_obj, 200128 def parse_id(self, manifest_id):129 try:130 source, doc_id = manifest_id.split(131 app.config['HX_MANIFEST_ID_SEPARATOR_IN_URL'])132 except Exception:133 return (None, None, None)134 else:135 # internal manifest_id136 mid = '{}{}{}'.format(137 source,138 app.config['HX_MANIFEST_ID_SEPARATOR_IN_HXPREZI'],139 doc_id)140 if source in app.config['PROXIES']:141 return (source, doc_id, mid)142 else:143 return ('hx', mid, mid)144 @classmethod145 def error_response(cls, code, msg):146 return {147 'error_code': code,148 'error_message': msg,149 }150 def get_service_info(self, source):151 if source in app.config['PROXIES']:152 return app.config['PROXIES'][source]153 else:154 return None155 def make_url_for_service(self, doc_id, service_info):156 service_url = 'https://{0}/{1}/{2}{3}'.format(157 service_info['manifests']['hostname'],158 service_info['manifests']['path'],159 service_info['manifests']['id_prefix'],160 doc_id,161 )162 return service_url163 def fetch_from_service(self, service_url):164 """ http request the manifest from 3rd party service (proxy)."""165 try:166 r = requests.get(service_url, timeout=REQUEST_TIMEOUT_IN_SEC)167 except requests.exceptions.RequestException as e:168 status_code = 503169 emsg = 'unable to fetch manifest from ({0}) - {1}'.format(170 service_url, e)171 return ManifestResourceResponse(503, error_message=emsg)172 if r.status_code != 200:173 emsg = 'error fetching manifest from ({0}) - {1}'.format(174 service_url, r.status_code)175 return ManifestResourceResponse(r.status_code, error_message=emsg)176 try:177 response = r.json()178 except ValueError as e:179 emsg = 'error decoding json response from ({0}) - {1}'.format(180 service_url, e)181 status_code = 502182 return ManifestResource.error_response(502, emsg)183 return ManifestResourceResponse(200, json_as_object=response)184 def fetch_from_file(self, doc_id, from_cache=False):185 """ load the manifest from local filesys; implies an hx manifest."""186 if from_cache:187 basedir = app.config['LOCAL_MANIFESTS_CACHE_DIR']188 else:189 basedir = app.config['LOCAL_MANIFESTS_SOURCE_DIR']190 manifest_as_json_string = None191 manifest_path = os.path.join(basedir, '{0}.json'.format(doc_id))192 logging.getLogger(__name__).debug(193 '********** manifest_path {}'.format(manifest_path))194 if os.path.exists(manifest_path) \195 and os.path.isfile(manifest_path) \196 and os.access(manifest_path, os.R_OK):197 with open(manifest_path, 'r') as fd:198 manifest_as_json_string = fd.read()199 if manifest_as_json_string is None:200 response = ManifestResourceResponse(201 404, # not found202 error_message='local manifest ({0}) not found'.format(doc_id))203 else:204 response = ManifestResourceResponse(205 200, json_as_string=manifest_as_json_string)206 return response207 def save_to_filecache_as_string(self, doc_id, manifest_string):208 """ save manifest string to filesys as hx source.209 once saved, we don't fetch it from 3rd party anymore.210 """211 manifest_path = os.path.join(212 app.config['LOCAL_MANIFESTS_CACHE_DIR'],213 '{0}.json'.format(doc_id))214 with open(manifest_path, 'w') as fd:215 fd.write(manifest_string)216 def fix_placeholders(217 self, json_string, service_info):218 # replace placeholders that point to hostnames that we are proxying!219 response_string = json_string.replace(220 service_info['manifests']['placeholder'],221 app.config['HX_SERVERS']['manifests']['hostname'],222 )223 response_string = response_string.replace(224 service_info['images']['placeholder'],225 app.config['HX_SERVERS']['images']['hostname'],226 )227 if app.config['HX_REPLACE_HTTPS']:228 response_string = response_string.replace('https:', 'http:')229 return response_string230 def fix_local_service_context(self, manifest_obj):...
setup_run.py
Source:setup_run.py
...8from dxs import paths9system_config_path = paths.config_path / "system_config.yaml"10with open(system_config_path, "r") as f:11 system_config = yaml.load(f, Loader=yaml.FullLoader)12def fix_placeholders(config, placeholders=None):13 if placeholders is None:14 placeholders = {}15 print("placeholders", placeholders)16 for k, v in config.items():17 if type(v) is str:18 if "@" in v:19 spl = v.split("/")20 rpl = [placeholders[x[1:]] if "@" in x else x for x in spl]21 join = Path(*tuple(rpl))22 config[k] = str(join)23 if "$" in v:24 print("in dollar rplace")25 for pl_k, pl_v in placeholders.items():26 if not pl_k.startswith("$"):27 continue28 if pl_k in v:29 v = v.replace(pl_k, pl_v)30 config[k] = v31 placeholders[k] = config[k]32 elif isinstance(v,dict):33 config[k] = fix_placeholders(v, placeholders=placeholders)34 return config35def read_run_config(run_config_path, placeholders=None):36 with open(run_config_path, "r") as f:37 run_config = yaml.load(f, Loader=yaml.FullLoader)38 run_config = fix_placeholders(run_config, placeholders=placeholders)39 for k, v in run_config.get("kwargs", {}).items():40 if v == "None":41 run_config["kwargs"][k] = None42 return run_config43 44 45if __name__ == "__main__":46 parser = ArgumentParser()47 parser.add_argument(48 "-c", "--config", action="store", required=True, 49 help="config required. see dxs/runner/blank_config.yaml for an example")50 args = parser.parse_args()51 52 datestr = datetime.datetime.today().strftime("%y%m%d")...
main.py
Source:main.py
...15 16 active.append(names[i])17def visual_name(interactions, name):18 return f":{interactions[name]['emoji']}: [b blue]{name}[/]"19def fix_placeholders(interactions, s, a, b=None, c=None):20 s = s.replace("<$A$>", visual_name(interactions, a))21 if b:22 s = s.replace("<$B$>", visual_name(interactions, b))23 if c:24 s = s.replace("<$C$>", visual_name(interactions, c))25 return s26if __name__ == '__main__':27 console = Console(highlight=False)28 with open(f'{__file__.rstrip("main.py")}/interactions.json', 'r', encoding="utf-8") as f:29 interactions = json.loads(f.read())30 names = list(interactions.keys())31 active = []32 activate_dwarf(names, active)33 activate_dwarf(names, active)34 35 i = 036 while len(active) > 0:37 console.print(f"[green][ {', '.join([visual_name(interactions, name) for name in active])} ][/]\n")38 match (ORDER[i], len(active)):39 case (_, 1): # Only one dwarf remaining40 console.print(fix_placeholders(interactions, interactions[active[-1]]["last"], active[-1]))41 del active[0]42 case (0, _): # Remove a dwarf43 console.print(fix_placeholders(interactions, interactions[active[-2]]["remove"], active[-2], active[-1]))44 del active[-2]45 case (1, _): # Add a dwarf46 activate_dwarf(names, active)47 console.print(fix_placeholders(interactions, interactions[active[-3]]["add"], active[-3], active[-2], active[-1]))48 console.print("[yellow]-------------------------------------------[/]")...
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!!