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

How to use list_identities method in localstack

Best Python code snippet using localstack_python

memcache_bucket_config.py

Source:memcache_bucket_config.py

1#!/usr/bin/env python2import random3import logging4from google.appengine.ext import db5from google.appengine.api import memcache6from google.appengine.datastore import entity_pb7from google.appengine.ext import deferred8from util.model import Model9from util.consts import MEMCACHE_TIMEOUT10from util.consts import MEMCACHE_BUCKET_COUNTS11class MemcacheBucketConfig(Model):12    """Used so we can dynamically scale the number of memcache buckets"""13    name = db.StringProperty(indexed = True)14    count = db.IntegerProperty(default = 20)15    _memcache_key_name = 'name'16    _memcache_fields = ['id', 'name']17    def __init__(self, *args, **kwargs):18        self._memcache_key = kwargs[self._memcache_key_name] if self._memcache_key_name in kwargs else None19        super(MemcacheBucketConfig, self).__init__(*args, **kwargs)20    def _validate_self(self):21        return True22    def get_bucket(self, number):23        return '%s:%s' % (self.name, number)24    def get_random_bucket(self):25        bucket = random.randint(0, self.count)26        return self.get_bucket(bucket)27    def decrement_count(self):28        if self.count == 0:29            logging.error('trying to decrement count for %s lower than 0' %30                    self.name)31        else:32            self.count -= 133            self.put()34    def increment_count(self):35        self.count += 136        self.put()37    def put_later(self, entity):38        """Will put this entity in a bucket!"""39        key = entity.get_key()40        memcache.set(key, db.model_to_protobuf(entity).Encode(), time=MEMCACHE_TIMEOUT)41        bucket = self.get_random_bucket()42        #logging.info('mbc: %s' % self.name)43        #logging.info('bucket: %s' % bucket)44        list_identities = memcache.get(bucket) or []45        list_identities.append(key)46        #logging.info('bucket length: %d/%d' % (len(list_identities), self.count))47        if len(list_identities) > self.count:48            memcache.set(bucket, [], time=MEMCACHE_TIMEOUT)49            #logging.warn('bucket overflowing, persisting!')50            deferred.defer(batch_put, self.name, bucket, list_identities, _queue='slow-deferred')51        else:52            memcache.set(bucket, list_identities, time=MEMCACHE_TIMEOUT)53        #logging.info('put_later: %s' % key)54    @staticmethod55    def create(name, count=None):56        if not count:57            if name in MEMCACHE_BUCKET_COUNTS:58                count = MEMCACHE_BUCKET_COUNTS[name]59            else:60                count = MEMCACHE_BUCKET_COUNTS['default']61        mbc = MemcacheBucketConfig(name=name, count=count)62        if mbc:63            mbc.put()64        return mbc65    @staticmethod66    def get_or_create(name, count=None):67        mbc = MemcacheBucketConfig.get(name)68        if not mbc:69            # we are creating this MBC for the first time70            mbc = MemcacheBucketConfig.create(name, count)71        return mbc72    @classmethod73    def _get_from_datastore(cls, name):74        """Datastore retrieval using memcache_key"""75        return cls.all().filter('%s =' % cls._memcache_key_name, name).get()76def batch_put(mbc_name, bucket_key, list_keys, decrementing=False):77    """ Deferred task to put buckets in datastore78    Deferred is going to pickle this, so to remain usable:79      - must be a function so the pickled size is reasonable80      - must import all module dependencies so that the unpickled code knows81        where to find them82    """83    from apps.user.models import MemcacheBucketConfig, batch_put84    logging.info("Batch putting %s to memcache: %s" % (mbc_name, list_keys))85    mbc = MemcacheBucketConfig.get_or_create(mbc_name)86    entities_to_put = []87    had_error = False88    object_dict = memcache.get_multi(list_keys)89    for key in list_keys:90        data = object_dict.get(key)91        try:92            entity = db.model_from_protobuf(entity_pb.EntityProto(data))93            if entity:94                entities_to_put.append(entity)95        except AssertionError, e:96            old_key = mbc.get_bucket(mbc.count)97            if bucket_key != old_key and not decrementing and not had_error:98                old_count = mbc.count99                mbc.decrement_count()100                logging.warn(101                    'encounted error, going to decrement buckets from %s to %s'102                    % (old_count, mbc.count), exc_info=True)103                last_keys = memcache.get(old_key) or []104                memcache.set(old_key, [], time=MEMCACHE_TIMEOUT)105                deferred.defer(batch_put, mbc_name, old_key, last_keys,106                        decrementing=True, _queue='slow-deferred')107                had_error = True108        except Exception, e:109            logging.error('error getting object: %s' % e, exc_info=True)110    try:111        #def txn():112        db.put_async(entities_to_put)113        #db.run_in_transaction(txn)114        for entity in entities_to_put:115            if entity.key():116                memcache_key = entity.get_key()117                memcache.set(memcache_key,118                        db.model_to_protobuf(entity).Encode(),119                        time=MEMCACHE_TIMEOUT)120    except Exception,e:121        logging.error('Error putting %s: %s' % (entities_to_put, e), exc_info=True)122    if decrementing:123        logging.warn('decremented mbc `%s` to %d and removed %s' % (...

