KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use delete_arn_binding_for method in localstack

Best Python code snippet using localstack_python
provider.py
Source:provider.py
...150    ) -> DeleteSecretResponse:151        secret_id: str = request["SecretId"]152        self._raise_if_invalid_secret_id(secret_id)153        res = self._call_moto_with_request_secret_id(context, request)154        delete_arn_binding_for(context.region, secret_id)155        return res156    @handler("DescribeSecret", expand=False)157    def describe_secret(158        self, context: RequestContext, request: DescribeSecretRequest159    ) -> DescribeSecretResponse:160        self._raise_if_invalid_secret_id(request["SecretId"])161        res = self._call_moto_with_request_secret_id(context, request)162        return res163    @handler("GetResourcePolicy", expand=False)164    def get_resource_policy(165        self, context: RequestContext, request: GetResourcePolicyRequest166    ) -> GetResourcePolicyResponse:167        self._raise_if_invalid_secret_id(request["SecretId"])168        return self._call_moto_with_request_secret_id(context, request)169    @handler("GetSecretValue", expand=False)170    def get_secret_value(171        self, context: RequestContext, request: GetSecretValueRequest172    ) -> GetSecretValueResponse:173        self._raise_if_invalid_secret_id(request["SecretId"])174        return self._call_moto_with_request_secret_id(context, request)175    @handler("ListSecretVersionIds", expand=False)176    def list_secret_version_ids(177        self, context: RequestContext, request: ListSecretVersionIdsRequest178    ) -> ListSecretVersionIdsResponse:179        self._raise_if_invalid_secret_id(request["SecretId"])180        return self._call_moto_with_request_secret_id(context, request)181    @handler("PutResourcePolicy", expand=False)182    def put_resource_policy(183        self, context: RequestContext, request: PutResourcePolicyRequest184    ) -> PutResourcePolicyResponse:185        self._raise_if_invalid_secret_id(request["SecretId"])186        return self._call_moto_with_request_secret_id(context, request)187    @handler("PutSecretValue", expand=False)188    def put_secret_value(189        self, context: RequestContext, request: PutSecretValueRequest190    ) -> PutSecretValueResponse:191        self._raise_if_invalid_secret_id(request["SecretId"])192        return self._call_moto_with_request_secret_id(context, request)193    @handler("RemoveRegionsFromReplication", expand=False)194    def remove_regions_from_replication(195        self, context: RequestContext, request: RemoveRegionsFromReplicationRequest196    ) -> RemoveRegionsFromReplicationResponse:197        self._raise_if_invalid_secret_id(request["SecretId"])198        return self._call_moto_with_request_secret_id(context, request)199    @handler("ReplicateSecretToRegions", expand=False)200    def replicate_secret_to_regions(201        self, context: RequestContext, request: ReplicateSecretToRegionsRequest202    ) -> ReplicateSecretToRegionsResponse:203        self._raise_if_invalid_secret_id(request["SecretId"])204        return self._call_moto_with_request_secret_id(context, request)205    @handler("RestoreSecret", expand=False)206    def restore_secret(207        self, context: RequestContext, request: RestoreSecretRequest208    ) -> RestoreSecretResponse:209        self._raise_if_invalid_secret_id(request["SecretId"])210        return self._call_moto_with_request_secret_id(context, request)211    @handler("RotateSecret", expand=False)212    def rotate_secret(213        self, context: RequestContext, request: RotateSecretRequest214    ) -> RotateSecretResponse:215        self._raise_if_invalid_secret_id(request["SecretId"])216        return self._call_moto_with_request_secret_id(context, request)217    @handler("StopReplicationToReplica", expand=False)218    def stop_replication_to_replica(219        self, context: RequestContext, request: StopReplicationToReplicaRequest220    ) -> StopReplicationToReplicaResponse:221        self._raise_if_invalid_secret_id(request["SecretId"])222        return self._call_moto_with_request_secret_id(context, request)223    @handler("TagResource", expand=False)224    def tag_resource(self, context: RequestContext, request: TagResourceRequest) -> None:225        self._raise_if_invalid_secret_id(request["SecretId"])226        return self._call_moto_with_request_secret_id(context, request)227    @handler("UntagResource", expand=False)228    def untag_resource(self, context: RequestContext, request: UntagResourceRequest) -> None:229        self._raise_if_invalid_secret_id(request["SecretId"])230        return self._call_moto_with_request_secret_id(context, request)231    @handler("UpdateSecret", expand=False)232    def update_secret(233        self, context: RequestContext, request: UpdateSecretRequest234    ) -> UpdateSecretResponse:235        self._raise_if_invalid_secret_id(request["SecretId"])236        return self._call_moto_with_request_secret_id(context, request)237    @handler("UpdateSecretVersionStage", expand=False)238    def update_secret_version_stage(239        self, context: RequestContext, request: UpdateSecretVersionStageRequest240    ) -> UpdateSecretVersionStageResponse:241        self._raise_if_invalid_secret_id(request["SecretId"])242        return self._call_moto_with_request_secret_id(context, request)243    @handler("ValidateResourcePolicy", expand=False)244    def validate_resource_policy(245        self, context: RequestContext, request: ValidateResourcePolicyRequest246    ) -> ValidateResourcePolicyResponse:247        self._raise_if_invalid_secret_id(request["SecretId"])248        return self._call_moto_with_request_secret_id(context, request)249@patch(FakeSecret.__init__)250def fake_secret__init__(fn, self, *args, **kwargs):251    fn(self, *args, **kwargs)252    # Fix time not including millis.253    time_now = time.time()254    if kwargs.get("last_changed_date", None):255        self.last_changed_date = time_now256    if kwargs.get("created_date", None):257        self.created_date = time_now258    # The last date that the secret value was retrieved.259    # This value does not include the time.260    # This field is omitted if the secret has never been retrieved.261    self.last_accessed_date = None262    # Results in RotationEnabled being returned only if rotation was ever overwritten,263    # in which case this field is non-null, but an integer.264    self.auto_rotate_after_days = None265    self.rotation_lambda_arn = None266@patch(FakeSecret.update)267def fake_secret_update(268    fn, self, description=None, tags=None, kms_key_id=None, last_changed_date=None269):270    fn(self, description, tags, kms_key_id, last_changed_date)271    if last_changed_date is not None:272        self.last_changed_date = time.time()273class FakeSecretVersionStore(dict):274    def __setitem__(self, key, value):275        self.put_version(key, value, time.time())276    def put_version(self, version_id: str, version: Dict, create_date: Optional[float] = None):277        if create_date and "createdate" in version:278            version["createdate"] = create_date279        super().__setitem__(version_id, version)280@patch(FakeSecret.set_versions)281def fake_secret_set_versions(_, self, versions):282    self.versions = FakeSecretVersionStore()283    for version_id, version in versions.items():284        self.versions.put_version(version_id, version, self.created_date)285@patch(SecretsManagerBackend.get_secret_value)286def moto_smb_get_secret_value(fn, self, secret_id, version_id, version_stage):287    res = fn(self, secret_id, version_id, version_stage)288    secret_id = self.secrets[secret_id]289    if secret_id:  # Redundant, we know from the response it exists: no exceptions.290        secret_id.last_accessed_date = today_no_time()291    else:292        LOG.warning(293            f'Expected Secret to exist on non failing GetSecretValue request for SecretId "{secret_id}"'294        )295    return res296@patch(SecretsManagerBackend.create_secret)297def moto_smb_create_secret(fn, self, name, *args, **kwargs):298    # Creating a secret with a SecretId equal to one that is scheduled for299    # deletion should raise an 'InvalidRequestException'.300    secret: Optional[FakeSecret] = self.secrets.get(name, None)301    if secret is not None and secret.deleted_date is not None:302        raise InvalidRequestException(AWS_INVALID_REQUEST_MESSAGE_CREATE_WITH_SCHEDULED_DELETION)303    return fn(self, name, *args, **kwargs)304@patch(FakeSecret.to_dict)305def fake_secret_to_dict(fn, self):306    res_dict = fn(self)307    if self.last_accessed_date:308        res_dict["LastAccessedDate"] = self.last_accessed_date309    if not self.description and "Description" in res_dict:310        del res_dict["Description"]311    if not self.rotation_enabled and "RotationEnabled" in res_dict:312        del res_dict["RotationEnabled"]313    if self.auto_rotate_after_days is None and "RotationRules" in res_dict:314        del res_dict["RotationRules"]315    for null_field in [key for key, value in res_dict.items() if value is None]:316        del res_dict[null_field]317    return res_dict318@patch(SecretsManagerBackend.update_secret)319def backend_update_secret(320    fn,321    self,322    secret_id,323    description=None,324    **kwargs,325):326    fn(self, secret_id, **kwargs)327    secret = self.secrets[secret_id]328    # Fix missing update of secret description.329    # Secret exists if this point is reached.330    if description is not None:331        secret.description = description332    return secret.to_short_dict()333@patch(SecretsManagerResponse.update_secret)334def response_update_secret(_, self):335    secret_id = self._get_param("SecretId")336    description = self._get_param("Description")337    secret_string = self._get_param("SecretString")338    secret_binary = self._get_param("SecretBinary")339    client_request_token = self._get_param("ClientRequestToken")340    kms_key_id = self._get_param("KmsKeyId")341    return secretsmanager_backends[self.region].update_secret(342        secret_id=secret_id,343        description=description,344        secret_string=secret_string,345        secret_binary=secret_binary,346        client_request_token=client_request_token,347        kms_key_id=kms_key_id,348    )349@patch(SecretsManagerBackend.update_secret_version_stage)350def backend_update_secret_version_stage(351    fn, self, secret_id, version_stage, remove_from_version_id, move_to_version_id352):353    fn(self, secret_id, version_stage, remove_from_version_id, move_to_version_id)354    secret = self.secrets[secret_id]355    if version_stage == AWSCURRENT:356        secret.default_version_id = move_to_version_id357        # Ensure only one AWSPREVIOUS tagged version is in the pool.358        # Remove secret versions with no version stages.359        versions_no_stages = []360        update_vid_set = {remove_from_version_id, move_to_version_id}361        for version_id, version in secret.versions.items():362            version_stages = version["version_stages"]363            if version_id not in update_vid_set and AWSPREVIOUS in version_stages:364                version_stages.remove(AWSPREVIOUS)365                if not version_stages:366                    versions_no_stages.append(version_id)367        for version_no_stages in versions_no_stages:368            del secret.versions[version_no_stages]369    return json.dumps({"ARN": secret.arn, "Name": secret.name})370@patch(FakeSecret.reset_default_version)371def fake_secret_reset_default_version(fn, self, secret_version, version_id):372    fn(self, secret_version, version_id)373    # Remove versions with no version stages.374    versions_no_stages = [375        version_id for version_id, version in self.versions.items() if not version["version_stages"]376    ]377    for version_no_stages in versions_no_stages:378        del self.versions[version_no_stages]379@patch(FakeSecret.remove_version_stages_from_old_versions)380def fake_secret_remove_version_stages_from_old_versions(fn, self, version_stages):381    fn(self, version_stages)382    # Remove versions with no version stages.383    versions_no_stages = [384        version_id for version_id, version in self.versions.items() if not version["version_stages"]385    ]386    for version_no_stages in versions_no_stages:387        del self.versions[version_no_stages]388@patch(SecretsManagerBackend.rotate_secret)389def backend_rotate_secret(390    _,391    self,392    secret_id,393    client_request_token=None,394    rotation_lambda_arn=None,395    rotation_rules=None,396):397    rotation_days = "AutomaticallyAfterDays"398    if not self._is_valid_identifier(secret_id):399        raise SecretNotFoundException(f"Unable to find secret '{secret_id}'")400    if self.secrets[secret_id].is_deleted():401        raise InvalidRequestException(402            "An error occurred (InvalidRequestException) when calling the RotateSecret operation: You tried to \403            perform the operation on a secret that's currently marked deleted."404        )405    if rotation_lambda_arn:406        if len(rotation_lambda_arn) > 2048:407            msg = "RotationLambdaARN " "must <= 2048 characters long."408            raise InvalidParameterException(msg)409    if rotation_rules:410        if rotation_days in rotation_rules:411            rotation_period = rotation_rules[rotation_days]412            if rotation_period < 1 or rotation_period > 1000:413                msg = "RotationRules.AutomaticallyAfterDays " "must be within 1-1000."414                raise InvalidParameterException(msg)415    rotation_func = None416    try:417        lm_client = aws_stack.connect_to_service("lambda", region_name=self.region_name)418        get_func_res = lm_client.get_function(FunctionName=rotation_lambda_arn)419        lm_spec = get_func_res["Configuration"]420        lm_spec["Code"] = {"ZipFile": str(short_uid())}421        rotation_func = LambdaFunction(lm_spec, self.region_name)422    except Exception:423        # Fall through to ResourceNotFoundException.424        pass425    #426    if not rotation_func:427        raise ResourceNotFoundException("Lambda does not exist or could not be accessed")428    secret = self.secrets[secret_id]429    # The rotation function must end with the versions of the secret in430    # one of two states:431    #432    #  - The AWSPENDING and AWSCURRENT staging labels are attached to the433    #    same version of the secret, or434    #  - The AWSPENDING staging label is not attached to any version of the secret.435    #436    # If the AWSPENDING staging label is present but not attached to the same437    # version as AWSCURRENT then any later invocation of RotateSecret assumes438    # that a previous rotation request is still in progress and returns an error.439    try:440        version = next(441            version442            for version in secret.versions.values()443            if AWSPENDING in version["version_stages"]444        )445        if AWSCURRENT in version["version_stages"]:446            msg = "Previous rotation request is still in progress."447            raise InvalidRequestException(msg)448    except StopIteration:449        # Pending is not present in any version450        pass451    # Begin the rotation process for the given secret by invoking the lambda function.452    #453    # We add the new secret version as "pending". The previous version remains454    # as "current" for now. Once we've passed the new secret through the lambda455    # rotation function (if provided) we can then update the status to "current".456    new_version_id = self._from_client_request_token(client_request_token)457    #458    self._add_secret(459        secret_id,460        None,461        description=secret.description,462        tags=secret.tags,463        version_id=new_version_id,464        version_stages=[AWSPENDING],465    )466    secret.rotation_lambda_arn = rotation_lambda_arn467    if rotation_rules:468        secret.auto_rotate_after_days = rotation_rules.get(rotation_days, 0)469    if secret.auto_rotate_after_days > 0:470        secret.rotation_enabled = True471    request_headers = {}472    response_headers = {}473    for step in ["create", "set", "test", "finish"]:474        rotation_func.invoke(475            json.dumps(476                {477                    "Step": step + "Secret",478                    "SecretId": secret.name,479                    "ClientRequestToken": new_version_id,480                }481            ),482            request_headers,483            response_headers,484        )485    secret.set_default_version_id(new_version_id)486    version_stages = secret.versions[new_version_id]["version_stages"]487    if AWSPENDING in version_stages:488        version_stages.remove(AWSPENDING)489    return secret.to_short_dict()490def get_arn_binding_key_for(region: str, secret_id: str) -> str:491    return f"{region}_{secret_id}"492def get_arn_binding_for(region, secret_id):493    k = get_arn_binding_key_for(region, secret_id)494    if k not in SECRET_ARN_STORAGE:495        id_string = short_uid()[:6]496        arn = aws_stack.secretsmanager_secret_arn(497            secret_id, account_id=get_aws_account_id(), region_name=region, random_suffix=id_string498        )499        SECRET_ARN_STORAGE[k] = arn500    return SECRET_ARN_STORAGE[k]501def delete_arn_binding_for(region: str, secret_id: str) -> None:502    k = get_arn_binding_key_for(region, secret_id)503    if k in SECRET_ARN_STORAGE:504        del SECRET_ARN_STORAGE[k]505# patching resource policy in moto506def get_resource_policy_model(self, secret_id):507    if self._is_valid_identifier(secret_id):508        result = {509            "ARN": self.secrets[secret_id].arn,510            "Name": self.secrets[secret_id].secret_id,511        }512        policy = getattr(self.secrets[secret_id], "policy", None)513        if policy:514            result["ResourcePolicy"] = json.dumps(policy)515        return json.dumps(result)...
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.