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

How to use _notify_event_subscribers method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...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:119            moto_res = call_moto(context)120        SsmProvider._notify_event_subscribers(nname, "Create")121        return PutParameterResult(**moto_res)122    def get_parameter(123        self,124        context: RequestContext,125        name: PSParameterName,126        with_decryption: Boolean = None,127    ) -> GetParameterResult:128        result = None129        #130        name = SsmProvider._normalize_name(name)131        details = name.split("/")132        if len(details) > 4:133            service = details[3]134            if service == "secretsmanager":135                resource_name = "/".join(details[4:])136                result = SsmProvider._get_secrets_information(name, resource_name)137        #138        if not result:139            result = call_moto_with_request(140                context, {"Name": name, "WithDecryption": bool(with_decryption)}141            )142        #143        return GetParameterResult(**result)144    def delete_parameter(145        self, context: RequestContext, name: PSParameterName146    ) -> DeleteParameterResult:147        SsmProvider._notify_event_subscribers(name, "Delete")148        call_moto(context)  # Return type is an emtpy type.149        return DeleteParameterResult()150    def label_parameter_version(151        self,152        context: RequestContext,153        name: PSParameterName,154        labels: ParameterLabelList,155        parameter_version: PSParameterVersion = None,156    ) -> LabelParameterVersionResult:157        SsmProvider._notify_event_subscribers(name, "LabelParameterVersion")...

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.