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

How to use is_queued method in lisa

Best Python code snippet using lisa_python

DataLayerOperator.py

Source:DataLayerOperator.py

1#   Copyright 2020 The KNIX Authors2#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 DataLayerClient import DataLayerClient15class DataLayerOperator:16    def __init__(self, suid, sid, wid, datalayer):17        self._storage_userid = suid18        self._sandboxid = sid19        self._workflowid = wid20        self._datalayer = datalayer21        # global data layer clients for either workflow-private data or user storage22        self._data_layer_client = None23        self._data_layer_client_private = None24        # TODO (?): use the local data layer for operations regarding KV, maps, sets and counters instead of in-memory data structures (e.g., transient_data_output)25        # and store the operations/data for is_queued = True operations,26        # so that we can synchronize it with the global data layer27        # (key, value) store28        self.transient_data_output = {}29        self.transient_data_output_private = {}30        self.data_to_be_deleted = {}31        self.data_to_be_deleted_private = {}32        self.map_output = {}33        self.set_output = {}34        self.counter_output = {}35        self.map_output_delete = {}36        self.set_output_delete = {}37        self.counter_output_delete = {}38    # TODO: update to use local data layer for (key, value) operations39    def put(self, key, value, is_private=False, is_queued=False, table=None):40        if is_queued:41            if is_private:42                self.transient_data_output_private[key] = value43                if key in self.data_to_be_deleted_private:44                    self.data_to_be_deleted_private.pop(key, None)45            else:46                self.transient_data_output[key] = value47                if key in self.data_to_be_deleted:48                    self.data_to_be_deleted.pop(key, None)49        else:50            data_layer_client = self._get_data_layer_client(is_private)51            data_layer_client.put(key, value, tableName=table)52    def get(self, key, is_private=False, table=None):53        # check first transient_output54        # if not there, return the actual (global) data layer data item55        # if not there either, return empty string (as defined in the DataLayerClient)56        value = None57        # if the put() or delete() were called with is_queued=False (default),58        # then the below checks will still result in 'value is None'59        # if not, then value will be obtained from the transient output60        if is_private:61            if key in self.data_to_be_deleted_private:62                return ""63            value = self.transient_data_output_private.get(key)64        else:65            if key in self.data_to_be_deleted:66                return ""67            value = self.transient_data_output.get(key)68        if value is None:69            data_layer_client = self._get_data_layer_client(is_private)70            value = data_layer_client.get(key, tableName=table)71        return value72    def delete(self, key, is_private=False, is_queued=False, table=None):73        if is_queued:74            if is_private:75                self.transient_data_output_private.pop(key, None)76                self.data_to_be_deleted_private[key] = True77            else:78                self.transient_data_output.pop(key, None)79                self.data_to_be_deleted[key] = True80        else:81            data_layer_client = self._get_data_layer_client(is_private)82            data_layer_client.delete(key, tableName=table)83    def getKeys(self, start_index, end_index, is_private=False):84        keys = set()85        # XXX: should follow "read your writes"86        # the final result should include:87        # 1. all created locally88        # 2. all existing globally minus the ones deleted locally89        # TODO: 1. check local data layer first: get locally created and deleted90        # 2. retrieve all existing globally91        dlc = self._get_data_layer_client(is_private)92        m2 = dlc.listKeys(start_index, end_index)93        if m2 is not None:94            # TODO: 3. remove the ones deleted locally95            keys = keys.union(m2)96        return list(keys)97    # map operations98    def createMap(self, mapname, is_private=False, is_queued=False):99        if is_queued:100            # TODO: use transient data structure in memory when the operation is queued101            pass102        else:103            dlc = self._get_data_layer_client(is_private)104            dlc.createMap(mapname)105    def putMapEntry(self, mapname, key, value, is_private=False, is_queued=False):106        if is_queued:107            # TODO: use transient data structure in memory when the operation is queued108            pass109        else:110            dlc = self._get_data_layer_client(is_private)111            dlc.putMapEntry(mapname, key, value)112    def getMapEntry(self, mapname, key, is_private=False):113        value = None114        # TODO: check transient data structure first115        if value is None:116            dlc = self._get_data_layer_client(is_private)117            value = dlc.getMapEntry(mapname, key)118        return value119    def deleteMapEntry(self, mapname, key, is_private=False, is_queued=False):120        if is_queued:121            # TODO: use transient data structure in memory when the operation is queued122            pass123        else:124            dlc = self._get_data_layer_client(is_private)125            dlc.deleteMapEntry(mapname, key)126    def containsMapKey(self, mapname, key, is_private=False):127        ret = False128        # TODO: check transient data structure first129        if not ret:130            dlc = self._get_data_layer_client(is_private)131            ret = dlc.containsMapKey(mapname, key)132        return ret133    def retrieveMap(self, mapname, is_private=False):134        retmap = {}135        # XXX: should follow "read your writes"136        # the final result should include:137        # 1. all created locally138        # 2. all existing globally minus the ones deleted locally139        # TODO: 1. check local data layer first: get locally created and deleted140        # 2. retrieve all existing globally141        dlc = self._get_data_layer_client(is_private)142        retmap2 = dlc.retrieveMap(mapname)143        if retmap2 is not None:144            for k in retmap2:145                retmap[k] = retmap2[k]146            # TODO: 3. remove the ones deleted locally147        return retmap148    def getMapKeys(self, mapname, is_private=False):149        keys = set()150        # XXX: should follow "read your writes"151        # the final result should include:152        # 1. all created locally153        # 2. all existing globally minus the ones deleted locally154        # TODO: 1. check local data layer first: get locally created and deleted155        # 2. retrieve all existing globally156        dlc = self._get_data_layer_client(is_private)157        k2 = dlc.getMapKeys(mapname)158        if k2 is not None:159            # TODO: 3. remove the ones deleted locally160            keys = keys.union(k2)161        return keys162    def clearMap(self, mapname, is_private=False, is_queued=False):163        if is_queued:164            # TODO: use transient data structure in memory when the operation is queued165            pass166        else:167            dlc = self._get_data_layer_client(is_private)168            dlc.clearMap(mapname)169    def deleteMap(self, mapname, is_private=False, is_queued=False):170        if is_queued:171            # TODO: use transient data structure in memory when the operation is queued172            pass173        else:174            dlc = self._get_data_layer_client(is_private)175            dlc.deleteMap(mapname)176    def getMapNames(self, start_index=0, end_index=2147483647, is_private=False):177        maps = set()178        # XXX: should follow "read your writes"179        # the final result should include:180        # 1. all created locally181        # 2. all existing globally minus the ones deleted locally182        # TODO: 1. check local data layer first: get locally created and deleted183        # 2. retrieve all existing globally184        dlc = self._get_data_layer_client(is_private)185        m2 = dlc.getMapNames(start_index, end_index)186        if m2 is not None:187            # TODO: 3. remove the ones deleted locally188            maps = maps.union(m2)189        return list(maps)190    # set operations191    def createSet(self, setname, is_private=False, is_queued=False):192        if is_queued:193            # TODO: use transient data structure in memory when the operation is queued194            pass195        else:196            dlc = self._get_data_layer_client(is_private)197            dlc.createSet(setname)198    def addSetEntry(self, setname, item, is_private=False, is_queued=False):199        if is_queued:200            # TODO: use transient data structure in memory when the operation is queued201            pass202        else:203            dlc = self._get_data_layer_client(is_private)204            dlc.addSetEntry(setname, item)205    def removeSetEntry(self, setname, item, is_private=False, is_queued=False):206        if is_queued:207            # TODO: use transient data structure in memory when the operation is queued208            pass209        else:210            dlc = self._get_data_layer_client(is_private)211            dlc.removeSetEntry(setname, item)212    def containsSetItem(self, setname, item, is_private=False):213        ret = False214        # TODO: check transient data structure first215        if not ret:216            dlc = self._get_data_layer_client(is_private)217            ret = dlc.containsSetItem(setname, item)218        return ret219    def retrieveSet(self, setname, is_private=False):220        items = set()221        # XXX: should follow "read your writes"222        # the final result should include:223        # 1. all created locally224        # 2. all existing globally minus the ones deleted locally225        # TODO: 1. check local data layer first: get locally created and deleted226        # 2. retrieve all existing globally227        dlc = self._get_data_layer_client(is_private)228        i2 = dlc.retrieveSet(setname)229        if i2 is not None:230            # TODO: 3. remove the ones deleted locally231            items = items.union(i2)232        return items233    def clearSet(self, setname, is_private=False, is_queued=False):234        if is_queued:235            # TODO: use transient data structure in memory when the operation is queued236            pass237        else:238            dlc = self._get_data_layer_client(is_private)239            dlc.clearSet(setname)240    def deleteSet(self, setname, is_private=False, is_queued=False):241        if is_queued:242            # TODO: use transient data structure in memory when the operation is queued243            pass244        else:245            dlc = self._get_data_layer_client(is_private)246            dlc.deleteSet(setname)247    def getSetNames(self, start_index=0, end_index=2147483647, is_private=False):248        sets = set()249        # XXX: should follow "read your writes"250        # the final result should include:251        # 1. all created locally252        # 2. all existing globally minus the ones deleted locally253        # TODO: 1. check local data layer first: get locally created and deleted254        # 2. retrieve all existing globally255        dlc = self._get_data_layer_client(is_private)256        s2 = dlc.getSetNames(start_index, end_index)257        if s2 is not None:258            # TODO: 3. remove the ones deleted locally259            sets = sets.union(s2)260        return list(sets)261    # counter operations262    def createCounter(self, countername, count, is_private=False, is_queued=False):263        if is_queued:264            # TODO: use transient data structure in memory when the operation is queued265            pass266        else:267            dlc = self._get_data_layer_client(is_private)268            dlc.createCounter(countername, count)269    def getCounterValue(self, countername, is_private=False):270        value = 0271        # TODO: check transient data structure first and apply any changes to the global value272        dlc = self._get_data_layer_client(is_private)273        value = dlc.getCounter(countername)274        return value275    def incrementCounter(self, countername, increment, is_private=False, is_queued=False):276        if is_queued:277            # TODO: use transient data structure in memory when the operation is queued278            pass279        else:280            dlc = self._get_data_layer_client(is_private)281            dlc.incrementCounter(countername, increment)282    def decrementCounter(self, countername, decrement, is_private=False, is_queued=False):283        if is_queued:284            # TODO: use transient data structure in memory when the operation is queued285            pass286        else:287            dlc = self._get_data_layer_client(is_private)288            dlc.decrementCounter(countername, decrement)289    def deleteCounter(self, countername, is_private=False, is_queued=False):290        if is_queued:291            # TODO: use transient data structure in memory when the operation is queued292            pass293        else:294            dlc = self._get_data_layer_client(is_private)295            dlc.deleteCounter(countername)296    def getCounterNames(self, start_index=0, end_index=2147483647, is_private=False):297        counters = set()298        # XXX: should follow "read your writes"299        # the final result should include:300        # 1. all created locally301        # 2. all existing globally minus the ones deleted locally302        # TODO: 1. check local data layer first: get locally created and deleted303        # 2. retrieve all existing globally304        dlc = self._get_data_layer_client(is_private)305        c2 = dlc.getCounterNames(start_index, end_index)306        if c2 is not None:307            # TODO: 3. remove the ones deleted locally308            counters = counters.union(c2)309        return list(counters)310    def get_transient_data_output(self, is_private=False):311        '''312        Return the transient data, so that it can be committed to the data layer313        when the function instance finishes.314        '''315        if is_private:316            return self.transient_data_output_private317        return self.transient_data_output318    def get_data_to_be_deleted(self, is_private=False):319        '''320        Return the list of deleted data items, so that they can be committed to the data layer321        when the function instance finishes.322        '''323        if is_private:324            return self.data_to_be_deleted_private325        return self.data_to_be_deleted326    def _get_data_layer_client(self, is_private=False):327        '''328        Return the data layer client, so that it can be used to commit to the data layer329        when the function instance finishes.330        If it is not initialized yet, it will be initialized here.331        '''332        # TODO: need also the locality information333        if is_private:334            if self._data_layer_client_private is None:335                self._data_layer_client_private = DataLayerClient(locality=1, sid=self._sandboxid, wid=self._workflowid, is_wf_private=True, connect=self._datalayer)336            return self._data_layer_client_private337        if self._data_layer_client is None:338            self._data_layer_client = DataLayerClient(locality=1, suid=self._storage_userid, is_wf_private=False, connect=self._datalayer)339        return self._data_layer_client340    def _shutdown_data_layer_client(self):341        '''342        Shut down the data layer client if it has been initialized343        after the function instance finishes committing changes344        to the data layer.345        '''346        if self._data_layer_client_private is not None:347            self._data_layer_client_private.shutdown()348            self._data_layer_client_private = None349        if self._data_layer_client is not None:350            self._data_layer_client.shutdown()...

command.py

Source:command.py

...23        """24        Args:25            id_(int): dobot protocol command id26            rw(int): 0 or 1. dobot protocol rw27            is_queued(int): 0 or 1. dobot protocol is_queued28            param_format(unicode): param binary format29            param_names(list of unicode): param names30        """31        self.builder = PacketBuilder(id_, rw, is_queued)32        self.parser = None33        self.params = None34        if param_format:35            self.parser = PacketParser(param_format, param_names)36    def build_packet(self):37        return self.builder.build(self.params)38    def parse_packet(self, return_packet):39        if self.parser is None:40            return None41        if self.parser.param_names is None:...

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.