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

How to use mock_rpcs method in autotest

Best Python code snippet using autotest_python

action_common_unittest.py

Source:action_common_unittest.py

...76                   u'invalid': 0,77                   u'kernel_config': u''}]78        mytest = action_common.atest_list()79        mytest.afe = rpc.afe_comm()80        self.mock_rpcs([('get_labels',81                         filters,82                         True,83                         values)])84        self.god.mock_io()85        self.assertEqual(values,86                         mytest.execute(op='get_labels',87                                        filters=filters,88                                        check_results=check_results))89        (out, err) = self.god.unmock_io()90        self.god.check_playback()91        return (out, err)92    def test_atest_list_execute_no_filters(self):93        self._atest_list_execute()94    def test_atest_list_execute_filters_all_good(self):95        filters = {}96        check_results = {}97        filters['name__in'] = ['label0', 'label1']98        check_results['name__in'] = 'name'99        (out, err) = self._atest_list_execute(filters, check_results)100        self.assertEqual(err, '')101    def test_atest_list_execute_filters_good_and_bad(self):102        filters = {}103        check_results = {}104        filters['name__in'] = ['label0', 'label1', 'label2']105        check_results['name__in'] = 'name'106        (out, err) = self._atest_list_execute(filters, check_results)107        self.assertWords(err, ['Unknown', 'label2'])108    def test_atest_list_execute_items_good_and_bad_no_check(self):109        filters = {}110        check_results = {}111        filters['name__in'] = ['label0', 'label1', 'label2']112        check_results['name__in'] = None113        (out, err) = self._atest_list_execute(filters, check_results)114        self.assertEqual(err, '')115    def test_atest_list_execute_filters_wildcard(self):116        filters = {}117        check_results = {}118        filters['name__in'] = ['label*']119        check_results['name__in'] = 'name'120        values = [{u'id': 180,121                   u'platform': False,122                   u'name': u'label0',123                   u'invalid': False,124                   u'kernel_config': u''},125                  {u'id': 338,126                   u'platform': False,127                   u'name': u'label1',128                   u'invalid': False,129                   u'kernel_config': u''}]130        mytest = action_common.atest_list()131        mytest.afe = rpc.afe_comm()132        self.mock_rpcs([('get_labels', {'name__startswith': 'label'},133                         True, values)])134        self.god.mock_io()135        self.assertEqual(values,136                         mytest.execute(op='get_labels',137                                        filters=filters,138                                        check_results=check_results))139        (out, err) = self.god.unmock_io()140        self.god.check_playback()141        self.assertEqual(err, '')142#143# Creation & Deletion of a topic (ACL, label, user)144#145class atest_create_or_delete_unittest(cli_mock.cli_unittest):146    def _create_cr_del(self, items):147        def _items():148            return items149        crdel = action_common.atest_create_or_delete()150        crdel.afe = rpc.afe_comm()151        crdel.topic =  crdel.usage_topic = 'label'152        crdel.op_action = 'add'153        crdel.get_items = _items154        crdel.data['platform'] = False155        crdel.data_item_key = 'name'156        crdel.no_confirmation = True157        return crdel158    def test_execute_create_one_topic(self):159        acr = self._create_cr_del(['label0'])160        self.mock_rpcs([('add_label',161                         {'name': 'label0', 'platform': False},162                         True, 42)])163        ret = acr.execute()164        self.god.check_playback()165        self.assert_(['label0'], ret)166    def test_execute_create_two_topics(self):167        acr = self._create_cr_del(['label0', 'label1'])168        self.mock_rpcs([('add_label',169                         {'name': 'label0', 'platform': False},170                         True, 42),171                        ('add_label',172                         {'name': 'label1', 'platform': False},173                         True, 43)])174        ret = acr.execute()175        self.god.check_playback()176        self.assertEqualNoOrder(['label0', 'label1'], ret)177    def test_execute_create_error(self):178        acr = self._create_cr_del(['label0'])179        self.mock_rpcs([('add_label',180                         {'name': 'label0', 'platform': False},181                         False,182                         '''ValidationError:183                         {'name': 'This value must be unique (label0)'}''')])184        ret = acr.execute()185        self.god.check_playback()186        self.assertEqualNoOrder([], ret)187#188# Adding or Removing users or hosts from a topic(ACL or label)189#190class atest_add_or_remove_unittest(cli_mock.cli_unittest):191    def _create_add_remove(self, items, users=None, hosts=None):192        def _items():193            return [items]194        addrm = action_common.atest_add_or_remove()195        addrm.afe = rpc.afe_comm()196        if users:197            addrm.users = users198        if hosts:199            addrm.hosts = hosts200        addrm.topic = 'acl_group'201        addrm.msg_topic = 'ACL'202        addrm.op_action = 'add'203        addrm.msg_done = 'Added to'204        addrm.get_items = _items205        return addrm206    def test__add_remove_uh_to_topic(self):207        acl_addrm = self._create_add_remove('acl0',208                                        users=['user0', 'user1'])209        self.mock_rpcs([('acl_group_add_users',210                         {'id': 'acl0',211                          'users': ['user0', 'user1']},212                         True,213                         None)])214        acl_addrm._add_remove_uh_to_topic('acl0', 'users')215        self.god.check_playback()216    def test__add_remove_uh_to_topic_raise(self):217        acl_addrm = self._create_add_remove('acl0',218                                        users=['user0', 'user1'])219        self.assertRaises(AttributeError,220                          acl_addrm._add_remove_uh_to_topic,221                          'acl0', 'hosts')222    def test_execute_add_or_remove_uh_to_topic_acl_users(self):223        acl_addrm = self._create_add_remove('acl0',224                                        users=['user0', 'user1'])225        self.mock_rpcs([('acl_group_add_users',226                         {'id': 'acl0',227                          'users': ['user0', 'user1']},228                         True,229                         None)])230        execute_result = acl_addrm.execute()231        self.god.check_playback()232        self.assertEqualNoOrder(['acl0'], execute_result['users'])233        self.assertEqual([], execute_result['hosts'])234    def test_execute_add_or_remove_uh_to_topic_acl_users_hosts(self):235        acl_addrm = self._create_add_remove('acl0',236                                            users=['user0', 'user1'],237                                            hosts=['host0', 'host1'])238        self.mock_rpcs([('acl_group_add_users',239                         {'id': 'acl0',240                          'users': ['user0', 'user1']},241                         True,242                         None),243                        ('acl_group_add_hosts',244                         {'id': 'acl0',245                          'hosts': ['host0', 'host1']},246                         True,247                         None)])248        execute_result = acl_addrm.execute()249        self.god.check_playback()250        self.assertEqualNoOrder(['acl0'], execute_result['users'])251        self.assertEqualNoOrder(['acl0'], execute_result['hosts'])252    def test_execute_add_or_remove_uh_to_topic_acl_bad_users(self):253        acl_addrm = self._create_add_remove('acl0',254                                            users=['user0', 'user1'])255        self.mock_rpcs([('acl_group_add_users',256                         {'id': 'acl0',257                          'users': ['user0', 'user1']},258                         False,259                         'DoesNotExist: The following users do not exist: '260                         'user0, user1')])261        execute_result = acl_addrm.execute()262        self.god.check_playback()263        self.assertEqual([], execute_result['users'])264        self.assertEqual([], execute_result['hosts'])265        self.assertOutput(acl_addrm, execute_result,266                          err_words_ok=['DoesNotExist',267                                        'acl_group_add_users',268                                        'user0', 'user1'],269                          err_words_no = ['acl_group_add_hosts'])270    def test_execute_add_or_remove_uh_to_topic_acl_bad_users_partial(self):271        acl_addrm = self._create_add_remove('acl0',272                                            users=['user0', 'user1'])273        self.mock_rpcs([('acl_group_add_users',274                         {'id': 'acl0',275                          'users': ['user0', 'user1']},276                         False,277                         'DoesNotExist: The following users do not exist: '278                         'user0'),279                        ('acl_group_add_users',280                         {'id': 'acl0',281                          'users': ['user1']},282                         True,283                         None)])284        execute_result = acl_addrm.execute()285        self.god.check_playback()286        self.assertEqual(['acl0'], execute_result['users'])287        self.assertEqual([], execute_result['hosts'])288        self.assertOutput(acl_addrm, execute_result,289                          out_words_ok=["Added to ACL 'acl0'", 'user1'],290                          err_words_ok=['DoesNotExist',291                                        'acl_group_add_users',292                                        'user0'],293                          err_words_no = ['acl_group_add_hosts'])294    def test_execute_add_or_remove_uh_to_topic_acl_bad_u_partial_kill(self):295        acl_addrm = self._create_add_remove('acl0',296                                            users=['user0', 'user1'])297        acl_addrm.kill_on_failure = True298        self.mock_rpcs([('acl_group_add_users',299                         {'id': 'acl0',300                          'users': ['user0', 'user1']},301                         False,302                         'DoesNotExist: The following users do not exist: '303                         'user0')])304        sys.exit.expect_call(1).and_raises(cli_mock.ExitException)305        self.god.mock_io()306        self.assertRaises(cli_mock.ExitException, acl_addrm.execute)307        (out, err) = self.god.unmock_io()308        self.god.check_playback()309        self._check_output(out=out, err=err,310                          err_words_ok=['DoesNotExist',311                                        'acl_group_add_users',312                                        'user0'],313                          err_words_no = ['acl_group_add_hosts'])314    def test_execute_add_or_remove_uh_to_topic_acl_bad_users_good_hosts(self):315        acl_addrm = self._create_add_remove('acl0',316                                        users=['user0', 'user1'],317                                        hosts=['host0', 'host1'])318        self.mock_rpcs([('acl_group_add_users',319                         {'id': 'acl0',320                          'users': ['user0', 'user1']},321                         False,322                         'DoesNotExist: The following users do not exist: '323                         'user0, user1'),324                        ('acl_group_add_hosts',325                         {'id': 'acl0',326                          'hosts': ['host0', 'host1']},327                         True,328                         None)])329        execute_result = acl_addrm.execute()330        self.god.check_playback()331        self.assertEqual([], execute_result['users'])332        self.assertEqual(['acl0'], execute_result['hosts'])333        self.assertOutput(acl_addrm, execute_result,334                          out_words_ok=["Added to ACL 'acl0' hosts:",335                                        "host0", "host1"],336                          err_words_ok=['DoesNotExist',337                                        'acl_group_add_users',338                                        'user0', 'user1'],339                          err_words_no = ['acl_group_add_hosts'])340    def test_execute_add_or_remove_uh_to_topic_acl_good_users_bad_hosts(self):341        acl_addrm = self._create_add_remove('acl0 with space',342                                        users=['user0', 'user1'],343                                        hosts=['host0', 'host1'])344        self.mock_rpcs([('acl_group_add_users',345                         {'id': 'acl0 with space',346                          'users': ['user0', 'user1']},347                         True,348                         None),349                        ('acl_group_add_hosts',350                         {'id': 'acl0 with space',351                          'hosts': ['host0', 'host1']},352                         False,353                         'DoesNotExist: The following hosts do not exist: '354                         'host0, host1')])355        execute_result = acl_addrm.execute()356        self.god.check_playback()357        self.assertEqual(['acl0 with space'], execute_result['users'])358        self.assertEqual([], execute_result['hosts'])359        self.assertOutput(acl_addrm, execute_result,360                          out_words_ok=["Added to ACL 'acl0 with space' users:",361                                        "user0", "user1"],362                          err_words_ok=['DoesNotExist',363                                        'acl_group_add_hosts',364                                        'host0', 'host1'],365                          err_words_no = ['acl_group_add_users'])366    def test_exe_add_or_remove_uh_to_topic_acl_good_u_bad_hosts_partial(self):367        acl_addrm = self._create_add_remove('acl0',368                                        users=['user0', 'user1'],369                                        hosts=['host0', 'host1'])370        self.mock_rpcs([('acl_group_add_users',371                         {'id': 'acl0',372                          'users': ['user0', 'user1']},373                         True,374                         None),375                        ('acl_group_add_hosts',376                         {'id': 'acl0',377                          'hosts': ['host0', 'host1']},378                         False,379                         'DoesNotExist: The following hosts do not exist: '380                         'host1'),381                        ('acl_group_add_hosts',382                         {'id': 'acl0',383                          'hosts': ['host0']},384                         True,385                         None)])386        execute_result = acl_addrm.execute()387        self.god.check_playback()388        self.assertEqual(['acl0'], execute_result['users'])389        self.assertEqual(['acl0'], execute_result['hosts'])390        self.assertOutput(acl_addrm, execute_result,391                          out_words_ok=["Added to ACL 'acl0' users:",392                                        "user0", "user1", "host0"],393                          err_words_ok=['DoesNotExist',394                                        'acl_group_add_hosts',395                                        'host1'],396                          err_words_no = ['acl_group_add_users'])397    def test_execute_add_or_remove_uh_to_topic_acl_bad_users_bad_hosts(self):398        acl_addrm = self._create_add_remove('acl0',399                                        users=['user0', 'user1'],400                                        hosts=['host0', 'host1'])401        self.mock_rpcs([('acl_group_add_users',402                         {'id': 'acl0',403                          'users': ['user0', 'user1']},404                         False,405                         'DoesNotExist: The following users do not exist: '406                         'user0, user1'),407                        ('acl_group_add_hosts',408                         {'id': 'acl0',409                          'hosts': ['host0', 'host1']},410                         False,411                         'DoesNotExist: The following hosts do not exist: '412                         'host0, host1')])413        execute_result = acl_addrm.execute()414        self.god.check_playback()415        self.assertEqual([], execute_result['users'])416        self.assertEqual([], execute_result['hosts'])417        self.assertOutput(acl_addrm, execute_result,418                          err_words_ok=['DoesNotExist',419                                        'acl_group_add_hosts',420                                        'host0', 'host1',421                                        'acl_group_add_users',422                                        'user0', 'user1'])423    def test_execute_add_or_remove_uh_to_topic_acl_bad_u_bad_h_partial(self):424        acl_addrm = self._create_add_remove('acl0',425                                        users=['user0', 'user1'],426                                        hosts=['host0', 'host1'])427        self.mock_rpcs([('acl_group_add_users',428                         {'id': 'acl0',429                          'users': ['user0', 'user1']},430                         False,431                         'DoesNotExist: The following users do not exist: '432                         'user0'),433                        ('acl_group_add_users',434                         {'id': 'acl0',435                          'users': ['user1']},436                         True,437                         None),438                        ('acl_group_add_hosts',439                         {'id': 'acl0',440                          'hosts': ['host0', 'host1']},441                         False,442                         'DoesNotExist: The following hosts do not exist: '443                         'host1'),444                        ('acl_group_add_hosts',445                         {'id': 'acl0',446                          'hosts': ['host0']},447                         True,448                         None)])449        execute_result = acl_addrm.execute()450        self.god.check_playback()451        self.assertEqual(['acl0'], execute_result['users'])452        self.assertEqual(['acl0'], execute_result['hosts'])453        self.assertOutput(acl_addrm, execute_result,454                          out_words_ok=["Added to ACL 'acl0' user:",455                                        "Added to ACL 'acl0' host:",456                                        'user1', 'host0'],457                          err_words_ok=['DoesNotExist',458                                        'acl_group_add_hosts',459                                        'host1',460                                        'acl_group_add_users',461                                        'user0'])462    def test_execute_add_or_remove_to_topic_bad_acl_uh(self):463        acl_addrm = self._create_add_remove('acl0',464                                        users=['user0', 'user1'],465                                        hosts=['host0', 'host1'])466        self.mock_rpcs([('acl_group_add_users',467                         {'id': 'acl0',468                          'users': ['user0', 'user1']},469                         False,470                         'DoesNotExist: acl_group matching '471                         'query does not exist.'),472                        ('acl_group_add_hosts',473                         {'id': 'acl0',474                          'hosts': ['host0', 'host1']},475                         False,476                         'DoesNotExist: acl_group matching '477                         'query does not exist.')])478        execute_result = acl_addrm.execute()479        self.god.check_playback()480        self.assertEqual([], execute_result['users'])...

