KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use read_api_key_safe method in localstack

Best Python code snippet using localstack_python

event_publisher.py

Source:event_publisher.py

...37        self.t = kwargs.get('timestamp') or kwargs.get('t') or timestamp()38        self.m_id = kwargs.get('machine_id') or kwargs.get('m_id') or get_machine_id()39        self.p_id = kwargs.get('process_id') or kwargs.get('p_id') or get_process_id()40        self.p = kwargs.get('payload') if kwargs.get('payload') is not None else kwargs.get('p')41        self.k = kwargs.get('api_key') or kwargs.get('k') or read_api_key_safe()42        self.e_t = kwargs.get('event_type') or kwargs.get('e_t')43    def timestamp(self):44        return self.t45    def machine_id(self):46        return self.m_id47    def process_id(self):48        return self.p_id49    def event_type(self):50        return self.e_t51    def payload(self):52        return self.p53    def api_key(self):54        return self.k55def read_api_key_safe():56    try:57        from localstack_ext.bootstrap.licensing import read_api_key58        return read_api_key()59    except Exception:60        return None61def get_or_create_file(config_file):62    if os.path.exists(config_file):63        return config_file64    try:65        save_file(config_file, '{}')66        return config_file67    except Exception:68        pass69def get_config_file_homedir():70    return get_or_create_file(CONFIG_FILE_PATH)71def get_config_file_tempdir():72    return get_or_create_file(os.path.join(TMP_FOLDER, '.localstack'))73def get_machine_id():74    global MACHINE_ID75    if MACHINE_ID:76        return MACHINE_ID77    # determine MACHINE_ID from config files78    configs_map = {}79    config_file_tmp = get_config_file_tempdir()80    config_file_home = get_config_file_homedir()81    for config_file in (config_file_home, config_file_tmp):82        if config_file:83            local_configs = load_file(config_file)84            local_configs = json.loads(to_str(local_configs))85            configs_map[config_file] = local_configs86            if 'machine_id' in local_configs:87                MACHINE_ID = local_configs['machine_id']88                break89    # if we can neither find NOR create the config files, fall back to process id90    if not configs_map:91        return PROCESS_ID92    # assign default id if empty93    if not MACHINE_ID:94        MACHINE_ID = short_uid()95    # update MACHINE_ID in all config files96    for config_file, configs in configs_map.items():97        configs['machine_id'] = MACHINE_ID98        save_file(config_file, json.dumps(configs))99    return MACHINE_ID100def get_process_id():101    return PROCESS_ID102def poll_and_send_messages(params):103    while True:104        try:105            event = EVENT_QUEUE.get(block=True, timeout=None)106            event = event.to_dict()107            endpoint = '%s/events' % API_ENDPOINT.rstrip('/')108            requests.post(endpoint, json=event)109        except Exception:110            # silently fail, make collection of usage data as non-intrusive as possible111            time.sleep(1)112def is_travis():113    return os.environ.get('TRAVIS', '').lower() in ['true', '1']114def get_hash(name):115    if not name:116        return '0'117    max_hash = 10000000000118    hashed = hash(name) % max_hash119    hashed = hex(hashed).replace('0x', '')120    return hashed121def fire_event(event_type, payload=None):122    global SENDER_THREAD123    if not SENDER_THREAD:124        SENDER_THREAD = FuncThread(poll_and_send_messages, {})125        SENDER_THREAD.start()126    api_key = read_api_key_safe()127    if not api_key:128        # only store events if API key has been specified129        return130    if payload is None:131        payload = {}132    if isinstance(payload, dict):133        if is_travis():134            payload['travis'] = True135        if os.environ.get(ENV_INTERNAL_TEST_RUN):136            payload['int'] = True137    event = AnalyticsEvent(event_type=event_type, payload=payload, api_key=api_key)...

metadata.py

Source:metadata.py

...34def read_client_metadata() -> ClientMetadata:35    return ClientMetadata(36        session_id=get_session_id(),37        machine_id=get_machine_id(),38        api_key=read_api_key_safe(),39        system=get_system(),40        version=get_version_string(),41        is_ci=os.getenv("CI") is not None,42        is_docker=config.is_in_docker,43        is_testing=config.is_env_true(constants.ENV_INTERNAL_TEST_RUN),44    )45@functools.lru_cache()46def get_session_id() -> str:47    return _generate_session_id()48@functools.lru_cache()49def get_client_metadata() -> ClientMetadata:50    metadata = read_client_metadata()51    if config.DEBUG_ANALYTICS:52        LOG.info("resolved client metadata: %s", metadata)53    return metadata54@functools.lru_cache()55def get_machine_id() -> str:56    cache_path = os.path.join(config.dirs.cache, "machine.json")57    doc = FileMappedDocument(cache_path)58    if "machine_id" not in doc:59        # generate a machine id60        doc["machine_id"] = _generate_machine_id()61        # try to cache the machine ID62        call_safe(doc.save)63    return doc["machine_id"]64@hooks.prepare_host()65def prepare_host_machine_id():66    # lazy-init machine ID into cache on the host, which can then be used in the container67    get_machine_id()68def _generate_session_id() -> str:69    return long_uid()70def _generate_machine_id() -> str:71    if config.is_in_docker:72        return short_uid()73    # this can potentially be useful when generated on the host using the CLI and then mounted into the container via74    # machine.json75    try:76        if os.path.exists("/etc/machine-id"):77            with open("/etc/machine-id") as fd:78                return md5(str(fd.read()))[:8]79    except Exception:80        pass81    # always fall back to short_uid()82    return short_uid()83def read_api_key_safe():84    try:85        from localstack_ext.bootstrap.licensing import read_api_key86        return read_api_key(raise_if_missing=False)87    except Exception:88        return None89def get_system() -> str:...

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.