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

How to use process_result method in localstack

Best Python code snippet using localstack_python

agent.py

Source:agent.py

1import random2import click3import add4import auth5import delete6import network7import query_processing as qp8import search9import update10# the pair of user credentials11credentials = "default", "default"12def print_msg(msg):13    """Echo a message with the prefix "Sammy>"14    15    :param msg: A string, the message to echo16    """17    click.echo("Sammy> " + msg)18def print_failure():19    """Print out a failure response chosen at random"""20    failure_responses = [21        "I'm sorry. I'm not sure what you mean.",22        "I didn't quite catch that.",23        "Hmm, I'm don't know what that means.",24        "I'm not entirely sure I know what it is you want.",25        "I don't quite understand what you mean.",26        "I'm sure not sure I understand."27    ]28    print_msg(random.choice(failure_responses))29def print_connection_error_msg():30    """Print out a message corresponding to a requests connection failure"""31    print_msg("Oh no, there was an error connecting to MAL. Please check your internet connection")32def authorise_user():33    """Get a pair of credentials from the user34    :return: True is successfully authenticated, False if their was an error and user quit35    """36    global credentials37    while True:38        # get the pair of credentials from the user (username, password)39        credentials = auth.get_user_credentials("Please enter your username", "And now your password")40        # check that the credentials are valid41        result = auth.validate_credentials(credentials)42        if result is not network.StatusCode.SUCCESS:43            if result == network.StatusCode.CONNECTION_ERROR:44                print_connection_error_msg()45            elif result == network.StatusCode.UNAUTHORISED:46                print_msg("Something was wrong with the username or password :(")47            elif result == network.StatusCode.OTHER_ERROR:48                print_msg("Some kind of error has occurred :'(")49            if click.confirm("Sammy> Do you want to try again?"):50                continue51            else:52                return False53        return True54def welcome():55    """Print out the welcome message and bootstrap program functionality56    57    :return: True if the user authenticated successfully, False otherwise58    """59    click.clear()60    click.echo("====== MAL Natural Language Interface ======")61    click.echo()62    print_msg("Hello! My name is Sammy and I am your MyAnimeList digital assistant.")63    print_msg("Before we get started, I need you to confirm your MAL account details.")64    click.echo()65    # authenticate the user, return True if successful, else False66    if authorise_user():67        click.echo()68        print_msg("Yay, everything checked out! Let's get started.")69        print_msg("What can I do for you today?")70        return True71    else:72        print_msg("Bye bye!")73        return False74def get_query():75    """Get the query from the user and process it"""76    # keep prompting for a query until the user quits77    while True:78        click.echo()79        query = click.prompt(credentials[0], prompt_suffix="> ")80        click.echo()81        # process the user query82        processed = process_query(query)83        # quit the program if the user decided84        if processed == qp.Extras.EXIT:85            return86def process_query(query):87    """Process the user query and carry out the requested action88    :param query: A string, the raw user query89    """90    # process the query and get a dictionary with the result91    process_result = qp.process(query)92    # print out the dictionary for debug purposes (REMOVE FOR PRODUCTION!)93    # print_msg(str(process_result))94    # the user wants to exit95    if process_result == qp.Extras.EXIT:96        print_msg("Bye bye!")97        return process_result98    # the user said hello99    if process_result["extra"] == qp.Extras.GREETING:100        greetings = ["Hi", "Hello", "Yo"]101        print_msg("{}, {}!".format(random.choice(greetings), credentials[0]))102    # the user said thanks103    elif process_result["extra"] == qp.Extras.THANKS:104        thanks = ["No problem", "You're welcome", "Any time", "You are very welcome"]105        print_msg("{} :D".format(random.choice(thanks)))106    # search database queries107    if process_result["operation"] == qp.OperationType.SEARCH:108        # search for an anime109        if process_result["type"] == qp.MediaType.ANIME:110            search.search(credentials, "anime", process_result["term"])111        # search for a manga112        elif process_result["type"] == qp.MediaType.MANGA:113            search.search(credentials, "manga", process_result["term"])114    # update list entry details queries115    elif process_result["operation"] == qp.OperationType.UPDATE:116        # update anime117        if process_result["type"] == qp.MediaType.ANIME:118            # update anime status119            if process_result["modifier"] == qp.UpdateModifier.STATUS:120                if process_result["value"] == qp.StatusType.WATCHING:121                    update.update_anime_list_entry(credentials, "status", process_result["term"], 1)122                elif process_result["value"] == qp.StatusType.COMPLETED:123                    update.update_anime_list_entry(credentials, "status", process_result["term"], 2)124                elif process_result["value"] == qp.StatusType.ON_HOLD:125                    update.update_anime_list_entry(credentials, "status", process_result["term"], 3)126                elif process_result["value"] == qp.StatusType.DROPPED:127                    update.update_anime_list_entry(credentials, "status", process_result["term"], 4)128                elif process_result["value"] == qp.StatusType.PLAN_TO_WATCH:129                    update.update_anime_list_entry(credentials, "status", process_result["term"], 6)130            # update anime score131            elif process_result["modifier"] == qp.UpdateModifier.SCORE:132                update.update_anime_list_entry(credentials, "score", process_result["term"], process_result["value"])133            # update anime episode count134            elif process_result["modifier"] == qp.UpdateModifier.EPISODE:135                update.update_anime_list_entry(credentials, "episode", process_result["term"], process_result["value"])136        # update manga137        elif process_result["type"] == qp.MediaType.MANGA:138            # update manga status139            if process_result["modifier"] == qp.UpdateModifier.STATUS:140                if process_result["value"] == qp.StatusType.READING:141                    update.update_manga_list_entry(credentials, "status", process_result["term"], 1)142                elif process_result["value"] == qp.StatusType.COMPLETED:143                    update.update_manga_list_entry(credentials, "status", process_result["term"], 2)144                elif process_result["value"] == qp.StatusType.ON_HOLD:145                    update.update_manga_list_entry(credentials, "status", process_result["term"], 3)146                elif process_result["value"] == qp.StatusType.DROPPED:147                    update.update_manga_list_entry(credentials, "status", process_result["term"], 4)148                elif process_result["value"] == qp.StatusType.PLAN_TO_READ:149                    update.update_manga_list_entry(credentials, "status", process_result["term"], 6)150            # update manga score151            elif process_result["modifier"] == qp.UpdateModifier.SCORE:152                update.update_manga_list_entry(credentials, "score", process_result["term"], process_result["value"])153            # update manga chapter count154            elif process_result["modifier"] == qp.UpdateModifier.CHAPTER:155                update.update_manga_list_entry(credentials, "chapter", process_result["term"], process_result["value"])156            # update manga volume count157            elif process_result["modifier"] == qp.UpdateModifier.VOLUME:158                update.update_manga_list_entry(credentials, "volume", process_result["term"], process_result["value"])159    # increment counts for list entries160    elif process_result["operation"] == qp.OperationType.UPDATE_INCREMENT:161        # increment episode count for anime162        if process_result["type"] == qp.MediaType.ANIME:163            update.update_anime_list_entry(credentials, "episode", process_result["term"])164        # increment manga counts165        elif process_result["type"] == qp.MediaType.MANGA:166            # increment chapter count for manga167            if process_result["modifier"] == qp.UpdateModifier.CHAPTER:168                update.update_manga_list_entry(credentials, "chapter", process_result["term"])169            # increment volume count for manga170            elif process_result["modifier"] == qp.UpdateModifier.VOLUME:171                update.update_manga_list_entry(credentials, "volume", process_result["term"])172    # add new entry queries173    elif process_result["operation"] == qp.OperationType.ADD:174        # add new anime entry175        if process_result["type"] == qp.MediaType.ANIME:176            add.add_entry(credentials, "anime", process_result["term"])177        # add new manga entry178        elif process_result["type"] == qp.MediaType.MANGA:179            add.add_entry(credentials, "manga", process_result["term"])180    # delete list entry queries181    elif process_result["operation"] == qp.OperationType.DELETE:182        # delete anime entry183        if process_result["type"] == qp.MediaType.ANIME:184            delete.delete_entry(credentials, "anime", process_result["term"])185        # delete manga entry186        elif process_result["type"] == qp.MediaType.MANGA:187            delete.delete_entry(credentials, "manga", process_result["term"])188    # view all list entries queries189    elif process_result["operation"] == qp.OperationType.VIEW_LIST:190        # view anime list191        if process_result["type"] == qp.MediaType.ANIME:192            update.view_list(credentials[0], "anime")193        # view manga list194        elif process_result["type"] == qp.MediaType.MANGA:195            update.view_list(credentials[0], "manga")196    # default response if the system failed to understand the query197    elif process_result["extra"] is None:...