executor.py

Source:executor.py

1"""Base class for RPC executors.2RPC Executors are objects whose job is to send an RPC and interpret3the response as a sensor graph reading.4They can either run the RPC on an actual device or they can run it5locally in a simulated device or just mock the response as needed.6"""7import struct8from iotile.core.exceptions import InternalError, HardwareError9class RPCExecutor:10    """RPC Executors run RPCs on behalf of sensor graph."""11    def __init__(self):12        self.warning_channel = None13        self.mock_rpcs = {}14    def mock(self, slot, rpc_id, value):15        """Store a mock return value for an RPC16        Args:17            slot (SlotIdentifier): The slot we are mocking18            rpc_id (int): The rpc we are mocking19            value (int): The value that should be returned20                when the RPC is called.21        """22        address = slot.address23        if address not in self.mock_rpcs:24            self.mock_rpcs[address] = {}25        self.mock_rpcs[address][rpc_id] = value26    def warn(self, message):27        """Let the user of this RPCExecutor know about an issue.28        warn should be used when the RPC processed in a nonstandard way29        that typically indicates a misconfiguration or problem but did30        technically complete successfully.31        """32        if self.warning_channel is None:33            return34    def rpc(self, address, rpc_id):35        """Call an RPC and receive the result as an integer.36        If the RPC does not properly return a 32 bit integer, raise a warning37        unless it cannot be converted into an integer at all, in which case38        a HardwareError is thrown.39        Args:40            address (int): The address of the tile we want to call the RPC41                on42            rpc_id (int): The id of the RPC that we want to call43        Returns:44            int: The result of the RPC call.  If the rpc did not succeed45                an error is thrown instead.46        """47        # Always allow mocking an RPC to override whatever the defaul behavior is48        if address in self.mock_rpcs and rpc_id in self.mock_rpcs[address]:49            value = self.mock_rpcs[address][rpc_id]50            return value51        result = self._call_rpc(address, rpc_id, bytes())52        if len(result) != 4:53            self.warn(u"RPC 0x%X on address %d: response had invalid length %d not equal to 4" % (rpc_id, address, len(result)))54        if len(result) < 4:55            raise HardwareError("Response from RPC was not long enough to parse as an integer", rpc_id=rpc_id, address=address, response_length=len(result))56        if len(result) > 4:57            result = result[:4]58        res, = struct.unpack("<L", result)59        return res60    def _call_rpc(self, address, rpc_id, payload):61        """Call an RPC with the given information and return its response.62        Must raise a hardware error of the appropriate kind if the RPC63        can not be executed correctly.  Otherwise it should return the binary64        response payload received from the RPC.65        Args:66            address (int): The address of the tile we want to call the RPC67                on68            rpc_id (int): The id of the RPC that we want to call69            payload (bytes, bytearray): The data that we want to send as the payload70        """...

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.