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

How to use describe_queries method in localstack

Best Python code snippet using localstack_python

main.py

Source:main.py

1'''SMART API Verifier main controller'''2# Developed by: Nikolai Schwertner3#4# CONFIG: Change the consumer_secret in _ENDPOINT!5#6# Revision history:7#     2012-02-24 Initial release8#     2013-03-27 Upgraded to SMART v0.6 OAuth - Arjun Sanyal9import os10import sys11abspath = os.path.dirname(__file__)12sys.path.append(abspath)13import logging14import json15import tempfile16import time17import urllib18import web19from smart_client.client import SMARTClient20from smart_client.common.rdf_tools import rdf_ontology21from settings import APP_PATH, ONTOLOGY_PATH, ENDPOINT22from tests import runTest, getMessages, describeQueries23from threading import Lock24# Configuration25###########################################################################26logging.basicConfig(level=logging.DEBUG)27# SMART Container OAuth Endpoint28_ENDPOINT = ENDPOINT29# webpy file based sessions30###########################################################################31_session = web.session.DiskStore(tempfile.mkdtemp())32# SMARTClient and OAuth Helper Functions33###########################################################################34_smart = None  # A global flag to check is the SMARTClient is configured35def _smart_client(api_base, record_id=None):36    """ Returns the SMART client, configured accordingly. """37    global _smart38    if _smart is None:39        try:40            _smart = SMARTClient(_ENDPOINT.get('app_id'), api_base, _ENDPOINT)41        except Exception, e:42            logging.critical("Could not init SMARTClient. " + str(e))43    _smart.record_id = record_id44    return _smart45def _request_token_for_record(api_base, record_id):46    """ Requests a request token for a given record_id """47    global _session48    _session['req_token'] = None49    logging.debug("Requesting token for %s on %s" % (record_id, api_base))50    smart = _smart_client(api_base, record_id)51    smart.token = None52    try:53        _session['req_token'] = smart.fetch_request_token()54    except Exception, e:55        return False, str(e)56    return True, None57def _exchange_token(verifier):58    """ Exchanges verifier for an acc_token and stores it in the session """59    global _session60    record_id = _session['record_id']61    req_token = _session['req_token']62    api_base = _session['api_base']63    if record_id is None:64        logging.error("No record_id, cannot exchange %s" % req_token)65        return66    logging.debug("Exchanging token: %s" % req_token)67    smart = _smart_client(api_base, record_id)68    smart.update_token(req_token)69    try:70        acc_token = smart.exchange_token(verifier)71    except Exception, e:72        logging.error("Token exchange failed: %s" % e)73        return74    # success! store it75    logging.debug("Exchanged req_token for acc_token: %s" % acc_token)76    _session['acc_token'] = acc_token77    smart.update_token(acc_token)78    # store in cookies too79    web.setcookie('oauth_token_secret', acc_token['oauth_token_secret'])80    web.setcookie('oauth_token', acc_token['oauth_token'])81    web.setcookie('user_id', acc_token['user_id'])82    web.setcookie('api_base', api_base)83    web.setcookie('record_id', record_id)84    return True85def _test_token():86    """ Tries to fetch demographics with session acc_token and returns a87        bool whether thas was successful. """88    smart = _smart_client(_session['api_base'],89                          _session['record_id'])90    if smart is None:91        return False92    smart.update_token(_session['acc_token'])93    try:94        demo = smart.get_demographics()95        if '200' == demo.response.get('status'):96            return True97    except Exception, e:98        pass99    return False100# App URLs101###########################################################################102# URL mappings for web.py103urls = ('/smartapp/index.html', 'index',104        '/smartapp/authorize', 'authorize',105        '/smartapp/getcalls', 'get_calls',106        '/smartapp/apicall', 'api_call',107        '/smartapp/runtests', 'run_tests',108        '/smartapp/describe', 'describe_queries')109class index:110    '''Disseminator for the SMART tester index page'''111    def GET(self):112        # We should have the api_base and record_id in the query string113        # e.g we're not going to redirect to the record selection UI114        global _session115        _session['api_base'] = api_base = _ENDPOINT.get('url')116        _session['record_id'] = record_id = web.input().get('record_id')117        logging.debug('api_base: ' + str(api_base) +118                      ' record_id: ' + str(record_id))119        # Init the SMARTClient120        smart = _smart_client(api_base, record_id)121        # Do we have a valid access token?122        if 'acc_token' not in _session or not _test_token():123            # Nope, clear the acc and req tokens124            web.setcookie('oauth_token_secret', '', -1)125            web.setcookie('oauth_token', '', -1)126            web.setcookie('record_id', '', -1)127            web.setcookie('user_id', '', -1)128            _session['acc_token'] = None129            _session['req_token'] = None130            fetched, error_msg = _request_token_for_record(api_base, record_id)131            if fetched:132                logging.debug("Redirecting to authorize url: " +133                              smart.auth_redirect_url)134                raise web.seeother(smart.auth_redirect_url)135            if error_msg:136                logging.debug('_request_token_for_record failed')137                web.internalerror()138        # We are good to go139        template_html = web.template.frender(APP_PATH +140                                             '/templates/index.html')141        html = template_html("1.0")142        web.header('Content-Type', 'text/html')143        return html144class authorize:145    def GET(self):146        """ Extract the oauth_verifier and exchange it for an access token. """147        global _session148        new_oauth_token = web.input().oauth_token149        req_token = _session['req_token']150        api_base = _session['api_base']151        record_id = _session['record_id']152        if new_oauth_token != req_token.get('oauth_token', None):153            logging.critical('Token mismatch in /authorize! Aborting.')154            web.internalerror()155            return156        if _exchange_token(web.input().oauth_verifier):157            return web.seeother(158                '/smartapp/index.html?api_base=%s&record_id=%s' %159                (api_base, record_id))160        else:161            logging.critical('Could not exchange token.')162            web.internalerror()163class get_calls:164    def GET(self):165        '''Returns the available python client calls based on the ontology'''166        out = {}167        for t in rdf_ontology.api_calls:168            path = str(t.path)169            method = str(t.http_method)170            target = str(t.target)171            category = str(t.category)172            cardinality = str(t.cardinality)173            # Process only GET calls of "record_items" category plus a few174            # specific exceptions by adding them to the dictionary175            if method == "GET" and \176               ((category == "record" and cardinality == "multiple") or177                t.client_method_name in (178                    "get_demographics",179                    "get_ontology",180                    "get_container_manifest",181                    #"get_user_preferences",       # disabled182                    "get_app_manifests")):183                # Build the generic python client call name and use it184                # in the dictionary185                out[target] = str(t.client_method_name)186        return json.dumps(out, sort_keys=True, indent=4)187class api_call:188    def POST(self):189        '''Executes a python client API call identified by its generic name'''190        global _smart191        # make sure the SMARTClient is init'd192        cookies = web.cookies()193        api_base = cookies.api_base194        record_id = cookies.record_id195        # reconstruct acc_token from cookies196        acc_token = {197            'oauth_token_secret': cookies.oauth_token_secret,198            'oauth_token': cookies.oauth_token,199            'record_id': record_id,200            'user_id': cookies.user_id201        }202        logging.debug('Cookies are: api_base: ' + api_base +203                ' record_id: ' + record_id +204                ' acc_token: ' + str(acc_token))205        smart = _smart_client(api_base, record_id)206        if smart is None:207            return False208        smart.update_token(acc_token)209        call_name = web.input().call_name210        # Figure out the SMART model corresponding to the API call211        model = get_model(call_name)212        logging.debug('Calling ' + call_name)213        method_to_call = getattr(SMARTClient, call_name)214        r = method_to_call(_smart)215        # Run the API tests on the result of the call216        contentType = r.response.get('content-type', None)217        messages = getMessages(runTest(model, r.body, contentType))218        # Encode and return the call and tests result as JSON219        return json.dumps({220            'body': r.body,221            'contentType': contentType,222            'messages': messages223        }, sort_keys=True, indent=4)224class run_tests:225    def POST(self):226        '''Executes the appropriate series of tests for a given data model'''227        # Get the input data from the HTTP header228        model = web.input().model229        data = web.input().data230        contentType = web.input().content_type231        # Run the tests and obtain the failure messages232        messages = getMessages(runTest(model, data, contentType))233        # Return the failure messages encoded as JSON234        return json.dumps(messages, sort_keys=True, indent=4)235class describe_queries:236    def GET(self):237        '''Returns a string describing the queries used in testing a DM'''238        model = web.input().model239        return describeQueries(model)240def get_api_calls():241    calls = {}242    for t in rdf_ontology.api_calls:243        target = str(t.target)244        method = str(t.http_method)245        path = str(t.path)246        category = str(t.category)247        if method == "GET" and (category == "record_items" or248                                path == "/ontology" or249                                path == "/apps/manifests/" or250                                path == "/manifest"):251            if target not in calls.keys():252                calls[target] = path253    return calls254def get_model(call):255    '''Returns the name of the target SMART data model256    corresponding to the SMART python client convenience method257    Expects a valid SMART python client convenience method name258    '''259    # We may have to load the ontology if it is not available yet260    if not rdf_ontology.api_types:261        rdf_ontology.parse_ontology(open(ONTOLOGY_PATH).read())262    # Look through the api calls array until a call with matching263    # convenience method name is found264    for c in rdf_ontology.api_calls:265        if call == c.client_method_name:266            return c.target.replace("http://smartplatforms.org/terms#", "")267# Simulate single threading using a mutex. This is to prevent errors268# in httplib2 (used by oauth2) which is comically not threadsafe! Argh!!!269def mutex_processor():270    mutex = Lock()271    def processor_func(handle):272        mutex.acquire()273        try:274            return handle()275        finally:276            mutex.release()277    return processor_func278web.config.debug = False279app = web.application(urls, globals())280app.add_processor(mutex_processor())281if __name__ == "__main__":282    app.run()283else:...