ses_barrel.py

Source:ses_barrel.py

...13    ])14    def __init__(self, oil, **kwargs):15        super().__init__(oil, **kwargs)16        self.cache = {}17    def list_identities(self):18        if self.cache.get('list_identities'):19            return self.cache['list_identities']20        identities_by_region = {}21        for region, client in self.clients.items():22            paginator = self.clients[region].get_paginator(23                'list_identities',24            )25            response_iterator = paginator.paginate()26            identities = []27            for page in response_iterator:28                identities.extend(page['Identities'])29            identities_by_region[region] = identities30        self.cache['list_identities'] = identities_by_region31        return identities_by_region32    def get_identity_dkim_attributes(self):33        """34        Note: There is a hard limit of 100 for identities sent into the35        getIdentityDkimAttributes API call. Additional logic will be needed36        if we need to expand functionality to accounts that exceed 10037        identities in a region.38        """39        identities_by_region = self.list_identities()40        attributes_by_region = {}41        for region, client in self.clients.items():42            identities = identities_by_region[region]43            response = client.get_identity_dkim_attributes(44                Identities=identities45            )46            attributes_by_region[region] = response['DkimAttributes']47        self.cache['get_identity_dkim_attributes'] = attributes_by_region...

list_identities_test.py

Source:list_identities_test.py

1# Copyright 2012 Viewfinder Inc. All Rights Reserved.2"""Test list_identities service API.3"""4__authors__ = ['spencer@emailscrubbed.com (Spencer Kimball)',5               'andy@emailscrubbed.com (Andy Kimball)']6from copy import deepcopy7from functools import partial8from viewfinder.backend.base.testing import async_test9from viewfinder.backend.db.identity import Identity10from viewfinder.backend.www.test import service_base_test11class ListIdentitiesTestCase(service_base_test.ServiceBaseTestCase):12  def testListIdentities(self):13    """Test listing of identities after linking and unlinking a new identity."""14    self._tester.LinkFacebookUser({'id': 100}, user_cookie=self._cookie)15    self._tester.ListIdentities(self._cookie)16    self._tester.UnlinkIdentity(self._cookie, 'FacebookGraph:100')17    self._tester.ListIdentities(self._cookie)18def _TestListIdentities(tester, user_cookie, request_dict):19  """Called by the ServiceTester in order to test list_identities service API call.20  """21  validator = tester.validator22  user_id, device_id = tester.GetIdsFromCookie(user_cookie)23  request_dict = deepcopy(request_dict)24  # Send list_identities request.25  actual_dict = tester.SendRequest('list_identities', user_cookie, request_dict)26  expected_dict = {'user_identities': []}27  predicate = lambda ident: ident.user_id == user_id28  for expected_ident in validator.QueryModelObjects(Identity, predicate=predicate):29    ident_dict = {'identity': expected_ident.key}30    if expected_ident.authority is not None:31      ident_dict['authority'] = expected_ident.authority32    expected_dict['user_identities'].append(ident_dict)33  tester._CompareResponseDicts('list_identities', user_id, request_dict, expected_dict, actual_dict)...

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.