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()),...
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!!