aws_logs_info.py

Source:aws_logs_info.py

...166                    logGroupName=module.params['name'],167                    status=module.params['query_status']168                ), True169            else:170                return client.describe_queries(171                    logGroupName=module.params['name'],172                    status=module.params['query_status']173                ), False174        else:175            return None, False176    except (BotoCoreError, ClientError) as e:177        module.fail_json_aws(e, msg='Failed to fetch Amazon CloudWatch Logs details')178def main():179    argument_spec = dict(180        name=dict(required=False, aliases=['log_stream_name']),181        task_status=dict(182            required=False,183            choices=['CANCELLED', 'COMPLETED', 'FAILED', 'PENDING', 'PENDING_CANCEL', 'RUNNING'],184            default='COMPLETED'...

cloudwatch.py

Source:cloudwatch.py

1# Copyright (c) 2014 Scopely, Inc.2# Copyright (c) 2015 Mitch Garnaat3#4# Licensed under the Apache License, Version 2.0 (the "License"). You5# may not use this file except in compliance with the License. A copy of6# the License is located at7#8# http://aws.amazon.com/apache2.0/9#10# or in the "license" file accompanying this file. This file is11# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF12# ANY KIND, either express or implied. See the License for the specific13# language governing permissions and limitations under the License.14import jmespath15import logging16from skew.resources.aws import AWSResource17LOG = logging.getLogger(__name__)18class Alarm(AWSResource):19    class Meta(object):20        service = 'cloudwatch'21        type = 'alarm'22        enum_spec = ('describe_alarms', 'MetricAlarms', None)23        id = 'AlarmName'24        filter_name = 'AlarmNames'25        filter_type = None26        detail_spec = None27        name = 'AlarmName'28        date = 'AlarmConfigurationUpdatedTimestamp'29        dimension = None30        tags_spec = ('list_tags_for_resource', 'Tags[]', 'ResourceARN', 'arn')31    @property32    def arn(self):33        return 'arn:aws:%s:%s:%s:%s:%s' % (34            self._client.service_name,35            self._client.region_name,36            self._client.account_id,37            self.resourcetype, self.id)38class LogGroup(AWSResource):39    class Meta(object):40        service = 'logs'41        type = 'log-group'42        enum_spec = ('describe_log_groups', 'logGroups[]', None)43        attr_spec = [44            ('describe_log_streams', 'logGroupName',45             'logStreams', 'logStreams'),46            ('describe_metric_filters', 'logGroupName',47             'metricFilters', 'metricFilters'),48            ('describe_subscription_filters', 'logGroupName',49             'subscriptionFilters', 'subscriptionFilters'),50            ('describe_queries', 'logGroupName',51             'queries', 'queries'),52        ]53        detail_spec = None54        id = 'logGroupName'55        tags_spec = ('list_tags_log_group', 'tags',56                     'logGroupName', 'id')57        filter_name = 'logGroupNamePrefix'58        filter_type = 'dict'59        name = 'logGroupName'60        date = 'creationTime'61        dimension = 'logGroupName'62    def __init__(self, client, data, query=None):63        super(LogGroup, self).__init__(client, data, query)64        self._data = data65        self._keys = []66        self._id = data['logGroupName']67        # add addition attribute data68        for attr in self.Meta.attr_spec:69            LOG.debug(attr)70            detail_op, param_name, detail_path, detail_key = attr71            params = {param_name: self._id}72            data = self._client.call(detail_op, **params)73            if not (detail_path is None):74                data = jmespath.search(detail_path, data)75            if 'ResponseMetadata' in data:76                del data['ResponseMetadata']77            self.data[detail_key] = data78            LOG.debug(data)79    @property80    def logGroupName(self):81        return self.data.get('logGroupName')82    @property83    def arn(self):84        return 'arn:aws:%s:%s:%s:%s:%s' % (85            self._client.service_name,86            self._client.region_name,...

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.