gateway.py

Source:gateway.py

...19        """20        Generates the PRE_SHARED_KEY from the gateway.21        Returns a Command.22        """23        def process_result(result):24            return result[ATTR_PSK]25        return Command('post', [ROOT_GATEWAY, ATTR_AUTH], {26            ATTR_IDENTITY: identity27        }, process_result=process_result)28    def get_endpoints(self):29        """30        Return all available endpoints on the gateway.31        Returns a Command.32        """33        def process_result(result):34            return [line.split(';')[0][2:-1] for line in result.split(',')]35        return Command('get', ['.well-known', 'core'], parse_json=False,36                       process_result=process_result)37    def get_devices(self):38        """39        Return the devices linked to the gateway.40        Returns a Command.41        """42        def process_result(result):43            return [self.get_device(dev) for dev in result]44        return Command('get', [ROOT_DEVICES], process_result=process_result)45    def get_device(self, device_id):46        """47        Return specified device.48        Returns a Command.49        """50        def process_result(result):51            return Device(result)52        return Command('get', [ROOT_DEVICES, device_id],53                       process_result=process_result)54    def get_groups(self):55        """56        Return the groups linked to the gateway.57        Returns a Command.58        """59        def process_result(result):60            return [self.get_group(group) for group in result]61        return Command('get', [ROOT_GROUPS], process_result=process_result)62    def get_group(self, group_id):63        """64        Return specified group.65        Returns a Command.66        """67        def process_result(result):68            return Group(self, result)69        return Command('get', [ROOT_GROUPS, group_id],70                       process_result=process_result)71    def get_gateway_info(self):72        """73        Return the gateway info.74        Returns a Command.75        """76        def process_result(result):77            return GatewayInfo(result)78        return Command('get',79                       [ROOT_GATEWAY, ATTR_GATEWAY_INFO],80                       process_result=process_result)81    def get_moods(self):82        """83        Return moods defined on the gateway.84        Returns a Command.85        """86        mood_parent = self._get_mood_parent()87        def process_result(result):88            return [self.get_mood(mood, mood_parent=mood_parent) for mood in89                    result]90        return Command('get', [ROOT_MOODS, mood_parent],91                       process_result=process_result)92    def get_mood(self, mood_id, *, mood_parent=None):93        """94        Return a mood.95        Returns a Command.96        """97        if mood_parent is None:98            mood_parent = self._get_mood_parent()99        def process_result(result):100            return Mood(result, mood_parent)101        return Command('get', [ROOT_MOODS, mood_parent, mood_id],102                       mood_parent, process_result=process_result)103    def _get_mood_parent(self):104        """105        Get the parent of all moods.106        Returns a Command.107        """108        def process_result(result):109            return result[0]110        return Command('get', [ROOT_MOODS], process_result=process_result)111    def get_smart_tasks(self):112        """113        Return the transitions linked to the gateway.114        Returns a Command.115        """116        def process_result(result):117            return [self.get_smart_task(task) for task in result]118        return Command('get', [ROOT_SMART_TASKS],119                       process_result=process_result)120    def get_smart_task(self, task_id):121        """122        Return specified transition.123        Returns a Command.124        """125        def process_result(result):126            return SmartTask(self, result)127        return Command('get', [ROOT_SMART_TASKS, task_id],128                       process_result=process_result)129    def reboot(self):130        """131        Reboot the Gateway132        Returns a Command.133        """134        return Command('post',135                       [ROOT_GATEWAY, ATTR_GATEWAY_REBOOT])136    def set_commissioning_timeout(self, timeout):137        """Put the gateway in a state in which it accepts pairings from138        switches, dimmers and motion sensors for up to timeout seconds.139        Returns a Command."""140        return Command('put',141                       [ROOT_GATEWAY, ATTR_GATEWAY_INFO],142                       {ATTR_COMMISSIONING_MODE: timeout})143    def factory_reset(self):144        """145        Resets the Gateway to factory defaults.146        WARNING: All data in Gateway is lost (pairing, groups, etc)147        Returns a Command.148        """149        return Command('post',150                       [ROOT_GATEWAY, ATTR_GATEWAY_FACTORY_DEFAULTS])151class GatewayInfo:152    """This class contains Gateway information."""153    def __init__(self, raw):154        self.raw = raw155    @property156    def id(self):157        """This looks like a value representing an id."""158        return self.raw.get(ATTR_GATEWAY_ID)159    @property160    def ntp_server(self):161        """NTP server in use."""162        return self.raw.get(ATTR_NTP)163    @property164    def firmware_version(self):165        """NTP server in use."""166        return self.raw.get(ATTR_FIRMWARE_VERSION)167    @property168    def current_time(self):169        if ATTR_CURRENT_TIME_UNIX not in self.raw:170            return None171        return datetime.utcfromtimestamp(self.raw[ATTR_CURRENT_TIME_UNIX])172    @property173    def current_time_iso8601(self):174        return self.raw.get(ATTR_CURRENT_TIME_ISO8601)175    @property176    def first_setup(self):177        """This is a guess of the meaning of this value."""178        if ATTR_FIRST_SETUP not in self.raw:179            return None180        return datetime.utcfromtimestamp(self.raw[ATTR_FIRST_SETUP])181    @property182    def homekit_id(self):183        return self.raw.get(ATTR_HOMEKIT_ID)184    @property185    def path(self):186        return [ROOT_GATEWAY, ATTR_GATEWAY_INFO]187    def set_values(self, values):188        """189        Helper to set values for mood.190        Returns a Command.191        """192        return Command('put', self.path, values)193    def update(self):194        """195        Update the info.196        Returns a Command.197        """198        def process_result(result):199            self.raw = result200        return Command('get', self.path, process_result=process_result)201    def __repr__(self):...

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.