Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use fix_placeholders method in localstack

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]-------------------------------------------[/]")...

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.