def sync_highperformance_routers(self, context, **kwargs):
     routers = self.sync_routers(context, **kwargs)
     for router in routers:
         for snat_port in router.get(constants.SNAT_ROUTER_INTF_KEY, []):
             network_segments = ml2_db.get_network_segments(context.session, 
                                                            snat_port['network_id'])
             if network_segments:
                 snat_port['vni'] = network_segments[0][api.SEGMENTATION_ID]
         gw_port = router.get('gw_port')
         if gw_port:
             network_segments = ml2_db.get_network_segments(context.session, 
                                                            gw_port['network_id'])
             if network_segments:
                 gw_port['vlan'] = network_segments[0][api.SEGMENTATION_ID]
     
     return routers

    def test_release_network_segments(self):
        data = {
            "network": {
                "name": "net1",
                "admin_state_up": True,
                "shared": False,
                pnet.NETWORK_TYPE: "vlan",
                pnet.PHYSICAL_NETWORK: "physnet1",
                pnet.SEGMENTATION_ID: 1,
                "tenant_id": "tenant_one",
            }
        }
        network_req = self.new_create_request("networks", data)
        res = network_req.get_response(self.api)
        network = self.deserialize(self.fmt, res)
        network_id = network["network"]["id"]
        segment = {driver_api.NETWORK_TYPE: "vlan", driver_api.PHYSICAL_NETWORK: "physnet2"}
        self.driver.type_manager.allocate_dynamic_segment(self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session, network_id, "physnet2")
        self.assertEqual("vlan", dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual("physnet2", dynamic_segment[driver_api.PHYSICAL_NETWORK])
        self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)

        with mock.patch.object(type_vlan.VlanTypeDriver, "release_segment") as rs:
            req = self.new_delete_request("networks", network_id)
            res = req.get_response(self.api)
            self.assertEqual(2, rs.call_count)
        self.assertEqual(ml2_db.get_network_segments(self.context.session, network_id), [])
        self.assertIsNone(ml2_db.get_dynamic_segment(self.context.session, network_id, "physnet2"))

    def _extend_network_dict_provider(self, context, network):
        if 'id' not in network:
            LOG.debug("Network has no id")
            network[provider.NETWORK_TYPE] = None
            network[provider.PHYSICAL_NETWORK] = None
            network[provider.SEGMENTATION_ID] = None
            return

        net_id = network['id']
        segments = db_ml2.get_network_segments(
            context.session, net_id, filter_dynamic=True)

        if not segments:
            LOG.debug("Network %s has no segments", net_id)
            network[provider.NETWORK_TYPE] = None
            network[provider.PHYSICAL_NETWORK] = None
            network[provider.SEGMENTATION_ID] = None
        elif len(segments) > 1:
            network[mpnet.SEGMENTS] = [
                {provider.NETWORK_TYPE: segment[api.NETWORK_TYPE],
                 provider.PHYSICAL_NETWORK: segment[api.PHYSICAL_NETWORK],
                 provider.SEGMENTATION_ID: segment[api.SEGMENTATION_ID]}
                for segment in segments]
        else:
            segment = segments[0]
            network[provider.NETWORK_TYPE] = segment[api.NETWORK_TYPE]
            network[provider.PHYSICAL_NETWORK] = segment[api.PHYSICAL_NETWORK]
            network[provider.SEGMENTATION_ID] = segment[api.SEGMENTATION_ID]

    def test_release_network_segments(self):
        data = {'network': {'name': 'net1',
                            'admin_state_up': True,
                            'shared': False,
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        network = self.deserialize(self.fmt, res)
        network_id = network['network']['id']
        segment = {driver_api.NETWORK_TYPE: 'vlan',
                   driver_api.PHYSICAL_NETWORK: 'physnet2'}
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     'physnet2')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet2',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)

        with mock.patch.object(type_vlan.VlanTypeDriver,
                               'release_segment') as rs:
            req = self.new_delete_request('networks', network_id)
            res = req.get_response(self.api)
            self.assertEqual(2, rs.call_count)
        self.assertEqual(ml2_db.get_network_segments(
            self.context.session, network_id), [])
        self.assertIsNone(ml2_db.get_dynamic_segment(
            self.context.session, network_id, 'physnet2'))

 def __init__(self, plugin, plugin_context, network,
              original_network=None):
     super(NetworkContext, self).__init__(plugin, plugin_context)
     self._network = network
     self._original_network = original_network
     self._segments = db.get_network_segments(plugin_context.session,
                                              network['id'])

    def get_network_segid(self, sid):
        """Get network segmentation id.

        :param sid: requested segment id
        """
        db_session = db.get_session()
        seg_entry = ml2db.get_network_segments(db_session, sid)
        return seg_entry[0]['segmentation_id']

 def get_shared_network_owner_id(self, network_id):
     filters = {'id': [network_id]}
     nets = self.get_networks(self.admin_ctx, filters=filters) or []
     segments = ml2_db.get_network_segments(self.admin_ctx.session,
                                            network_id)
     if not nets or not segments:
         return
     if (nets[0]['shared'] and
        segments[0][driver_api.NETWORK_TYPE] == p_const.TYPE_VLAN):
         return nets[0]['tenant_id']

    def release_network_segments(self, session, network_id):
        segments = db.get_network_segments(session, network_id, filter_dynamic=None)

        for segment in segments:
            network_type = segment.get(api.NETWORK_TYPE)
            driver = self.drivers.get(network_type)
            if driver:
                driver.obj.release_segment(session, segment)
            else:
                LOG.error(_LE("Failed to release segment '%s' because " "network type is not supported."), segment)

    def delete_network(self, context, id):
        session = context.session
        with session.begin(subtransactions=True):
            segments = db.get_network_segments(session, id)
            super(Ml2Plugin, self).delete_network(context, id)
            for segment in segments:
                self.type_manager.release_segment(session, segment)
            # The segment records are deleted via cascade from the
            # network record, so explicit removal is not necessary.

        self.notifier.network_delete(context, id)

    def _create_segments(self, segments, is_seg_dynamic=False, network_id="foo-network-id"):
        self._setup_neutron_network(network_id)
        for segment in segments:
            ml2_db.add_network_segment(self.ctx.session, network_id, segment, is_dynamic=is_seg_dynamic)

        net_segments = ml2_db.get_network_segments(self.ctx.session, network_id, filter_dynamic=is_seg_dynamic)

        for segment_index, segment in enumerate(segments):
            self.assertEqual(segment, net_segments[segment_index])

        return net_segments

def get_vlan_id_of_physical_network(context, network_id, physical_network):
    if hasattr(context, 'session'):
        session = context.session
    else:
        session = context.network._plugin_context.session
    segments = db_ml2.get_network_segments(session, network_id)
    if not segments:
        return ''
    for segment in segments:
        if segment[api.PHYSICAL_NETWORK] == physical_network:
            return str(segment[api.SEGMENTATION_ID])
    return str(segments[0][api.SEGMENTATION_ID])

 def _log_subnet_information(self, method_name, current_context, prev_context):
     driver_logger.info("**** %s ****" % (method_name))
     driver_logger.info("Current Subnet Name: %s" % (current_context['name']))
     driver_logger.info("Current Subnet CIDR: %s" % (current_context['cidr']))
     # Extract the Network ID from the Subnet Context
     network_id = current_context['network_id']
     # Get the Neutron DB Session Handle
     session = db.get_session()
     # Using ML2 DB API, fetch the Network that matches the Network ID
     networks = ml2_db.get_network_segments(session, network_id)
     driver_logger.info("Network associated to the Subnet: %s" % (networks))
     driver_logger.info("**** %s ****" % (method_name))

 def _notify_port_updated(self, context, port):
     session = context.session
     with session.begin(subtransactions=True):
         network_id = port['network_id']
         segments = db.get_network_segments(session, network_id)
         if not segments:
             LOG.warning(_("In _notify_port_updated() for port %(port_id), "
                           "network %(network_id) has no segments"),
                         {'port_id': port['id'],
                          'network_id': network_id})
             return
         # TODO(rkukura): Use port binding to select segment.
         segment = segments[0]
     self.notifier.port_update(context, port,
                               segment[api.NETWORK_TYPE],
                               segment[api.SEGMENTATION_ID],
                               segment[api.PHYSICAL_NETWORK])

 def _extend_network_dict_provider(self, context, network):
     id = network['id']
     segments = db.get_network_segments(context.session, id)
     if not segments:
         LOG.error(_("Network %s has no segments"), id)
         network[provider.NETWORK_TYPE] = None
         network[provider.PHYSICAL_NETWORK] = None
         network[provider.SEGMENTATION_ID] = None
     elif len(segments) > 1:
         network[provider.NETWORK_TYPE] = TYPE_MULTI_SEGMENT
         network[provider.PHYSICAL_NETWORK] = None
         network[provider.SEGMENTATION_ID] = None
     else:
         segment = segments[0]
         network[provider.NETWORK_TYPE] = segment[api.NETWORK_TYPE]
         network[provider.PHYSICAL_NETWORK] = segment[api.PHYSICAL_NETWORK]
         network[provider.SEGMENTATION_ID] = segment[api.SEGMENTATION_ID]

 def extend_network_dict_provider(self, context, network):
     id = network['id']
     segments = db.get_network_segments(context.session, id)
     if not segments:
         LOG.error(_LE("Network %s has no segments"), id)
         for attr in provider.ATTRIBUTES:
             network[attr] = None
     elif len(segments) > 1:
         network[mpnet.SEGMENTS] = [
             {provider.NETWORK_TYPE: segment[api.NETWORK_TYPE],
              provider.PHYSICAL_NETWORK: segment[api.PHYSICAL_NETWORK],
              provider.SEGMENTATION_ID: segment[api.SEGMENTATION_ID]}
             for segment in segments]
     else:
         segment = segments[0]
         network[provider.NETWORK_TYPE] = segment[api.NETWORK_TYPE]
         network[provider.PHYSICAL_NETWORK] = segment[api.PHYSICAL_NETWORK]
         network[provider.SEGMENTATION_ID] = segment[api.SEGMENTATION_ID]

 def external_dns_not_needed(self, context, network):
     if not DNS_DRIVER:
         return True
     if network['router:external']:
         return True
     segments = db.get_network_segments(context.session, network['id'])
     if len(segments) > 1:
         return False
     provider_net = segments[0]
     if provider_net['network_type'] == 'local':
         return True
     if provider_net['network_type'] == 'flat':
         return False
     if provider_net['network_type'] == 'vlan':
         return self._is_vlan_tenant_network(provider_net)
     if provider_net['network_type'] in ['gre', 'vxlan', 'geneve']:
         return self._is_tunnel_tenant_network(provider_net)
     return True

 def external_dns_not_needed(self, context, network):
     dns_driver = _get_dns_driver()
     if not dns_driver:
         return True
     if network["router:external"]:
         return True
     segments = db.get_network_segments(context.session, network["id"])
     if len(segments) > 1:
         return False
     provider_net = segments[0]
     if provider_net["network_type"] == "local":
         return True
     if provider_net["network_type"] == "flat":
         return False
     if provider_net["network_type"] == "vlan":
         return self._is_vlan_tenant_network(provider_net)
     if provider_net["network_type"] in ["gre", "vxlan", "geneve"]:
         return self._is_tunnel_tenant_network(provider_net)
     return True

    def get_network_segment(self, context, agent_configuration, network):
        data = None

        network_segment_physical_network = \
            agent_configuration.get('network_segment_physical_network', None)

        supported_encapsulations = [
            x.lower() for x in self.supported_encapsulations +
            agent_configuration.get('tunnel_types', [])
        ]
        # look up segment details in the ml2_network_segments table
        segments = db.get_network_segments(context.session, network['id'],
                                           filter_dynamic=None)

        for segment in segments:
            LOG.debug("F5 disconnected service check segment: %s" % segment)
            if ((network_segment_physical_network ==
                 segment['physical_network']) and
                (segment['network_type'].lower() in
                 supported_encapsulations)):
                data = segment
                break
            elif (network['provider:network_type'] == 'opflex' and
                  segment['network_type'] == 'vlan'):
                data = segment
                LOG.debug("Got OPFLEX segment: %s" % segment)
                break

        if not data:
            LOG.debug('Using default segment for network %s' %
                      (network['id']))

            # neutron is expected to provide this data immediately
            data = {
                'segmentation_id': network['provider:segmentation_id']
            }
            if 'provider:network_type' in network:
                data['network_type'] = network['provider:network_type']
            if 'provider:physical_network' in network:
                data['physical_network'] = network['provider:physical_network']

        return data

    def get_device_details(self, rpc_context, **kwargs):
        """Agent requests device details."""
        agent_id = kwargs.get('agent_id')
        device = kwargs.get('device')
        LOG.debug(_("Device %(device)s details requested by agent "
                    "%(agent_id)s"),
                  {'device': device, 'agent_id': agent_id})
        port_id = self._device_to_port_id(device)

        session = db_api.get_session()
        with session.begin(subtransactions=True):
            port = db.get_port(session, port_id)
            if not port:
                LOG.warning(_("Device %(device)s requested by agent "
                              "%(agent_id)s not found in database"),
                            {'device': device, 'agent_id': agent_id})
                return {'device': device}
            segments = db.get_network_segments(session, port.network_id)
            if not segments:
                LOG.warning(_("Device %(device)s requested by agent "
                              "%(agent_id)s has network %(network_id) with "
                              "no segments"),
                            {'device': device,
                             'agent_id': agent_id,
                             'network_id': port.network_id})
                return {'device': device}
            #TODO(rkukura): Use/create port binding
            segment = segments[0]
            new_status = (q_const.PORT_STATUS_ACTIVE if port.admin_state_up
                          else q_const.PORT_STATUS_DOWN)
            if port.status != new_status:
                port.status = new_status
            entry = {'device': device,
                     'network_id': port.network_id,
                     'port_id': port.id,
                     'admin_state_up': port.admin_state_up,
                     'network_type': segment[api.NETWORK_TYPE],
                     'segmentation_id': segment[api.SEGMENTATION_ID],
                     'physical_network': segment[api.PHYSICAL_NETWORK]}
            LOG.debug(_("Returning: %s"), entry)
            return entry

 def get_network_segment(self, context, agent_configuration, network):
     data = None
     network_segment_physical_network = \
         agent_configuration.get('network_segment_physical_network', None)
     if network_segment_physical_network:
         # look up segment details in the ml2_network_segments table
         segments = db.get_network_segments(context.session, network['id'])
         for segment in segments:
             if (network_segment_physical_network ==
                     segment['physical_network']):
                 data = segment
                 break
         if not data:
             LOG.error('network_id %s does not match physical_network %s' %
                       (network['id'], network_segment_physical_network))
     else:
         # neutron is expected to provide this data immediately
         data = {
             'segmentation_id': network['provider:segmentation_id']
         }
         if 'provider:network_type' in network:
             data['network_type'] = network['provider:network_type']
     return data