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

How to use _generate_arn method in localstack

Best Python code snippet using localstack_python

workspaces.py

Source:workspaces.py

1from __future__ import (absolute_import, division, print_function,2                        unicode_literals)3import functools4import itertools5from c7n.filters import ValueFilter6from c7n.manager import resources7from c7n.query import QueryResourceManager8from c7n.tags import universal_augment, register_universal_tags9from c7n.utils import generate_arn, local_session, type_schema, chunks10@resources.register('workspaces')11class Workspace(QueryResourceManager):12    class resource_type(object):13        service = 'workspaces'14        enum_spec = ('describe_workspaces', 'Workspaces', None)15        type = 'workspace'16        name = id = dimension = 'WorkspaceId'17        filter_name = None18    augment = universal_augment19    _generate_arn = None20    @property21    def generate_arn(self):22        if self._generate_arn is None:23            self._generate_arn = functools.partial(24                generate_arn, 'workspaces', region=self.config.region,25                account_id=self.account_id, resource_type='workspace', separator='/')26        return self._generate_arn27register_universal_tags(Workspace.filter_registry, Workspace.action_registry)28@Workspace.filter_registry.register('connection-status')29class WorkspaceConnectionStatusFilter(ValueFilter):30    """Filter Workspaces based on user connection information31    :example:32    .. code-block:: yaml33            policies:34              - name: workspaces-abandoned35                resource: workspaces36                filters:37                  - type: connection-status38                    value_type: age39                    key: LastKnownUserConnectionTimestamp40                    op: ge41                    value: 9042              - name: workspaces-expensive-zombies43                resource: workspaces44                filters:45                  - "WorkspaceProperties.RunningMode": ALWAYS_ON46                  - type: connection-status47                    value_type: age48                    key: LastKnownUserConnectionTimestamp49                    op: ge50                    value: 3051    """52    schema = type_schema('connection-status', rinherit=ValueFilter.schema)53    permissions = ('workspaces:DescribeConnectionStatus',)54    annotation_key = 'c7n:ConnectionStatus'55    def get_connection_status(self, client, workspace_ids):56        connection_status_chunk = self.manager.retry(57            client.describe_workspaces_connection_status,58            WorkspaceIds=workspace_ids59        )['WorkspacesConnectionStatus']60        return connection_status_chunk61    def process(self, resources, event=None):62        client = local_session(self.manager.session_factory).client('workspaces')63        annotate_map = {r['WorkspaceId']: r for r in resources if self.annotation_key not in r}64        with self.executor_factory(max_workers=2) as w:65            self.log.debug(66                'Querying connection status for %d workspaces' % len(annotate_map))67            for status in itertools.chain(*w.map(68                functools.partial(self.get_connection_status, client),69                chunks(annotate_map.keys(), 25)70            )):71                annotate_map[status['WorkspaceId']][self.annotation_key] = status72        return list(filter(self, resources))73    def get_resource_value(self, k, i):74        return super(WorkspaceConnectionStatusFilter, self).get_resource_value(...

10540_elasticsearch.py

Source:10540_elasticsearch.py

1# Copyright 2016-2017 Capital One Services, LLC2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from __future__ import absolute_import, division, print_function, unicode_literals15import functools16import logging17import itertools18from c7n.actions import Action19from c7n.manager import resources20from c7n.query import QueryResourceManager21from c7n.utils import (22    chunks, local_session, get_retry, type_schema, generate_arn)23log = logging.getLogger('custodian.es')24@resources.register('elasticsearch')25class ElasticSearchDomain(QueryResourceManager):26    class resource_type(object):27        service = 'es'28        type = "elasticsearch"29        enum_spec = (30            'list_domain_names', 'DomainNames[].DomainName', None)31        id = 'DomainName'32        name = 'Name'33        dimension = "DomainName"34        filter_name = None35    _generate_arn = _account_id = None36    retry = staticmethod(get_retry(('Throttled',)))37    @property38    def generate_arn(self):39        if self._generate_arn is None:40            self._generate_arn = functools.partial(41                generate_arn,42                'es',43                region=self.config.region,44                account_id=self.config.account_id,45                resource_type='domain',46                separator='/')47        return self._generate_arn48    def get_resources(self, resource_ids):49        client = local_session(self.session_factory).client('es')50        return client.describe_elasticsearch_domains(51            DomainNames=resource_ids)['DomainStatusList']52    def augment(self, domains):53        client = local_session(self.session_factory).client('es')54        model = self.get_model()55        def _augment(resource_set):56            resources = self.retry(57                client.describe_elasticsearch_domains,58                DomainNames=resource_set)['DomainStatusList']59            for r in resources:60                rarn = self.generate_arn(r[model.id])61                r['Tags'] = self.retry(62                    client.list_tags, ARN=rarn).get('TagList', [])63            return resources64        with self.executor_factory(max_workers=1) as w:65            return list(itertools.chain(66                *w.map(_augment, chunks(domains, 5))))67@ElasticSearchDomain.action_registry.register('delete')68class Delete(Action):69    schema = type_schema('delete')70    permissions = ('es:DeleteElastisearchDomain',)71    def process(self, resources):72        client = local_session(self.manager.session_factory).client('es')73        for r in resources:...

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.