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

How to use list_networks method in tempest

Best Python code snippet using tempest_python

test_quantumv2.py

Source:test_quantumv2.py

...206            tenant_id=self.instance['project_id'],207            device_id=self.instance['uuid']).AndReturn(208                {'ports': port_data})209        nets = number == 1 and self.nets1 or self.nets2210        self.moxed_client.list_networks(211            tenant_id=self.instance['project_id'],212            shared=False).AndReturn({'networks': nets})213        self.moxed_client.list_networks(214            shared=True).AndReturn({'networks': []})215        for i in xrange(1, number + 1):216            subnet_data = i == 1 and self.subnet_data1 or self.subnet_data2217            self.moxed_client.list_subnets(218                id=mox.SameElementsAs(['my_subid%s' % i])).AndReturn(219                    {'subnets': subnet_data})220            self.moxed_client.list_ports(221                network_id=subnet_data[0]['network_id'],222                device_owner='network:dhcp').AndReturn(223                    {'ports': []})224        self.mox.ReplayAll()225        nw_inf = api.get_instance_nw_info(self.context, self.instance)226        for i in xrange(0, number):227            self._verify_nw_info(nw_inf, i)228    def test_get_instance_nw_info_1(self):229        """Test to get one port in one network and subnet."""230        self._get_instance_nw_info(1)231    def test_get_instance_nw_info_2(self):232        """Test to get one port in each of two networks and subnets."""233        self._get_instance_nw_info(2)234    def test_get_instance_nw_info_with_nets(self):235        """Test get instance_nw_info with networks passed in."""236        api = quantumapi.API()237        self.mox.StubOutWithMock(api.db, 'instance_info_cache_update')238        api.db.instance_info_cache_update(239            mox.IgnoreArg(),240            self.instance['uuid'], mox.IgnoreArg())241        self.moxed_client.list_ports(242            tenant_id=self.instance['project_id'],243            device_id=self.instance['uuid']).AndReturn(244                {'ports': self.port_data1})245        self.moxed_client.list_subnets(246            id=mox.SameElementsAs(['my_subid1'])).AndReturn(247                {'subnets': self.subnet_data1})248        self.moxed_client.list_ports(249            network_id='my_netid1',250            device_owner='network:dhcp').AndReturn(251                {'ports': self.dhcp_port_data1})252        self.mox.ReplayAll()253        nw_inf = api.get_instance_nw_info(self.context,254                                          self.instance,255                                          networks=self.nets1)256        self._verify_nw_info(nw_inf, 0)257    def test_get_instance_nw_info_without_subnet(self):258        """Test get instance_nw_info for a port without subnet."""259        api = quantumapi.API()260        self.mox.StubOutWithMock(api.db, 'instance_info_cache_update')261        api.db.instance_info_cache_update(262            mox.IgnoreArg(),263            self.instance['uuid'], mox.IgnoreArg())264        self.moxed_client.list_ports(265            tenant_id=self.instance['project_id'],266            device_id=self.instance['uuid']).AndReturn(267                {'ports': self.port_data3})268        self.moxed_client.list_networks(269            shared=False,270            tenant_id=self.instance['project_id']).AndReturn(271                {'networks': self.nets1})272        self.moxed_client.list_networks(273            shared=True).AndReturn({'networks': []})274        self.mox.ReplayAll()275        nw_inf = api.get_instance_nw_info(self.context,276                                          self.instance)277        id_suffix = 3278        self.assertEquals(0, len(nw_inf.fixed_ips()))279        self.assertEquals('my_netname1', nw_inf[0]['network']['label'])280        self.assertEquals('my_portid%s' % id_suffix, nw_inf[0]['id'])281        self.assertEquals('my_mac%s' % id_suffix, nw_inf[0]['address'])282        self.assertEquals(0, len(nw_inf[0]['network']['subnets']))283    def _allocate_for_instance(self, net_idx=1, **kwargs):284        api = quantumapi.API()285        self.mox.StubOutWithMock(api, 'get_instance_nw_info')286        # Net idx is 1-based for compatibility with existing unit tests287        nets = self.nets[net_idx - 1]288        api.get_instance_nw_info(mox.IgnoreArg(),289                                 self.instance,290                                 networks=nets).AndReturn(None)291        ports = {}292        fixed_ips = {}293        req_net_ids = []294        if 'requested_networks' in kwargs:295            for id, fixed_ip, port_id in kwargs['requested_networks']:296                if port_id:297                    self.moxed_client.show_port(port_id).AndReturn(298                        {'port': {'id': 'my_portid1',299                         'network_id': 'my_netid1'}})300                    req_net_ids.append('my_netid1')301                    ports['my_netid1'] = self.port_data1[0]302                    id = 'my_netid1'303                else:304                    fixed_ips[id] = fixed_ip305                req_net_ids.append(id)306        search_ids = [net['id'] for net in nets if net['id'] in req_net_ids]307        mox_list_network_params = dict(tenant_id=self.instance['project_id'],308                                       shared=False)309        if search_ids:310            mox_list_network_params['id'] = search_ids311        self.moxed_client.list_networks(312            **mox_list_network_params).AndReturn({'networks': nets})313        mox_list_network_params = dict(shared=True)314        if search_ids:315            mox_list_network_params['id'] = search_ids316        self.moxed_client.list_networks(317            **mox_list_network_params).AndReturn({'networks': []})318        for network in nets:319            port_req_body = {320                'port': {321                    'device_id': self.instance['uuid'],322                    'device_owner': 'compute:nova',323                },324            }325            port = ports.get(network['id'], None)326            if port:327                port_id = port['id']328                self.moxed_client.update_port(port_id,329                                              MyComparator(port_req_body)330                                              ).AndReturn(331                                                  {'port': port})332            else:333                fixed_ip = fixed_ips.get(network['id'])334                if fixed_ip:335                    port_req_body['port']['fixed_ip'] = fixed_ip336                port_req_body['port']['network_id'] = network['id']337                port_req_body['port']['admin_state_up'] = True338                port_req_body['port']['tenant_id'] = \339                    self.instance['project_id']340                res_port = {'port': {'id': 'fake'}}341                self.moxed_client.create_port(342                    MyComparator(port_req_body)).AndReturn(res_port)343        self.mox.ReplayAll()344        api.allocate_for_instance(self.context, self.instance, **kwargs)345    def test_allocate_for_instance_1(self):346        """Allocate one port in one network env."""347        self._allocate_for_instance(1)348    def test_allocate_for_instance_2(self):349        """Allocate one port in two networks env."""350        self._allocate_for_instance(2)351    def test_allocate_for_instance_with_requested_networks(self):352        # specify only first and last network353        requested_networks = [(net['id'], None, None)354                              for net in (self.nets3[0], self.nets3[-1])]355        self._allocate_for_instance(net_idx=3,356                                    requested_networks=requested_networks)357    def test_allocate_for_instance_with_requested_networks_with_fixedip(self):358        # specify only first and last network359        requested_networks = [(self.nets1[0]['id'], '10.0.1.0/24', None)]360        self._allocate_for_instance(net_idx=1,361                                    requested_networks=requested_networks)362    def test_allocate_for_instance_with_requested_networks_with_port(self):363        # specify only first and last network364        requested_networks = [(None, None, 'myportid1')]365        self._allocate_for_instance(net_idx=1,366                                    requested_networks=requested_networks)367    def test_allocate_for_instance_ex1(self):368        """verify we will delete created ports369        if we fail to allocate all net resources.370        Mox to raise exception when creating a second port.371        In this case, the code should delete the first created port.372        """373        api = quantumapi.API()374        self.moxed_client.list_networks(375            tenant_id=self.instance['project_id'],376            shared=False).AndReturn(377                {'networks': self.nets2})378        self.moxed_client.list_networks(shared=True).AndReturn(379                {'networks': []})380        index = 0381        for network in self.nets2:382            port_req_body = {383                'port': {384                    'network_id': network['id'],385                    'admin_state_up': True,386                    'device_id': self.instance['uuid'],387                    'device_owner': 'compute:nova',388                    'tenant_id': self.instance['project_id'],389                },390            }391            port = {'id': 'portid_' + network['id']}392            if index == 0:393                self.moxed_client.create_port(394                    MyComparator(port_req_body)).AndReturn({'port': port})395            else:396                self.moxed_client.create_port(397                    MyComparator(port_req_body)).AndRaise(398                        Exception("fail to create port"))399            index += 1400        self.moxed_client.delete_port('portid_' + self.nets2[0]['id'])401        self.mox.ReplayAll()402        self.assertRaises(QUANTUM_CLIENT_EXCEPTION, api.allocate_for_instance,403                          self.context, self.instance)404    def test_allocate_for_instance_ex2(self):405        """verify we have no port to delete406        if we fail to allocate the first net resource.407        Mox to raise exception when creating the first port.408        In this case, the code should not delete any ports.409        """410        api = quantumapi.API()411        self.moxed_client.list_networks(412            tenant_id=self.instance['project_id'],413            shared=False).AndReturn(414                {'networks': self.nets2})415        self.moxed_client.list_networks(shared=True).AndReturn(416                {'networks': []})417        port_req_body = {418            'port': {419                'network_id': self.nets2[0]['id'],420                'admin_state_up': True,421                'device_id': self.instance['uuid'],422                'tenant_id': self.instance['project_id'],423            },424        }425        self.moxed_client.create_port(426            MyComparator(port_req_body)).AndRaise(427                Exception("fail to create port"))428        self.mox.ReplayAll()429        self.assertRaises(QUANTUM_CLIENT_EXCEPTION, api.allocate_for_instance,430                          self.context, self.instance)431    def _deallocate_for_instance(self, number):432        port_data = number == 1 and self.port_data1 or self.port_data2433        self.moxed_client.list_ports(434            device_id=self.instance['uuid']).AndReturn(435                {'ports': port_data})436        for port in port_data:437            self.moxed_client.delete_port(port['id'])438        self.mox.ReplayAll()439        api = quantumapi.API()440        api.deallocate_for_instance(self.context, self.instance)441    def test_deallocate_for_instance_1(self):442        """Test to deallocate in one port env."""443        self._deallocate_for_instance(1)444    def test_deallocate_for_instance_2(self):445        """Test to deallocate in two ports env."""446        self._deallocate_for_instance(2)447    def test_validate_networks(self):448        requested_networks = [('my_netid1', 'test', None),449                              ('my_netid2', 'test2', None)]450        self.moxed_client.list_networks(451            id=mox.SameElementsAs(['my_netid1', 'my_netid2']),452            tenant_id=self.context.project_id,453            shared=False).AndReturn(454                {'networks': self.nets2})455        self.moxed_client.list_networks(456            id=mox.SameElementsAs(['my_netid1', 'my_netid2']),457            shared=True).AndReturn(458                {'networks': []})459        self.mox.ReplayAll()460        api = quantumapi.API()461        api.validate_networks(self.context, requested_networks)462    def test_validate_networks_ex_1(self):463        requested_networks = [('my_netid1', 'test', None),464                              ('my_netid2', 'test2', None)]465        self.moxed_client.list_networks(466            id=mox.SameElementsAs(['my_netid1', 'my_netid2']),467            tenant_id=self.context.project_id,468            shared=False).AndReturn(469                {'networks': self.nets1})470        self.moxed_client.list_networks(471            id=mox.SameElementsAs(['my_netid1', 'my_netid2']),472            shared=True).AndReturn(473                {'networks': []})474        self.mox.ReplayAll()475        api = quantumapi.API()476        try:477            api.validate_networks(self.context, requested_networks)478        except exception.NetworkNotFound as ex:479            self.assertTrue("my_netid2" in str(ex))480    def test_validate_networks_ex_2(self):481        requested_networks = [('my_netid1', 'test', None),482                              ('my_netid2', 'test2', None),483                              ('my_netid3', 'test3', None)]484        self.moxed_client.list_networks(485            id=mox.SameElementsAs(['my_netid1', 'my_netid2', 'my_netid3']),486            tenant_id=self.context.project_id,487            shared=False).AndReturn(488                {'networks': self.nets1})489        self.moxed_client.list_networks(490            id=mox.SameElementsAs(['my_netid1', 'my_netid2', 'my_netid3']),491            shared=True).AndReturn(492                {'networks': []})493        self.mox.ReplayAll()494        api = quantumapi.API()495        try:496            api.validate_networks(self.context, requested_networks)497        except exception.NetworkNotFound as ex:498            self.assertTrue("my_netid2, my_netid3" in str(ex))499    def _mock_list_ports(self, port_data=None):500        if port_data is None:501            port_data = self.port_data2502        address = self.port_address503        self.moxed_client.list_ports(504            fixed_ips=MyComparator('ip_address=%s' % address)).AndReturn(505                {'ports': port_data})506        self.mox.ReplayAll()507        return address508    def test_get_instance_uuids_by_ip_filter(self):509        self._mock_list_ports()510        filters = {'ip': '^10\\.0\\.1\\.2$'}511        api = quantumapi.API()512        result = api.get_instance_uuids_by_ip_filter(self.context, filters)513        self.assertEquals('device_id1', result[0]['instance_uuid'])514        self.assertEquals('device_id2', result[1]['instance_uuid'])515    def test_get_fixed_ip_by_address_fails_for_no_ports(self):516        address = self._mock_list_ports(port_data=[])517        api = quantumapi.API()518        self.assertRaises(exception.FixedIpNotFoundForAddress,519                          api.get_fixed_ip_by_address,520                          self.context, address)521    def test_get_fixed_ip_by_address_succeeds_for_1_port(self):522        address = self._mock_list_ports(port_data=self.port_data1)523        api = quantumapi.API()524        result = api.get_fixed_ip_by_address(self.context, address)525        self.assertEquals('device_id1', result['instance_uuid'])526    def test_get_fixed_ip_by_address_fails_for_more_than_1_port(self):527        address = self._mock_list_ports()528        api = quantumapi.API()529        self.assertRaises(exception.FixedIpAssociatedWithMultipleInstances,530                          api.get_fixed_ip_by_address,531                          self.context, address)532    def _get_available_networks(self, prv_nets, pub_nets, req_ids=None):533        api = quantumapi.API()534        nets = prv_nets + pub_nets535        mox_list_network_params = dict(tenant_id=self.instance['project_id'],536                                       shared=False)537        if req_ids:538            mox_list_network_params['id'] = req_ids539        self.moxed_client.list_networks(540            **mox_list_network_params).AndReturn({'networks': prv_nets})541        mox_list_network_params = dict(shared=True)542        if req_ids:543            mox_list_network_params['id'] = req_ids544        self.moxed_client.list_networks(545            **mox_list_network_params).AndReturn({'networks': pub_nets})546        self.mox.ReplayAll()547        rets = api._get_available_networks(self.context,548                                           self.instance['project_id'],549                                           req_ids)550        self.assertEqual(rets, nets)551    def test_get_available_networks_all_private(self):552        self._get_available_networks(prv_nets=self.nets2, pub_nets=[])553    def test_get_available_networks_all_public(self):554        self._get_available_networks(prv_nets=[], pub_nets=self.nets2)555    def test_get_available_networks_private_and_public(self):556        self._get_available_networks(prv_nets=self.nets1, pub_nets=self.nets4)557    def test_get_available_networks_with_network_ids(self):558        prv_nets = [self.nets3[0]]...

