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

How to use _attributes_to_dict method in autotest

Best Python code snippet using autotest_python

extended_kernel.py

Source:extended_kernel.py

...58    #            if isinstance(data_type_lvl1, dict) or isinstance(data_type_lvl1, list):59    #        for k,v in dict_:60                ## Find types 61    62    def _attributes_to_dict(self, input_):63        if input_:64            {att.get("key"):att.get("value") for att in input_.get("attributes")}65        else:66            return input_67    def _add_attributes(self, dict_, task_context):68        dict_["Version"] = task_context.get_version()69        dict_["Model Name"] = task_context.get_model_name()70        dict_["Task ID"] = task_context.get_id()71        dict_["Session ID"] = task_context.get_session_id()72        return dict_73    def _flatten_tasks(self, task_context):74        data_list = []75        attributes_list = []76        while task_context:77            input_data = json.loads(task_context.get_input_data())78            attributes = self._add_attributes(input_data.get("attributes"), task_context)79            data = input_data.get("data")80            data_list.append(data)81            attributes_list.append(attributes)82            task_context.next()83        return data_list, attributes_list84    def on_kernel_start(self, kernel_context):85        InferenceKernel.log_info("Kernel Start")86        InferenceKernel.log_info("Model Instance ID: " + kernel_context.get_instance_id())87        try: 88            self.install_modules()89            self.import_modules()90        except Exception as e:91            InferenceKernel.log_error("-------------------------")92            InferenceKernel.log_error( str(e) )93            task_context.set_output_data("Failed due to: " + str(e))94            InferenceKernel.log_error("-------------------------")95        model_details = json.loads(kernel_context.get_model_description())96        97        InferenceKernel.log_info("Model Input Details: " +  json.dumps(model_details)98        )99        model_path = model_details.get("model_path")100        InferenceKernel.log_info("Model Path Details: " +  model_path)101        self.attributes = self._attributes_to_dict(model_details.get("attributes"))102        InferenceKernel.log_info("Getting Model Path")103        try:104            model_path = self.get_model(model_path, **attributes)105            InferenceKernel.log_info("New Model Path: " + model_path)106        except:107            InferenceKernel.log_error("Failed to get model path")108        try:109            self.model = self.load_model(model_path, **self.attributes)110        except:111            InferenceKernel.log_error("Failed to load model")112    def on_task_invoke(self, task_context):113        try:114            ## Setup required empty lists to store data 115            ## NOTE task context is an array of dicts ...

calnet.py

Source:calnet.py

...61            csids_batch = csids[i:i + BATCH_QUERY_MAXIMUM]62            with self.connect() as conn:63                search_filter = self._ldap_search_filter(csids_batch, 'berkeleyeducsid', search_expired)64                conn.search('dc=berkeley,dc=edu', search_filter, attributes=ldap3.ALL_ATTRIBUTES)65                all_out += [_attributes_to_dict(entry, search_expired) for entry in conn.entries]66        return all_out67    def search_uids(self, uids, search_expired=False):68        all_out = []69        for i in range(0, len(uids), BATCH_QUERY_MAXIMUM):70            uids_batch = uids[i:i + BATCH_QUERY_MAXIMUM]71            with self.connect() as conn:72                search_filter = self._ldap_search_filter(uids_batch, 'uid', search_expired)73                conn.search('dc=berkeley,dc=edu', search_filter, attributes=ldap3.ALL_ATTRIBUTES)74                all_out += [_attributes_to_dict(entry, search_expired) for entry in conn.entries]75        return all_out76    @classmethod77    def _ldap_search_filter(cls, ids, id_type, search_expired=False):78        ids_filter = ''.join(f'({id_type}={_id})' for _id in ids)79        ou_scope = '(ou=expired people)' if search_expired else '(ou=people) (ou=advcon people)'80        return f"""(&81            (objectclass=person)82            (|83                {ids_filter}84            )85            (|86                { ou_scope }87            )88        )"""89class MockClient(Client):90    def __init__(self, app):91        self.app = app92        self.host = app.config['LDAP_HOST']93        self.bind = app.config['LDAP_BIND']94        self.password = app.config['LDAP_PASSWORD']95        server = ldap3.Server.from_definition(self.host, _fixture_path('server_info'), _fixture_path('server_schema'))96        self.server = server97    def connect(self):98        conn = ldap3.Connection(self.server, user=self.bind, password=self.password, client_strategy=ldap3.MOCK_SYNC)99        conn.strategy.entries_from_json(_fixture_path('search_entries'))100        return conn101def _attributes_to_dict(entry, expired_per_ldap):102    out = dict.fromkeys(SCHEMA_DICT.values(), None)103    out['expired'] = expired_per_ldap104    # ldap3's entry.entry_attributes_as_dict would work for us, except that it wraps a single value as a list.105    for attr in SCHEMA_DICT:106        if attr in entry.entry_attributes:107            out[SCHEMA_DICT[attr]] = entry[attr].value108    return out109def _create_fixtures(app, sample_csids):110    fixture_output = os.environ.get('FIXTURE_OUTPUT_PATH') or mockingbird._get_fixtures_path()111    cl = Client(app)112    cl.server.info.to_file(f'{fixture_output}/calnet_server_info.json')113    cl.server.schema.to_file(f'{fixture_output}/calnet_server_schema.json')114    conn = cl.connect()115    conn.search('ou=people,dc=berkeley,dc=edu', cl._csids_filter(sample_csids), attributes=ldap3.ALL_ATTRIBUTES)...

db.py

Source:db.py

...10        region_name=AWS_REGION,11        aws_access_key_id=AWS_ACCESS_KEY_ID,12        aws_secret_access_key=AWS_SECRET_KEY13    )14def _attributes_to_dict(attributes):15    return {x['Name']: x['Value'] for x in attributes}16def _dict_to_attributes(attrs):17    return [{'Name': k, 'Value': str(v), 'Replace': True} for k, v in attrs.iteritems()]18def set_current_gps_data(data):19    data['timestamp'] = arrow.utcnow().isoformat()20    _db().put_attributes(21        DomainName=TRACKING_DOMAIN,22        ItemName=str(uuid.uuid4()),23        Attributes=_dict_to_attributes(data)24    )25def start_journey():26    _db().put_attributes(27        DomainName=EVENTS_DOMAIN,28        ItemName=str(uuid.uuid4()),...

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.