Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _test_create_segment method in tempest

Best Python code snippet using tempest_python
test_segment.py
Source:test_segment.py
...85    def segment(self, **kwargs):86        kwargs.setdefault('network_type', 'net_type')87        return self._make_segment(88            self.fmt, tenant_id=self._tenant_id, **kwargs)89    def _test_create_segment(self, expected=None, **kwargs):90        keys = kwargs.copy()91        segment = self.segment(**keys)92        self._validate_resource(segment, keys, 'segment')93        if expected:94            self._compare_resource(segment, expected, 'segment')95        return segment96class SegmentTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,97                        portbindings_db.PortBindingMixin,98                        db.SegmentDbMixin):99    __native_pagination_support = True100    __native_sorting_support = True101    supported_extension_aliases = ["segment", "binding", "ip_allocation"]102    def get_plugin_description(self):103        return "Network Segments"104    @classmethod105    def get_plugin_type(cls):106        return "segments"107    def create_port(self, context, port):108        port_dict = super(SegmentTestPlugin, self).create_port(context, port)109        self._process_portbindings_create_and_update(110            context, port['port'], port_dict)111        return port_dict112    def update_port(self, context, id, port):113        port_dict = super(SegmentTestPlugin, self).update_port(114            context, id, port)115        self._process_portbindings_create_and_update(116            context, port['port'], port_dict)117        return port_dict118class TestSegmentNameDescription(SegmentTestCase):119    def setUp(self):120        super(TestSegmentNameDescription, self).setUp()121        with self.network() as network:122            self.network = network['network']123    def _test_create_segment(self, expected=None, **kwargs):124        for d in (kwargs, expected):125            if d is None:126                continue127            d.setdefault('network_id', self.network['id'])128            d.setdefault('name', None)129            d.setdefault('description', None)130            d.setdefault('physical_network', 'phys_net')131            d.setdefault('network_type', 'net_type')132            d.setdefault('segmentation_id', 200)133        return super(TestSegmentNameDescription, self)._test_create_segment(134            expected, **kwargs)135    def test_create_segment_no_name_description(self):136        self._test_create_segment(expected={})137    def test_create_segment_with_name(self):138        expected_segment = {'name': 'segment_name'}139        self._test_create_segment(name='segment_name',140                                  expected=expected_segment)141    def test_create_segment_with_description(self):142        expected_segment = {'description': 'A segment'}143        self._test_create_segment(description='A segment',144                                  expected=expected_segment)145    def test_update_segment_set_name(self):146        segment = self._test_create_segment()147        result = self._update('segments',148                              segment['segment']['id'],149                              {'segment': {'name': 'Segment name'}},150                              expected_code=webob.exc.HTTPOk.code)151        self.assertEqual('Segment name', result['segment']['name'])152    def test_update_segment_set_description(self):153        segment = self._test_create_segment()154        result = self._update('segments',155                              segment['segment']['id'],156                              {'segment': {'description': 'Segment desc'}},157                              expected_code=webob.exc.HTTPOk.code)158        self.assertEqual('Segment desc', result['segment']['description'])159    def test_update_segment_set_name_to_none(self):160        segment = self._test_create_segment(161            description='A segment', name='segment')162        result = self._update('segments',163                              segment['segment']['id'],164                              {'segment': {'name': None}},165                              expected_code=webob.exc.HTTPOk.code)166        self.assertIsNone(result['segment']['name'])167    def test_update_segment_set_description_to_none(self):168        segment = self._test_create_segment(169            description='A segment', name='segment')170        result = self._update('segments',171                              segment['segment']['id'],172                              {'segment': {'description': None}},173                              expected_code=webob.exc.HTTPOk.code)174        self.assertIsNone(result['segment']['description'])175class TestSegment(SegmentTestCase):176    def test_create_segment(self):177        with self.network() as network:178            network = network['network']179        expected_segment = {'network_id': network['id'],180                            'physical_network': 'phys_net',181                            'network_type': 'net_type',182                            'segmentation_id': 200}183        self._test_create_segment(network_id=network['id'],184                                  physical_network='phys_net',185                                  segmentation_id=200,186                                  expected=expected_segment)187    def test_create_segment_non_existent_network(self):188        exc = self.assertRaises(webob.exc.HTTPClientError,189                                self._test_create_segment,190                                network_id=uuidutils.generate_uuid(),191                                physical_network='phys_net',192                                segmentation_id=200)193        self.assertEqual(HTTP_NOT_FOUND, exc.code)194        self.assertIn('NetworkNotFound', exc.explanation)195    def test_create_segment_no_phys_net(self):196        with self.network() as network:197            network = network['network']198        expected_segment = {'network_id': network['id'],199                            'physical_network': None,200                            'network_type': 'net_type',201                            'segmentation_id': 200}202        self._test_create_segment(network_id=network['id'],203                                  segmentation_id=200,204                                  expected=expected_segment)205    def test_create_segment_no_segmentation_id(self):206        def _mock_reserve_segmentation_id(rtype, event, trigger,207                                          context, segment):208            if not segment.get('segmentation_id'):209                segment['segmentation_id'] = 200210        with self.network() as network:211            network = network['network']212        registry.subscribe(_mock_reserve_segmentation_id, resources.SEGMENT,213                           events.PRECOMMIT_CREATE)214        expected_segment = {'network_id': network['id'],215                            'physical_network': 'phys_net',216                            'network_type': 'net_type',217                            'segmentation_id': 200}218        self._test_create_segment(network_id=network['id'],219                                  physical_network='phys_net',220                                  expected=expected_segment)221    def test_create_segment_with_exception_in_core_plugin(self):222        cxt = context.get_admin_context()223        with self.network() as network:224            network = network['network']225        with mock.patch.object(registry, 'notify') as notify:226            notify.side_effect = exceptions.CallbackFailure(errors=Exception)227            self.assertRaises(webob.exc.HTTPClientError,228                              self.segment,229                              network_id=network['id'],230                              segmentation_id=200)231        network_segments = segments_db.get_network_segments(cxt.session,232                                                            network['id'])233        self.assertEqual([], network_segments)234    def test_create_segments_in_certain_order(self):235        cxt = context.get_admin_context()236        with self.network() as network:237            network = network['network']238            segment1 = self.segment(239                network_id=network['id'], segmentation_id=200)240            segment2 = self.segment(241                network_id=network['id'], segmentation_id=201)242            segment3 = self.segment(243                network_id=network['id'], segmentation_id=202)244            network_segments = segments_db.get_network_segments(cxt.session,245                                                                network['id'])246            self.assertEqual(segment1['segment']['id'],247                             network_segments[0]['id'])248            self.assertEqual(segment2['segment']['id'],249                             network_segments[1]['id'])250            self.assertEqual(segment3['segment']['id'],251                             network_segments[2]['id'])252    def test_delete_segment(self):253        with self.network() as network:254            network = network['network']255        self.segment(network_id=network['id'], segmentation_id=200)256        segment = self.segment(network_id=network['id'], segmentation_id=201)257        self._delete('segments', segment['segment']['id'])258        self._show('segments', segment['segment']['id'],259                   expected_code=webob.exc.HTTPNotFound.code)260    def test_delete_segment_failed_with_subnet_associated(self):261        with self.network() as network:262            net = network['network']263            segment = self._test_create_segment(network_id=net['id'],264                                                segmentation_id=200)265            segment_id = segment['segment']['id']266            with self.subnet(network=network, segment_id=segment_id):267                self._delete('segments', segment_id,268                             expected_code=webob.exc.HTTPConflict.code)269                exist_segment = self._show('segments', segment_id)270                self.assertEqual(segment_id, exist_segment['segment']['id'])271    def test_get_segment(self):272        with self.network() as network:273            network = network['network']274        segment = self._test_create_segment(network_id=network['id'],275                                            physical_network='phys_net',276                                            segmentation_id=200)277        req = self.new_show_request('segments', segment['segment']['id'])278        res = self.deserialize(self.fmt, req.get_response(self.ext_api))279        self.assertEqual(segment['segment']['id'], res['segment']['id'])280    def test_list_segments(self):281        with self.network() as network:282            network = network['network']283        self._test_create_segment(network_id=network['id'],284                                  physical_network='phys_net1',285                                  segmentation_id=200)286        self._test_create_segment(network_id=network['id'],287                                  physical_network='phys_net2',288                                  segmentation_id=201)289        res = self._list('segments')290        self.assertEqual(2, len(res['segments']))291    def test_update_segments(self):292        with self.network() as network:293            net = network['network']294            segment = self._test_create_segment(network_id=net['id'],295                                                segmentation_id=200)296            segment['segment']['segmentation_id'] = '201'297            self._update('segments', segment['segment']['id'], segment,298                         expected_code=webob.exc.HTTPClientError.code)299class TestSegmentML2(SegmentTestCase):300    def setUp(self):301        super(TestSegmentML2, self).setUp(plugin='ml2')302    def test_segment_notification_on_create_network(self):303        with mock.patch.object(registry, 'notify') as notify:304            with self.network():305                pass306        notify.assert_any_call(resources.SEGMENT,307                               events.PRECOMMIT_CREATE,308                               context=mock.ANY,309                               segment=mock.ANY,310                               trigger=mock.ANY)311class TestSegmentSubnetAssociation(SegmentTestCase):312    def test_basic_association(self):313        with self.network() as network:314            net = network['network']315        segment = self._test_create_segment(network_id=net['id'],316                                            segmentation_id=200)317        segment_id = segment['segment']['id']318        with self.subnet(network=network, segment_id=segment_id) as subnet:319            subnet = subnet['subnet']320        request = self.new_show_request('subnets', subnet['id'])321        response = request.get_response(self.api)322        res = self.deserialize(self.fmt, response)323        self.assertEqual(segment_id,324                         res['subnet']['segment_id'])325    def test_association_network_mismatch(self):326        with self.network() as network1:327            with self.network() as network2:328                net = network1['network']329        segment = self._test_create_segment(network_id=net['id'],330                                            segmentation_id=200)331        res = self._create_subnet(self.fmt,332                                  net_id=network2['network']['id'],333                                  tenant_id=network2['network']['tenant_id'],334                                  gateway_ip=constants.ATTR_NOT_SPECIFIED,335                                  cidr='10.0.0.0/24',336                                  segment_id=segment['segment']['id'])337        self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)338    def test_association_segment_not_found(self):339        with self.network() as network:340            net = network['network']341        segment_id = uuidutils.generate_uuid()342        res = self._create_subnet(self.fmt,343                                  net_id=net['id'],344                                  tenant_id=net['tenant_id'],345                                  gateway_ip=constants.ATTR_NOT_SPECIFIED,346                                  cidr='10.0.0.0/24',347                                  segment_id=segment_id)348        self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)349    def test_only_some_subnets_associated_not_allowed(self):350        with self.network() as network:351            with self.subnet(network=network):352                net = network['network']353        segment = self._test_create_segment(network_id=net['id'],354                                            segmentation_id=200)355        res = self._create_subnet(self.fmt,356                                  net_id=net['id'],357                                  tenant_id=net['tenant_id'],358                                  gateway_ip=constants.ATTR_NOT_SPECIFIED,359                                  cidr='10.0.1.0/24',360                                  segment_id=segment['segment']['id'])361        self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)362    def test_association_to_dynamic_segment_not_allowed(self):363        cxt = context.get_admin_context()364        with self.network() as network:365            net = network['network']366        # Can't create a dynamic segment through the API367        segment = {segments_db.NETWORK_TYPE: 'phys_net',368                   segments_db.PHYSICAL_NETWORK: 'net_type',369                   segments_db.SEGMENTATION_ID: 200}370        segments_db.add_network_segment(cxt,371                                        network_id=net['id'],372                                        segment=segment,373                                        is_dynamic=True)374        res = self._create_subnet(self.fmt,375                                  net_id=net['id'],376                                  tenant_id=net['tenant_id'],377                                  gateway_ip=constants.ATTR_NOT_SPECIFIED,378                                  cidr='10.0.0.0/24',379                                  segment_id=segment['id'])380        self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)381class HostSegmentMappingTestCase(SegmentTestCase):382    _mechanism_drivers = ['logger']383    def setUp(self, plugin=None):384        config.cfg.CONF.set_override('mechanism_drivers',385                                     self._mechanism_drivers,386                                     group='ml2')387        config.cfg.CONF.set_override('network_vlan_ranges',388                                     ['phys_net1', 'phys_net2'],389                                     group='ml2_type_vlan')390        if not plugin:391            plugin = 'ml2'392        super(HostSegmentMappingTestCase, self).setUp(plugin=plugin)393        db.subscribe()394    def _get_segments_for_host(self, host):395        ctx = context.get_admin_context()396        segments_host_list = ctx.session.query(397            db.SegmentHostMapping).filter_by(host=host)398        return {seg_host['segment_id']: seg_host399                for seg_host in segments_host_list}400    def _register_agent(self, host, mappings=None, plugin=None,401                        start_flag=True):402        helpers.register_ovs_agent(host=host, bridge_mappings=mappings,403                                   plugin=self.plugin, start_flag=start_flag)404    def _test_one_segment_one_host(self, host):405        physical_network = 'phys_net1'406        with self.network() as network:407            network = network['network']408        segment = self._test_create_segment(409            network_id=network['id'], physical_network=physical_network,410            segmentation_id=200, network_type=p_constants.TYPE_VLAN)['segment']411        self._register_agent(host, mappings={physical_network: 'br-eth-1'},412                             plugin=self.plugin)413        segments_host_db = self._get_segments_for_host(host)414        self.assertEqual(1, len(segments_host_db))415        self.assertEqual(segment['id'],416                         segments_host_db[segment['id']]['segment_id'])417        self.assertEqual(host, segments_host_db[segment['id']]['host'])418        return segment419class TestMl2HostSegmentMappingNoAgent(HostSegmentMappingTestCase):420    def setUp(self, plugin=None):421        if not plugin:422            plugin = TEST_PLUGIN_KLASS423        super(TestMl2HostSegmentMappingNoAgent, self).setUp(plugin=plugin)424    def test_update_segment_host_mapping(self):425        ctx = context.get_admin_context()426        host = 'host1'427        physnets = ['phys_net1']428        with self.network() as network:429            network = network['network']430        segment = self._test_create_segment(431            network_id=network['id'], physical_network='phys_net1',432            segmentation_id=200, network_type=p_constants.TYPE_VLAN)['segment']433        self._test_create_segment(434            network_id=network['id'], physical_network='phys_net2',435            segmentation_id=201, network_type=p_constants.TYPE_VLAN)['segment']436        segments = db.get_segments_with_phys_nets(ctx, physnets)437        segment_ids = {segment['id'] for segment in segments}438        db.update_segment_host_mapping(ctx, host, segment_ids)439        segments_host_db = self._get_segments_for_host(host)440        self.assertEqual(1, len(segments_host_db))441        self.assertEqual(segment['id'],442                         segments_host_db[segment['id']]['segment_id'])443        self.assertEqual(host, segments_host_db[segment['id']]['host'])444    def test_map_segment_to_hosts(self):445        ctx = context.get_admin_context()446        hosts = {'host1', 'host2', 'host3'}447        with self.network() as network:448            network = network['network']449        segment = self._test_create_segment(450            network_id=network['id'], physical_network='phys_net1',451            segmentation_id=200, network_type=p_constants.TYPE_VLAN)['segment']452        db.map_segment_to_hosts(ctx, segment['id'], hosts)453        updated_segment = self.plugin.get_segment(ctx, segment['id'])454        self.assertEqual(hosts, set(updated_segment['hosts']))455    def test_get_all_hosts_mapped_with_segments(self):456        ctx = context.get_admin_context()457        hosts = set()458        with self.network() as network:459            network_id = network['network']['id']460        for i in range(1, 3):461            host = "host%s" % i462            segment = self._test_create_segment(463                network_id=network_id, physical_network='phys_net%s' % i,464                segmentation_id=200 + i, network_type=p_constants.TYPE_VLAN)465            db.update_segment_host_mapping(466                ctx, host, {segment['segment']['id']})467            hosts.add(host)468        # Now they are 2 hosts with segment being mapped.469        actual_hosts = db.get_hosts_mapped_with_segments(ctx)470        self.assertEqual(hosts, actual_hosts)471class TestMl2HostSegmentMappingOVS(HostSegmentMappingTestCase):472    _mechanism_drivers = ['openvswitch', 'logger']473    mock_path = 'neutron.services.segments.db.update_segment_host_mapping'474    def test_new_agent(self):475        host = 'host1'476        self._test_one_segment_one_host(host)477    def test_updated_agent_changed_physical_networks(self):478        host = 'host1'479        physical_networks = ['phys_net1', 'phys_net2']480        networks = []481        segments = []482        for i in range(len(physical_networks)):483            with self.network() as network:484                networks.append(network['network'])485            segments.append(self._test_create_segment(486                network_id=networks[i]['id'],487                physical_network=physical_networks[i],488                segmentation_id=200,489                network_type=p_constants.TYPE_VLAN)['segment'])490        self._register_agent(host, mappings={physical_networks[0]: 'br-eth-1',491                                             physical_networks[1]: 'br-eth-2'},492                             plugin=self.plugin)493        segments_host_db = self._get_segments_for_host(host)494        self.assertEqual(len(physical_networks), len(segments_host_db))495        for segment in segments:496            self.assertEqual(segment['id'],497                             segments_host_db[segment['id']]['segment_id'])498            self.assertEqual(host, segments_host_db[segment['id']]['host'])499        self._register_agent(host, mappings={physical_networks[0]: 'br-eth-1'},500                             plugin=self.plugin)501        segments_host_db = self._get_segments_for_host(host)502        self.assertEqual(1, len(segments_host_db))503        self.assertEqual(segments[0]['id'],504                         segments_host_db[segments[0]['id']]['segment_id'])505        self.assertEqual(host, segments_host_db[segments[0]['id']]['host'])506    def test_same_segment_two_hosts(self):507        host1 = 'host1'508        host2 = 'host2'509        physical_network = 'phys_net1'510        segment = self._test_one_segment_one_host(host1)511        self._register_agent(host2, mappings={physical_network: 'br-eth-1'},512                             plugin=self.plugin)513        segments_host_db = self._get_segments_for_host(host2)514        self.assertEqual(1, len(segments_host_db))515        self.assertEqual(segment['id'],516                         segments_host_db[segment['id']]['segment_id'])517        self.assertEqual(host2, segments_host_db[segment['id']]['host'])518    def test_update_agent_only_change_agent_host_mapping(self):519        host1 = 'host1'520        host2 = 'host2'521        physical_network = 'phys_net1'522        with self.network() as network:523            network = network['network']524        segment1 = self._test_create_segment(525            network_id=network['id'],526            physical_network=physical_network,527            segmentation_id=200,528            network_type=p_constants.TYPE_VLAN)['segment']529        self._register_agent(host1, mappings={physical_network: 'br-eth-1'},530                             plugin=self.plugin)531        self._register_agent(host2, mappings={physical_network: 'br-eth-1'},532                             plugin=self.plugin)533        # Update agent at host2 should only change mapping with host2.534        other_phys_net = 'phys_net2'535        segment2 = self._test_create_segment(536            network_id=network['id'],537            physical_network=other_phys_net,538            segmentation_id=201,539            network_type=p_constants.TYPE_VLAN)['segment']540        self._register_agent(host2, mappings={other_phys_net: 'br-eth-2'},541                             plugin=self.plugin)542        # We should have segment1 map to host1 and segment2 map to host2 now543        segments_host_db1 = self._get_segments_for_host(host1)544        self.assertEqual(1, len(segments_host_db1))545        self.assertEqual(segment1['id'],546                         segments_host_db1[segment1['id']]['segment_id'])547        self.assertEqual(host1, segments_host_db1[segment1['id']]['host'])548        segments_host_db2 = self._get_segments_for_host(host2)549        self.assertEqual(1, len(segments_host_db2))550        self.assertEqual(segment2['id'],551                         segments_host_db2[segment2['id']]['segment_id'])552        self.assertEqual(host2, segments_host_db2[segment2['id']]['host'])553    def test_new_segment_after_host_reg(self):554        host1 = 'host1'555        physical_network = 'phys_net1'556        segment = self._test_one_segment_one_host(host1)557        with self.network() as network:558            network = network['network']559        segment2 = self._test_create_segment(560            network_id=network['id'], physical_network=physical_network,561            segmentation_id=201, network_type=p_constants.TYPE_VLAN)['segment']562        segments_host_db = self._get_segments_for_host(host1)563        self.assertEqual(set((segment['id'], segment2['id'])),564                         set(segments_host_db))565    def test_segment_deletion_removes_host_mapping(self):566        host = 'host1'567        segment = self._test_one_segment_one_host(host)568        self._delete('segments', segment['id'])569        segments_host_db = self._get_segments_for_host(host)570        self.assertFalse(segments_host_db)571    @mock.patch(mock_path)572    def test_agent_with_no_mappings(self, mock):573        host = 'host1'574        physical_network = 'phys_net1'575        with self.network() as network:576            network = network['network']577        self._test_create_segment(578            network_id=network['id'], physical_network=physical_network,579            segmentation_id=200, network_type=p_constants.TYPE_VLAN)580        self._register_agent(host, plugin=self.plugin)581        segments_host_db = self._get_segments_for_host(host)582        self.assertFalse(segments_host_db)583        self.assertFalse(mock.mock_calls)584class TestMl2HostSegmentMappingLinuxBridge(TestMl2HostSegmentMappingOVS):585    _mechanism_drivers = ['linuxbridge', 'logger']586    def _register_agent(self, host, mappings=None, plugin=None):587        helpers.register_linuxbridge_agent(host=host,588                                           bridge_mappings=mappings,589                                           plugin=self.plugin)590class TestMl2HostSegmentMappingMacvtap(TestMl2HostSegmentMappingOVS):591    _mechanism_drivers = ['macvtap', 'logger']592    def _register_agent(self, host, mappings=None, plugin=None):593        helpers.register_macvtap_agent(host=host, interface_mappings=mappings,594                                       plugin=self.plugin)595class TestMl2HostSegmentMappingSriovNicSwitch(TestMl2HostSegmentMappingOVS):596    _mechanism_drivers = ['sriovnicswitch', 'logger']597    def _register_agent(self, host, mappings=None, plugin=None):598        helpers.register_sriovnicswitch_agent(host=host,599                                              device_mappings=mappings,600                                              plugin=self.plugin)601class NoSupportHostSegmentMappingPlugin(db_base_plugin_v2.NeutronDbPluginV2,602                                        db.SegmentDbMixin,603                                        agents_db.AgentDbMixin):604    __native_pagination_support = True605    __native_sorting_support = True606    supported_extension_aliases = []607class TestHostSegmentMappingNoSupportFromPlugin(HostSegmentMappingTestCase):608    mock_path = 'neutron.services.segments.db.update_segment_host_mapping'609    def setUp(self):610        plugin = ('neutron.tests.unit.extensions.test_segment.'611                  'NoSupportHostSegmentMappingPlugin')612        super(TestHostSegmentMappingNoSupportFromPlugin, self).setUp(613              plugin=plugin)614    @mock.patch(mock_path)615    def test_host_segments_not_updated(self, mock):616        host = 'host1'617        physical_network = 'phys_net1'618        with self.network() as network:619            network = network['network']620        self._test_create_segment(network_id=network['id'],621                                  physical_network=physical_network,622                                  segmentation_id=200,623                                  network_type=p_constants.TYPE_VLAN)624        self._register_agent(host, mappings={physical_network: 'br-eth-1'},625                             plugin=self.plugin)626        segments_host_db = self._get_segments_for_host(host)627        self.assertFalse(segments_host_db)628        self.assertFalse(mock.mock_calls)629class TestMl2HostSegmentMappingAgentServerSynch(HostSegmentMappingTestCase):630    _mechanism_drivers = ['openvswitch', 'logger']631    mock_path = 'neutron.services.segments.db.update_segment_host_mapping'632    @mock.patch(mock_path)633    def test_starting_server_processes_agents(self, mock_function):634        host = 'agent_updating_starting_server'635        physical_network = 'phys_net1'636        self._register_agent(host, mappings={physical_network: 'br-eth-1'},637                             plugin=self.plugin, start_flag=False)638        self.assertTrue(host in db.reported_hosts)639        self.assertEqual(1, mock_function.call_count)640        expected_call = mock.call(mock.ANY, host, set())641        mock_function.assert_has_calls([expected_call])642    @mock.patch(mock_path)643    def test_starting_agent_is_processed(self, mock_function):644        host = 'starting_agent'645        physical_network = 'phys_net1'646        self._register_agent(host, mappings={physical_network: 'br-eth-1'},647                             plugin=self.plugin, start_flag=False)648        self.assertTrue(host in db.reported_hosts)649        self._register_agent(host, mappings={physical_network: 'br-eth-1'},650                             plugin=self.plugin, start_flag=True)651        self.assertTrue(host in db.reported_hosts)652        self.assertEqual(2, mock_function.call_count)653        expected_call = mock.call(mock.ANY, host, set())654        mock_function.assert_has_calls([expected_call, expected_call])655    @mock.patch(mock_path)656    def test_no_starting_agent_is_not_processed(self, mock_function):657        host = 'agent_with_no_start_update'658        physical_network = 'phys_net1'659        self._register_agent(host, mappings={physical_network: 'br-eth-1'},660                             plugin=self.plugin, start_flag=False)661        self.assertTrue(host in db.reported_hosts)662        mock_function.reset_mock()663        self._register_agent(host, mappings={physical_network: 'br-eth-1'},664                             plugin=self.plugin, start_flag=False)665        self.assertTrue(host in db.reported_hosts)666        mock_function.assert_not_called()667class TestSegmentAwareIpam(SegmentTestCase):668    def _setup_host_mappings(self, mappings=()):669        ctx = context.get_admin_context()670        with ctx.session.begin(subtransactions=True):671            for segment_id, host in mappings:672                record = db.SegmentHostMapping(673                    segment_id=segment_id,674                    host=host)675                ctx.session.add(record)676    def _create_test_segment_with_subnet(self,677                                         network=None,678                                         cidr='2001:db8:0:0::/64',679                                         physnet='physnet'):680        """Creates one network with one segment and one subnet"""681        if not network:682            with self.network() as network:683                pass684        segment = self._test_create_segment(685            network_id=network['network']['id'],686            physical_network=physnet,687            network_type=p_constants.TYPE_VLAN)688        ip_version = netaddr.IPNetwork(cidr).version if cidr else None689        with self.subnet(network=network,690                         segment_id=segment['segment']['id'],691                         ip_version=ip_version,692                         cidr=cidr) as subnet:693            self._validate_l2_adjacency(network['network']['id'],694                                        is_adjacent=False)695            return network, segment, subnet696    def _create_test_segments_with_subnets(self, num):697        """Creates one network with num segments and num subnets"""698        with self.network() as network:699            segments, subnets = [], []700            for i in range(num):701                cidr = '2001:db8:0:%s::/64' % i702                physnet = 'physnet%s' % i703                _net, segment, subnet = self._create_test_segment_with_subnet(704                    network=network, cidr=cidr, physnet=physnet)705                segments.append(segment)706                subnets.append(subnet)707            return network, segments, subnets708    def test_port_create_with_segment_subnets(self):709        """No binding information is provided, defer IP allocation"""710        network, segment, subnet = self._create_test_segment_with_subnet()711        response = self._create_port(self.fmt,712                                     net_id=network['network']['id'],713                                     tenant_id=network['network']['tenant_id'])714        res = self.deserialize(self.fmt, response)715        # Don't allocate IPs in this case because we didn't give binding info716        self.assertEqual(0, len(res['port']['fixed_ips']))717    def _assert_one_ip_in_subnet(self, response, cidr):718        res = self.deserialize(self.fmt, response)719        self.assertEqual(1, len(res['port']['fixed_ips']))720        ip = res['port']['fixed_ips'][0]['ip_address']721        ip_net = netaddr.IPNetwork(cidr)722        self.assertIn(ip, ip_net)723    def test_port_create_with_binding_information(self):724        """Binding information is provided, subnets are on segments"""725        network, segments, subnets = self._create_test_segments_with_subnets(3)726        # Map the host to the middle segment (by mocking host/segment mapping)727        self._setup_host_mappings([728            (segments[1]['segment']['id'], 'fakehost'),729            (segments[1]['segment']['id'], 'otherhost'),730            (segments[0]['segment']['id'], 'thirdhost')])731        response = self._create_port(self.fmt,732                                     net_id=network['network']['id'],733                                     tenant_id=network['network']['tenant_id'],734                                     arg_list=(portbindings.HOST_ID,),735                                     **{portbindings.HOST_ID: 'fakehost'})736        res = self.deserialize(self.fmt, response)737        self._validate_immediate_ip_allocation(res['port']['id'])738        # Since host mapped to middle segment, IP must come from middle subnet739        self._assert_one_ip_in_subnet(response, subnets[1]['subnet']['cidr'])740    def test_port_create_with_binding_and_no_subnets(self):741        """Binding information is provided, no subnets."""742        with self.network() as network:743            segment = self._test_create_segment(744                network_id=network['network']['id'],745                physical_network='physnet',746                network_type=p_constants.TYPE_VLAN)747        # Map the host to the segment748        self._setup_host_mappings([(segment['segment']['id'], 'fakehost')])749        response = self._create_port(self.fmt,750                                     net_id=network['network']['id'],751                                     tenant_id=network['network']['tenant_id'],752                                     arg_list=(portbindings.HOST_ID,),753                                     **{portbindings.HOST_ID: 'fakehost'})754        res = self.deserialize(self.fmt, response)755        # No subnets, so no allocation.  But, it shouldn't be an error.756        self.assertEqual(0, len(res['port']['fixed_ips']))757    def test_port_create_with_binding_information_fallback(self):758        """Binding information is provided, subnets not on segments"""759        with self.network() as network:760            with self.subnet(network=network,761                             ip_version=6,762                             cidr='2001:db8:0:0::/64') as subnet:763                segment = self._test_create_segment(764                    network_id=network['network']['id'],765                    physical_network='physnet',766                    network_type=p_constants.TYPE_VLAN)767        self._validate_l2_adjacency(network['network']['id'], is_adjacent=True)768        # Map the host to the segment769        self._setup_host_mappings([(segment['segment']['id'], 'fakehost')])770        response = self._create_port(self.fmt,771                                     net_id=network['network']['id'],772                                     tenant_id=network['network']['tenant_id'],773                                     arg_list=(portbindings.HOST_ID,),774                                     **{portbindings.HOST_ID: 'fakehost'})775        res = self.deserialize(self.fmt, response)776        self._validate_immediate_ip_allocation(res['port']['id'])777        # Since the subnet is not on a segment, fall back to it778        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])779    def test_port_create_on_unconnected_host(self):780        """Binding information provided, host not connected to any segment"""781        network, segment, _subnet = self._create_test_segment_with_subnet()782        response = self._create_port(self.fmt,783                                     net_id=network['network']['id'],784                                     tenant_id=network['network']['tenant_id'],785                                     arg_list=(portbindings.HOST_ID,),786                                     **{portbindings.HOST_ID: 'fakehost'})787        res = self.deserialize(self.fmt, response)788        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)789        self.assertEqual(segment_exc.HostNotConnectedToAnySegment.__name__,790                         res['NeutronError']['type'])791        # Ensure that mapping the segment to other hosts doesn't trip it up792        self._setup_host_mappings([(segment['segment']['id'], 'otherhost')])793        response = self._create_port(self.fmt,794                                     net_id=network['network']['id'],795                                     tenant_id=network['network']['tenant_id'],796                                     arg_list=(portbindings.HOST_ID,),797                                     **{portbindings.HOST_ID: 'fakehost'})798        res = self.deserialize(self.fmt, response)799        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)800        self.assertEqual(segment_exc.HostNotConnectedToAnySegment.__name__,801                         res['NeutronError']['type'])802    def test_port_create_on_multiconnected_host(self):803        """Binding information provided, host connected to multiple segments"""804        network, segments, subnets = self._create_test_segments_with_subnets(2)805        # This host is bound to multiple hosts806        self._setup_host_mappings([(segments[0]['segment']['id'], 'fakehost'),807                                   (segments[1]['segment']['id'], 'fakehost')])808        response = self._create_port(self.fmt,809                                     net_id=network['network']['id'],810                                     tenant_id=network['network']['tenant_id'],811                                     arg_list=(portbindings.HOST_ID,),812                                     **{portbindings.HOST_ID: 'fakehost'})813        res = self.deserialize(self.fmt, response)814        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)815        self.assertEqual(segment_exc.HostConnectedToMultipleSegments.__name__,816                         res['NeutronError']['type'])817    def test_port_update_excludes_hosts_on_segments(self):818        """No binding information is provided, subnets on segments"""819        with self.network() as network:820            segment = self._test_create_segment(821                network_id=network['network']['id'],822                physical_network='physnet',823                network_type=p_constants.TYPE_VLAN)824        # Create a port with no IP address (since there is no subnet)825        port = self._create_deferred_ip_port(network)826        # Create the subnet and try to update the port to get an IP827        with self.subnet(network=network,828                         segment_id=segment['segment']['id']) as subnet:829            # Try requesting an IP (but the only subnet is on a segment)830            data = {'port': {831                'fixed_ips': [{'subnet_id': subnet['subnet']['id']}]}}832            port_id = port['port']['id']833            port_req = self.new_update_request('ports', data, port_id)834            response = port_req.get_response(self.api)835        # Gets bad request because there are no eligible subnets.836        self.assertEqual(webob.exc.HTTPBadRequest.code, response.status_int)837    def _create_port_and_show(self, network, **kwargs):838        response = self._create_port(839            self.fmt,840            net_id=network['network']['id'],841            tenant_id=network['network']['tenant_id'],842            **kwargs)843        port = self.deserialize(self.fmt, response)844        request = self.new_show_request('ports', port['port']['id'])845        return self.deserialize(self.fmt, request.get_response(self.api))846    def test_port_create_with_no_fixed_ips_no_ipam_on_routed_network(self):847        """Ports requesting no fixed_ips not deferred, even on routed net"""848        with self.network() as network:849            segment = self._test_create_segment(850                network_id=network['network']['id'],851                physical_network='physnet',852                network_type=p_constants.TYPE_VLAN)853            with self.subnet(network=network,854                             segment_id=segment['segment']['id']):855                pass856        # Create an unbound port requesting no IP addresses857        response = self._create_port_and_show(network, fixed_ips=[])858        self.assertEqual([], response['port']['fixed_ips'])859        self.assertEqual(ip_allocation.IP_ALLOCATION_NONE,860                         response['port'][ip_allocation.IP_ALLOCATION])861    def test_port_create_with_no_fixed_ips_no_ipam(self):862        """Ports without addresses on non-routed networks are not deferred"""863        with self.network() as network:864            with self.subnet(network=network):865                pass866        # Create an unbound port requesting no IP addresses867        response = self._create_port_and_show(network, fixed_ips=[])868        self.assertEqual([], response['port']['fixed_ips'])869        self.assertEqual(ip_allocation.IP_ALLOCATION_NONE,870                         response['port'][ip_allocation.IP_ALLOCATION])871    def test_port_without_ip_not_deferred(self):872        """Ports without addresses on non-routed networks are not deferred"""873        with self.network() as network:874            pass875        # Create a bound port with no IP address (since there is no subnet)876        response = self._create_port(self.fmt,877                                     net_id=network['network']['id'],878                                     tenant_id=network['network']['tenant_id'],879                                     arg_list=(portbindings.HOST_ID,),880                                     **{portbindings.HOST_ID: 'fakehost'})881        port = self.deserialize(self.fmt, response)882        request = self.new_show_request('ports', port['port']['id'])883        response = self.deserialize(self.fmt, request.get_response(self.api))884        self.assertEqual([], response['port']['fixed_ips'])885        self.assertEqual(ip_allocation.IP_ALLOCATION_IMMEDIATE,886                         response['port'][ip_allocation.IP_ALLOCATION])887    def test_port_without_ip_not_deferred_no_binding(self):888        """Ports without addresses on non-routed networks are not deferred"""889        with self.network() as network:890            pass891        # Create a unbound port with no IP address (since there is no subnet)892        response = self._create_port_and_show(network)893        self.assertEqual([], response['port']['fixed_ips'])894        self.assertEqual(ip_allocation.IP_ALLOCATION_IMMEDIATE,895                         response['port'][ip_allocation.IP_ALLOCATION])896    def test_port_update_is_host_aware(self):897        """Binding information is provided, subnets on segments"""898        with self.network() as network:899            segment = self._test_create_segment(900                network_id=network['network']['id'],901                physical_network='physnet',902                network_type=p_constants.TYPE_VLAN)903        # Map the host to the segment904        self._setup_host_mappings([(segment['segment']['id'], 'fakehost')])905        # Create a bound port with no IP address (since there is no subnet)906        response = self._create_port(self.fmt,907                                     net_id=network['network']['id'],908                                     tenant_id=network['network']['tenant_id'],909                                     arg_list=(portbindings.HOST_ID,),910                                     **{portbindings.HOST_ID: 'fakehost'})911        port = self.deserialize(self.fmt, response)912        # Create the subnet and try to update the port to get an IP913        with self.subnet(network=network,914                         segment_id=segment['segment']['id']) as subnet:915            self._validate_l2_adjacency(network['network']['id'],916                                        is_adjacent=False)917            # Try requesting an IP (but the only subnet is on a segment)918            data = {'port': {919                'fixed_ips': [{'subnet_id': subnet['subnet']['id']}]}}920            port_id = port['port']['id']921            port_req = self.new_update_request('ports', data, port_id)922            response = port_req.get_response(self.api)923        # Since port is bound and there is a mapping to segment, it succeeds.924        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)925        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])926    def _validate_l2_adjacency(self, network_id, is_adjacent):927        request = self.new_show_request('networks', network_id)928        response = self.deserialize(self.fmt, request.get_response(self.api))929        self.assertEqual(is_adjacent,930                         response['network'][l2_adjacency.L2_ADJACENCY])931    def _validate_deferred_ip_allocation(self, port_id):932        request = self.new_show_request('ports', port_id)933        response = self.deserialize(self.fmt, request.get_response(self.api))934        self.assertEqual(ip_allocation.IP_ALLOCATION_DEFERRED,935                         response['port'][ip_allocation.IP_ALLOCATION])936        ips = response['port']['fixed_ips']937        self.assertEqual(0, len(ips))938    def _validate_immediate_ip_allocation(self, port_id):939        request = self.new_show_request('ports', port_id)940        response = self.deserialize(self.fmt, request.get_response(self.api))941        self.assertEqual(ip_allocation.IP_ALLOCATION_IMMEDIATE,942                         response['port'][ip_allocation.IP_ALLOCATION])943        ips = response['port']['fixed_ips']944        self.assertNotEqual(0, len(ips))945    def _create_deferred_ip_port(self, network):946        response = self._create_port(self.fmt,947                                     net_id=network['network']['id'],948                                     tenant_id=network['network']['tenant_id'])949        port = self.deserialize(self.fmt, response)950        ips = port['port']['fixed_ips']951        self.assertEqual(0, len(ips))952        return port953    def test_port_update_deferred_allocation(self):954        """Binding information is provided on update, subnets on segments"""955        network, segment, subnet = self._create_test_segment_with_subnet()956        # Map the host to the segment957        self._setup_host_mappings([(segment['segment']['id'], 'fakehost')])958        port = self._create_deferred_ip_port(network)959        self._validate_deferred_ip_allocation(port['port']['id'])960        # Try requesting an IP (but the only subnet is on a segment)961        data = {'port': {portbindings.HOST_ID: 'fakehost'}}962        port_id = port['port']['id']963        port_req = self.new_update_request('ports', data, port_id)964        response = port_req.get_response(self.api)965        # Port update succeeds and allocates a new IP address.966        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)967        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])968    def test_port_update_deferred_allocation_no_segments(self):969        """Binding information is provided, subnet created after port"""970        with self.network() as network:971            pass972        port = self._create_deferred_ip_port(network)973        # Create the subnet and try to update the port to get an IP974        with self.subnet(network=network):975            data = {'port': {portbindings.HOST_ID: 'fakehost'}}976            port_id = port['port']['id']977            port_req = self.new_update_request('ports', data, port_id)978            response = port_req.get_response(self.api)979        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)980        res = self.deserialize(self.fmt, response)981        self.assertEqual(0, len(res['port']['fixed_ips']))982    def test_port_update_deferred_allocation_no_ipam(self):983        """Binding information is provided on update. Don't allocate."""984        with self.network() as network:985            with self.subnet(network=network):986                pass987        response = self._create_port(self.fmt,988                                     net_id=network['network']['id'],989                                     tenant_id=network['network']['tenant_id'],990                                     fixed_ips=[])991        port = self.deserialize(self.fmt, response)992        ips = port['port']['fixed_ips']993        self.assertEqual(0, len(ips))994        # Create the subnet and try to update the port to get an IP995        data = {'port': {portbindings.HOST_ID: 'fakehost'}}996        port_id = port['port']['id']997        port_req = self.new_update_request('ports', data, port_id)998        response = port_req.get_response(self.api)999        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1000        res = self.deserialize(self.fmt, response)1001        self.assertEqual(0, len(res['port']['fixed_ips']))1002    def test_port_update_deferred_allocation_no_segments_manual_alloc(self):1003        """Binding information is provided, subnet created after port"""1004        with self.network() as network:1005            pass1006        port = self._create_deferred_ip_port(network)1007        # Create the subnet and try to update the port to get an IP1008        with self.subnet(network=network) as subnet:1009            data = {'port': {1010                portbindings.HOST_ID: 'fakehost',1011                'fixed_ips': [{'subnet_id': subnet['subnet']['id']}]}}1012            port_id = port['port']['id']1013            port_req = self.new_update_request('ports', data, port_id)1014            response = port_req.get_response(self.api)1015        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1016        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])1017        # Do a show to be sure that only one IP is recorded1018        port_req = self.new_show_request('ports', port_id)1019        response = port_req.get_response(self.api)1020        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1021        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])1022    def test_port_update_deferred_allocation_no_segments_empty_alloc(self):1023        """Binding information is provided, subnet created after port"""1024        with self.network() as network:1025            pass1026        port = self._create_deferred_ip_port(network)1027        # Create the subnet and update the port but specify no IPs1028        with self.subnet(network=network):1029            data = {'port': {1030                portbindings.HOST_ID: 'fakehost',1031                'fixed_ips': []}}1032            port_id = port['port']['id']1033            port_req = self.new_update_request('ports', data, port_id)1034            response = port_req.get_response(self.api)1035        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1036        res = self.deserialize(self.fmt, response)1037        # Since I specifically requested no IP addresses, I shouldn't get one.1038        self.assertEqual(0, len(res['port']['fixed_ips']))1039    def test_port_update_deferred_allocation_no_host_mapping(self):1040        """Binding information is provided on update, subnets on segments"""1041        network, segment, subnet = self._create_test_segment_with_subnet()1042        port = self._create_deferred_ip_port(network)1043        self._validate_deferred_ip_allocation(port['port']['id'])1044        # Try requesting an IP (but the only subnet is on a segment)1045        data = {'port': {portbindings.HOST_ID: 'fakehost'}}1046        port_id = port['port']['id']1047        port_req = self.new_update_request('ports', data, port_id)1048        response = port_req.get_response(self.api)1049        res = self.deserialize(self.fmt, response)1050        # Gets conflict because it can't map the host to a segment1051        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)1052        self.assertEqual(segment_exc.HostNotConnectedToAnySegment.__name__,1053                         res['NeutronError']['type'])1054    def test_port_update_deferred_allocation_multiple_host_mapping(self):1055        """Binding information is provided on update, subnets on segments"""1056        network, segments, _s = self._create_test_segments_with_subnets(2)1057        port = self._create_deferred_ip_port(network)1058        self._validate_deferred_ip_allocation(port['port']['id'])1059        # This host is bound to multiple segments1060        self._setup_host_mappings([(segments[0]['segment']['id'], 'fakehost'),1061                                   (segments[1]['segment']['id'], 'fakehost')])1062        # Try requesting an IP (but the only subnet is on a segment)1063        data = {'port': {portbindings.HOST_ID: 'fakehost'}}1064        port_id = port['port']['id']1065        port_req = self.new_update_request('ports', data, port_id)1066        response = port_req.get_response(self.api)1067        res = self.deserialize(self.fmt, response)1068        # Gets conflict because it can't map the host to a segment1069        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)1070        self.assertEqual(segment_exc.HostConnectedToMultipleSegments.__name__,1071                         res['NeutronError']['type'])1072    def test_port_update_allocate_no_segments(self):1073        """Binding information is provided, subnet created after port"""1074        with self.network() as network:1075            pass1076        # Create a bound port with no IP address (since there is not subnet)1077        port = self._create_deferred_ip_port(network)1078        # Create the subnet and try to update the port to get an IP1079        with self.subnet(network=network) as subnet:1080            # Try requesting an IP (but the only subnet is on a segment)1081            data = {'port': {1082                'fixed_ips': [{'subnet_id': subnet['subnet']['id']}]}}1083            port_id = port['port']['id']1084            port_req = self.new_update_request('ports', data, port_id)1085            response = port_req.get_response(self.api)1086        # Since port is bound and there is a mapping to segment, it succeeds.1087        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1088        self._assert_one_ip_in_subnet(response, subnet['subnet']['cidr'])1089    def test_port_update_deferred_allocation_no_ips(self):1090        """Binding information is provided on update, subnets on segments"""1091        network, segments, subnets = self._create_test_segments_with_subnets(2)1092        self._setup_host_mappings([(segments[0]['segment']['id'], 'fakehost2'),1093                                   (segments[1]['segment']['id'], 'fakehost')])1094        port = self._create_deferred_ip_port(network)1095        # Update the subnet on the second segment to be out of IPs1096        subnet_data = {'subnet': {'allocation_pools': []}}1097        subnet_req = self.new_update_request('subnets',1098                                             subnet_data,1099                                             subnets[1]['subnet']['id'])1100        subnet_response = subnet_req.get_response(self.api)1101        res = self.deserialize(self.fmt, subnet_response)1102        # Try requesting an IP (but the subnet ran out of ips)1103        data = {'port': {portbindings.HOST_ID: 'fakehost'}}1104        port_id = port['port']['id']1105        port_req = self.new_update_request('ports', data, port_id)1106        response = port_req.get_response(self.api)1107        res = self.deserialize(self.fmt, response)1108        # Since port is bound and there is a mapping to segment, it succeeds.1109        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)1110        self.assertEqual(n_exc.IpAddressGenerationFailure.__name__,1111                         res['NeutronError']['type'])1112    def test_port_update_fails_if_host_on_wrong_segment(self):1113        """Update a port with existing IPs to a host where they don't work"""1114        network, segments, subnets = self._create_test_segments_with_subnets(2)1115        self._setup_host_mappings([(segments[0]['segment']['id'], 'fakehost2'),1116                                   (segments[1]['segment']['id'], 'fakehost')])1117        # Create a bound port with an IP address1118        response = self._create_port(self.fmt,1119                                     net_id=network['network']['id'],1120                                     tenant_id=network['network']['tenant_id'],1121                                     arg_list=(portbindings.HOST_ID,),1122                                     **{portbindings.HOST_ID: 'fakehost'})1123        self._assert_one_ip_in_subnet(response, subnets[1]['subnet']['cidr'])1124        port = self.deserialize(self.fmt, response)1125        # Now, try to update binding to a host on the other segment1126        data = {'port': {portbindings.HOST_ID: 'fakehost2'}}1127        port_req = self.new_update_request('ports', data, port['port']['id'])1128        response = port_req.get_response(self.api)1129        # It fails since the IP address isn't compatible with the new segment1130        self.assertEqual(webob.exc.HTTPConflict.code, response.status_int)1131    def test_port_update_fails_if_host_on_good_segment(self):1132        """Update a port with existing IPs to a host where they don't work"""1133        network, segments, subnets = self._create_test_segments_with_subnets(2)1134        self._setup_host_mappings([(segments[0]['segment']['id'], 'fakehost2'),1135                                   (segments[1]['segment']['id'], 'fakehost1'),1136                                   (segments[1]['segment']['id'], 'fakehost')])1137        # Create a bound port with an IP address1138        response = self._create_port(self.fmt,1139                                     net_id=network['network']['id'],1140                                     tenant_id=network['network']['tenant_id'],1141                                     arg_list=(portbindings.HOST_ID,),1142                                     **{portbindings.HOST_ID: 'fakehost'})1143        self._assert_one_ip_in_subnet(response, subnets[1]['subnet']['cidr'])1144        port = self.deserialize(self.fmt, response)1145        # Now, try to update binding to another host in same segment1146        data = {'port': {portbindings.HOST_ID: 'fakehost1'}}1147        port_req = self.new_update_request('ports', data, port['port']['id'])1148        response = port_req.get_response(self.api)1149        # Since the new host is in the same segment, it succeeds.1150        self.assertEqual(webob.exc.HTTPOk.code, response.status_int)1151class TestSegmentAwareIpamML2(TestSegmentAwareIpam):1152    def setUp(self):1153        config.cfg.CONF.set_override('network_vlan_ranges',1154                                     ['physnet:200:209', 'physnet0:200:209',1155                                      'physnet1:200:209', 'physnet2:200:209'],1156                                     group='ml2_type_vlan')1157        super(TestSegmentAwareIpamML2, self).setUp(plugin='ml2')1158class TestDhcpAgentSegmentScheduling(HostSegmentMappingTestCase):1159    _mechanism_drivers = ['openvswitch', 'logger']1160    mock_path = 'neutron.services.segments.db.update_segment_host_mapping'1161    def setUp(self):1162        super(TestDhcpAgentSegmentScheduling, self).setUp()1163        self.dhcp_agent_db = agentschedulers_db.DhcpAgentSchedulerDbMixin()1164        self.ctx = context.get_admin_context()1165    def _test_create_network_and_segment(self, phys_net):1166        with self.network() as net:1167            network = net['network']1168        segment = self._test_create_segment(network_id=network['id'],1169                                            physical_network=phys_net,1170                                            segmentation_id=200,1171                                            network_type='vlan')1172        dhcp_agents = self.dhcp_agent_db.get_dhcp_agents_hosting_networks(1173            self.ctx, [network['id']])1174        self.assertEqual(0, len(dhcp_agents))1175        return network, segment['segment']1176    def _test_create_subnet(self, network, segment, cidr=None,1177                            enable_dhcp=True):1178        cidr = cidr or '10.0.0.0/24'1179        ip_version = 41180        with self.subnet(network={'network': network},1181                         segment_id=segment['id'],1182                         ip_version=ip_version,1183                         cidr=cidr,1184                         enable_dhcp=enable_dhcp) as subnet:1185            pass1186        return subnet['subnet']1187    def _register_dhcp_agents(self, hosts=None):1188        hosts = hosts or [DHCP_HOSTA, DHCP_HOSTB]1189        for host in hosts:1190            helpers.register_dhcp_agent(host)1191    def test_network_scheduling_on_segment_creation(self):1192        self._register_dhcp_agents()1193        self._test_create_network_and_segment('phys_net1')1194    def test_segment_scheduling_no_host_mapping(self):1195        self._register_dhcp_agents()1196        network, segment = self._test_create_network_and_segment('phys_net1')1197        self._test_create_subnet(network, segment)1198        dhcp_agents = self.dhcp_agent_db.get_dhcp_agents_hosting_networks(1199            self.ctx, [network['id']])1200        self.assertEqual(0, len(dhcp_agents))1201    def test_segment_scheduling_with_host_mapping(self):1202        phys_net1 = 'phys_net1'1203        self._register_dhcp_agents()1204        network, segment = self._test_create_network_and_segment(phys_net1)1205        self._register_agent(DHCP_HOSTA,1206                             mappings={phys_net1: 'br-eth-1'},1207                             plugin=self.plugin)1208        self._test_create_subnet(network, segment)1209        dhcp_agents = self.dhcp_agent_db.get_dhcp_agents_hosting_networks(1210            self.ctx, [network['id']])1211        self.assertEqual(1, len(dhcp_agents))1212        self.assertEqual(DHCP_HOSTA, dhcp_agents[0]['host'])1213    def test_segment_scheduling_with_multiple_host_mappings(self):1214        phys_net1 = 'phys_net1'1215        phys_net2 = 'phys_net2'1216        self._register_dhcp_agents([DHCP_HOSTA, DHCP_HOSTB, 'MEHA', 'MEHB'])1217        network, segment1 = self._test_create_network_and_segment(phys_net1)1218        segment2 = self._test_create_segment(network_id=network['id'],1219                                             physical_network=phys_net2,1220                                             segmentation_id=200,1221                                             network_type='vlan')['segment']1222        self._register_agent(DHCP_HOSTA,1223                             mappings={phys_net1: 'br-eth-1'},1224                             plugin=self.plugin)1225        self._register_agent(DHCP_HOSTB,1226                             mappings={phys_net2: 'br-eth-1'},1227                             plugin=self.plugin)1228        self._test_create_subnet(network, segment1)1229        self._test_create_subnet(network, segment2, cidr='11.0.0.0/24')1230        dhcp_agents = self.dhcp_agent_db.get_dhcp_agents_hosting_networks(1231            self.ctx, [network['id']])1232        self.assertEqual(2, len(dhcp_agents))...
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.