fixtures.py

Source:fixtures.py

1import mock2import os3import pytest4import time5from contextlib import contextmanager6from pyfakefs import fake_filesystem7from typing import Callable, Dict, List, Optional8with mock.patch('builtins.open', create=True):9    from importlib.machinery import SourceFileLoader10    cd = SourceFileLoader('cephadm', 'cephadm').load_module()11def mock_docker():12    docker = mock.Mock(cd.Docker)13    docker.path = '/usr/bin/docker'14    return docker15def mock_podman():16    podman = mock.Mock(cd.Podman)17    podman.path = '/usr/bin/podman'18    podman.version = (2, 1, 0)19    return podman20def _daemon_path():21    return os.getcwd()22def _mock_scrape_host(obj, interval):23    try:24        raise ValueError("wah")25    except Exception as e:26        obj._handle_thread_exception(e, 'host')27def _mock_run(obj):28    t = obj._create_thread(obj._scrape_host_facts, 'host', 5)29    time.sleep(1)30    if not t.is_alive():31        obj.cephadm_cache.update_health('host', "inactive", "host thread stopped")32    33@pytest.fixture34def exporter():35    with mock.patch('cephadm.CephadmDaemon.daemon_path', _daemon_path()), \36       mock.patch('cephadm.CephadmDaemon.can_run', return_value=True), \37       mock.patch('cephadm.CephadmDaemon.run', _mock_run), \38       mock.patch('cephadm.CephadmDaemon._scrape_host_facts', _mock_scrape_host):39        ctx = cd.CephadmContext()40        exporter = cd.CephadmDaemon(ctx, fsid='foobar', daemon_id='test')41        assert exporter.token == 'MyAccessToken' 42        yield exporter43@pytest.fixture()44def cephadm_fs(45    fs: fake_filesystem.FakeFilesystem,46):47    """48    use pyfakefs to stub filesystem calls49    """50    uid = os.getuid()51    gid = os.getgid()52    with mock.patch('os.fchown'), \53         mock.patch('os.fchmod'), \54         mock.patch('platform.processor', return_value='x86_64'), \55         mock.patch('cephadm.extract_uid_gid', return_value=(uid, gid)):56            fs.create_dir(cd.DATA_DIR)57            fs.create_dir(cd.LOG_DIR)58            fs.create_dir(cd.LOCK_DIR)59            fs.create_dir(cd.LOGROTATE_DIR)60            fs.create_dir(cd.UNIT_DIR)61            yield fs62@contextmanager63def with_cephadm_ctx(64    cmd: List[str],65    container_engine: Callable = mock_podman(),66    list_networks: Optional[Dict[str,Dict[str,List[str]]]] = None,67    hostname: Optional[str] = None,68):69    """70    :param cmd: cephadm command argv71    :param container_engine: container engine mock (podman or docker)72    :param list_networks: mock 'list-networks' return73    :param hostname: mock 'socket.gethostname' return74    """75    if not list_networks:76        list_networks = {}77    if not hostname:78        hostname = 'host1'79    with mock.patch('cephadm.attempt_bind'), \80         mock.patch('cephadm.call', return_value=('', '', 0)), \81         mock.patch('cephadm.call_timeout', return_value=0), \82         mock.patch('cephadm.find_executable', return_value='foo'), \83         mock.patch('cephadm.is_available', return_value=True), \84         mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}), \85         mock.patch('cephadm.list_networks', return_value=list_networks), \86         mock.patch('socket.gethostname', return_value=hostname):87             ctx: cd.CephadmContext = cd.cephadm_init_ctx(cmd)88             ctx.container_engine = container_engine...

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.