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

How to use _has_secrets method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...23from localstack.utils.bootstrap import is_api_enabled24PARAM_PREFIX_SECRETSMANAGER = "/aws/reference/secretsmanager"25class SsmProvider(SsmApi, ABC):26    @staticmethod27    def _has_secrets(names: ParameterNameList) -> Boolean:28        maybe_secret = next(29            filter(lambda n: n.startswith(PARAM_PREFIX_SECRETSMANAGER), names), None30        )31        return maybe_secret is not None32    @staticmethod33    def _normalize_name(param_name: ParameterName) -> ParameterName:34        param_name = param_name.strip("/")35        param_name = param_name.replace("//", "/")36        if "/" in param_name:37            param_name = "/%s" % param_name38        return param_name39    @staticmethod40    def _get_secrets_information(41        name: ParameterName, resource_name: str42    ) -> Optional[GetParameterResult]:43        client = aws_stack.connect_to_service("secretsmanager")44        try:45            secret_info = client.get_secret_value(SecretId=resource_name)46            del secret_info["ResponseMetadata"]47            created_date_timestamp = time.mktime(secret_info["CreatedDate"].timetuple())48            secret_info["CreatedDate"] = created_date_timestamp49            result = {50                "Parameter": {51                    "SourceResult": json.dumps(secret_info, default=str),52                    "Name": name,53                    "Value": secret_info.get("SecretString"),54                    "Type": "SecureString",55                    "LastModifiedDate": created_date_timestamp,56                }57            }58            return GetParameterResult(**result)59        except client.exceptions.ResourceNotFoundException:60            return None61    @staticmethod62    def _get_params_and_secrets(names: ParameterNameList) -> GetParametersResult:63        ssm_client = aws_stack.connect_to_service("ssm")64        result = {"Parameters": [], "InvalidParameters": []}65        for name in names:66            if name.startswith(PARAM_PREFIX_SECRETSMANAGER):67                secret = SsmProvider._get_secrets_information(68                    name, name[len(PARAM_PREFIX_SECRETSMANAGER) + 1 :]69                )70                if secret is not None:71                    secret = secret["Parameter"]72                    result["Parameters"].append(secret)73                else:74                    result["InvalidParameters"].append(name)75            else:76                try:77                    param = ssm_client.get_parameter(Name=name)78                    param["Parameter"]["LastModifiedDate"] = time.mktime(79                        param["Parameter"]["LastModifiedDate"].timetuple()80                    )81                    result["Parameters"].append(param["Parameter"])82                except ssm_client.exceptions.ParameterNotFound:83                    result["InvalidParameters"].append(name)84        return GetParametersResult(**result)85    @staticmethod86    def _notify_event_subscribers(name: ParameterName, operation: str):87        """Publish an EventBridge event to notify subscribers of changes."""88        if not is_api_enabled("events"):89            return90        events = aws_stack.connect_to_service("events")91        detail = {"name": name, "operation": operation}92        event = {93            "Source": "aws.ssm",94            "Detail": json.dumps(detail),95            "DetailType": "Parameter Store Change",96        }97        events.put_events(Entries=[event])98    def get_parameters(99        self,100        context: RequestContext,101        names: ParameterNameList,102        with_decryption: Boolean = None,103    ) -> GetParametersResult:104        if SsmProvider._has_secrets(names):105            return SsmProvider._get_params_and_secrets(names)106        names = list([SsmProvider._normalize_name(name) for name in names])107        request = {"Names": names, "WithDecryption": bool(with_decryption)}108        res = call_moto_with_request(context, request)109        return GetParametersResult(**res)110    def put_parameter(111        self, context: RequestContext, request: PutParameterRequest112    ) -> PutParameterResult:113        name = request["Name"]114        nname = SsmProvider._normalize_name(name)115        if name != nname:116            request.update({"Name": nname})117            moto_res = call_moto_with_request(context, request)118        else:...

env_secrets.py

Source:env_secrets.py

...27    return uncached28class LookupModule(LookupBase):29    @classmethod30    @cached31    def _has_secrets(cls):32        return os.path.exists("/keybase")33    def run(self, terms, variables, **kwargs):34        if len(terms) != 2:35            raise AnsibleError('Usage: lookup("env_secrets", SECTION, KEY)')36        [section, key] = terms37        if self._has_secrets():38            self._templar.available_variables = variables39            return [self._templar.template(variables["env_secret_" + section])[key]]40        else:...

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.