neutron/tests/unit/plugins/ml2/test_db.py

   1 # Copyright (c) 2014 OpenStack Foundation, all rights reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import mock
  17 from oslo_utils import uuidutils
  18 from sqlalchemy.orm import query
  19
  20 from neutron import context
  21 from neutron.db import db_base_plugin_v2
  22 from neutron.db import l3_db
  23 from neutron.db import models_v2
  24 from neutron.extensions import portbindings
  25 from neutron.plugins.ml2 import db as ml2_db
  26 from neutron.plugins.ml2 import driver_api as api
  27 from neutron.plugins.ml2 import models
  28 from neutron.tests.unit import testlib_api
  29
  30
  31 class Ml2DBTestCase(testlib_api.SqlTestCase):
  32
  33     def setUp(self):
  34         super(Ml2DBTestCase, self).setUp()
  35         self.ctx = context.get_admin_context()
  36
  37     def _setup_neutron_network(self, network_id):
  38         with self.ctx.session.begin(subtransactions=True):
  39             self.ctx.session.add(models_v2.Network(id=network_id))
  40
  41     def _setup_neutron_port(self, network_id, port_id):
  42         mac_address = db_base_plugin_v2.NeutronDbPluginV2._generate_mac()
  43         with self.ctx.session.begin(subtransactions=True):
  44             port = models_v2.Port(id=port_id,
  45                                   network_id=network_id,
  46                                   mac_address=mac_address,
  47                                   admin_state_up=True,
  48                                   status='DOWN',
  49                                   device_id='',
  50                                   device_owner='')
  51             self.ctx.session.add(port)
  52         return port
  53
  54     def _setup_neutron_portbinding(self, port_id, vif_type, host):
  55         with self.ctx.session.begin(subtransactions=True):
  56             self.ctx.session.add(models.PortBinding(port_id=port_id,
  57                                                     vif_type=vif_type,
  58                                                     host=host))
  59
  60     def _create_segments(self, segments, is_seg_dynamic=False,
  61                          network_id='foo-network-id'):
  62         self._setup_neutron_network(network_id)
  63         for segment in segments:
  64             ml2_db.add_network_segment(
  65                 self.ctx.session, network_id, segment,
  66                 is_dynamic=is_seg_dynamic)
  67
  68         net_segments = ml2_db.get_network_segments(
  69                            self.ctx.session, network_id,
  70                            filter_dynamic=is_seg_dynamic)
  71
  72         for segment_index, segment in enumerate(segments):
  73             self.assertEqual(segment, net_segments[segment_index])
  74
  75         return net_segments
  76
  77     def test_network_segments_for_provider_network(self):
  78         segment = {api.NETWORK_TYPE: 'vlan',
  79                    api.PHYSICAL_NETWORK: 'physnet1',
  80                    api.SEGMENTATION_ID: 1}
  81         self._create_segments([segment])
  82
  83     def test_network_segments_is_dynamic_true(self):
  84         segment = {api.NETWORK_TYPE: 'vlan',
  85                    api.PHYSICAL_NETWORK: 'physnet1',
  86                    api.SEGMENTATION_ID: 1}
  87         self._create_segments([segment], is_seg_dynamic=True)
  88
  89     def test_network_segments_for_multiprovider_network(self):
  90         segments = [{api.NETWORK_TYPE: 'vlan',
  91                     api.PHYSICAL_NETWORK: 'physnet1',
  92                     api.SEGMENTATION_ID: 1},
  93                     {api.NETWORK_TYPE: 'vlan',
  94                      api.PHYSICAL_NETWORK: 'physnet1',
  95                      api.SEGMENTATION_ID: 2}]
  96         self._create_segments(segments)
  97
  98     def test_get_networks_segments(self):
  99         segments1 = [{api.NETWORK_TYPE: 'vlan',
 100                       api.PHYSICAL_NETWORK: 'physnet1',
 101                       api.SEGMENTATION_ID: 1},
 102                      {api.NETWORK_TYPE: 'vlan',
 103                       api.PHYSICAL_NETWORK: 'physnet1',
 104                       api.SEGMENTATION_ID: 2}]
 105         segments2 = [{api.NETWORK_TYPE: 'vlan',
 106                       api.PHYSICAL_NETWORK: 'physnet1',
 107                       api.SEGMENTATION_ID: 3},
 108                      {api.NETWORK_TYPE: 'vlan',
 109                       api.PHYSICAL_NETWORK: 'physnet1',
 110                       api.SEGMENTATION_ID: 4}]
 111         net1segs = self._create_segments(segments1, network_id='net1')
 112         net2segs = self._create_segments(segments2, network_id='net2')
 113         segs = ml2_db.get_networks_segments(self.ctx.session, ['net1', 'net2'])
 114         self.assertEqual(net1segs, segs['net1'])
 115         self.assertEqual(net2segs, segs['net2'])
 116
 117     def test_get_networks_segments_no_segments(self):
 118         self._create_segments([], network_id='net1')
 119         self._create_segments([], network_id='net2')
 120         segs = ml2_db.get_networks_segments(self.ctx.session, ['net1', 'net2'])
 121         self.assertEqual([], segs['net1'])
 122         self.assertEqual([], segs['net2'])
 123
 124     def test_get_segment_by_id(self):
 125         segment = {api.NETWORK_TYPE: 'vlan',
 126                    api.PHYSICAL_NETWORK: 'physnet1',
 127                    api.SEGMENTATION_ID: 1}
 128
 129         net_segment = self._create_segments([segment])[0]
 130         segment_uuid = net_segment[api.ID]
 131
 132         net_segment = ml2_db.get_segment_by_id(self.ctx.session, segment_uuid)
 133         self.assertEqual(segment, net_segment)
 134
 135     def test_get_segment_by_id_result_not_found(self):
 136         segment_uuid = uuidutils.generate_uuid()
 137         net_segment = ml2_db.get_segment_by_id(self.ctx.session, segment_uuid)
 138         self.assertIsNone(net_segment)
 139
 140     def test_delete_network_segment(self):
 141         segment = {api.NETWORK_TYPE: 'vlan',
 142                    api.PHYSICAL_NETWORK: 'physnet1',
 143                    api.SEGMENTATION_ID: 1}
 144
 145         net_segment = self._create_segments([segment])[0]
 146         segment_uuid = net_segment[api.ID]
 147
 148         ml2_db.delete_network_segment(self.ctx.session, segment_uuid)
 149         # Get segment and verify its empty
 150         net_segment = ml2_db.get_segment_by_id(self.ctx.session, segment_uuid)
 151         self.assertIsNone(net_segment)
 152
 153     def test_add_port_binding(self):
 154         network_id = 'foo-network-id'
 155         port_id = 'foo-port-id'
 156         self._setup_neutron_network(network_id)
 157         self._setup_neutron_port(network_id, port_id)
 158
 159         port = ml2_db.add_port_binding(self.ctx.session, port_id)
 160         self.assertEqual(port_id, port.port_id)
 161         self.assertEqual(portbindings.VIF_TYPE_UNBOUND, port.vif_type)
 162
 163     def test_get_port_binding_host(self):
 164         network_id = 'foo-network-id'
 165         port_id = 'foo-port-id'
 166         host = 'fake_host'
 167         vif_type = portbindings.VIF_TYPE_UNBOUND
 168         self._setup_neutron_network(network_id)
 169         self._setup_neutron_port(network_id, port_id)
 170         self._setup_neutron_portbinding(port_id, vif_type, host)
 171
 172         port_host = ml2_db.get_port_binding_host(self.ctx.session, port_id)
 173         self.assertEqual(host, port_host)
 174
 175     def test_get_port_binding_host_multiple_results_found(self):
 176         network_id = 'foo-network-id'
 177         port_id = 'foo-port-id'
 178         port_id_one = 'foo-port-id-one'
 179         port_id_two = 'foo-port-id-two'
 180         host = 'fake_host'
 181         vif_type = portbindings.VIF_TYPE_UNBOUND
 182         self._setup_neutron_network(network_id)
 183         self._setup_neutron_port(network_id, port_id_one)
 184         self._setup_neutron_portbinding(port_id_one, vif_type, host)
 185         self._setup_neutron_port(network_id, port_id_two)
 186         self._setup_neutron_portbinding(port_id_two, vif_type, host)
 187
 188         port_host = ml2_db.get_port_binding_host(self.ctx.session, port_id)
 189         self.assertIsNone(port_host)
 190
 191     def test_get_port_binding_host_result_not_found(self):
 192         port_id = uuidutils.generate_uuid()
 193
 194         port_host = ml2_db.get_port_binding_host(self.ctx.session, port_id)
 195         self.assertIsNone(port_host)
 196
 197     def test_get_port(self):
 198         network_id = 'foo-network-id'
 199         port_id = 'foo-port-id'
 200         self._setup_neutron_network(network_id)
 201         self._setup_neutron_port(network_id, port_id)
 202
 203         port = ml2_db.get_port(self.ctx.session, port_id)
 204         self.assertEqual(port_id, port.id)
 205
 206     def test_get_port_multiple_results_found(self):
 207         network_id = 'foo-network-id'
 208         port_id = 'foo-port-id'
 209         port_id_one = 'foo-port-id-one'
 210         port_id_two = 'foo-port-id-two'
 211         self._setup_neutron_network(network_id)
 212         self._setup_neutron_port(network_id, port_id_one)
 213         self._setup_neutron_port(network_id, port_id_two)
 214
 215         port = ml2_db.get_port(self.ctx.session, port_id)
 216         self.assertIsNone(port)
 217
 218     def test_get_port_result_not_found(self):
 219         port_id = uuidutils.generate_uuid()
 220         port = ml2_db.get_port(self.ctx.session, port_id)
 221         self.assertIsNone(port)
 222
 223     def test_get_port_from_device_mac(self):
 224         network_id = 'foo-network-id'
 225         port_id = 'foo-port-id'
 226         self._setup_neutron_network(network_id)
 227         port = self._setup_neutron_port(network_id, port_id)
 228
 229         observed_port = ml2_db.get_port_from_device_mac(self.ctx,
 230                                                         port['mac_address'])
 231         self.assertEqual(port_id, observed_port.id)
 232
 233     def test_get_locked_port_and_binding(self):
 234         network_id = 'foo-network-id'
 235         port_id = 'foo-port-id'
 236         host = 'fake_host'
 237         vif_type = portbindings.VIF_TYPE_UNBOUND
 238         self._setup_neutron_network(network_id)
 239         self._setup_neutron_port(network_id, port_id)
 240         self._setup_neutron_portbinding(port_id, vif_type, host)
 241
 242         port, binding = ml2_db.get_locked_port_and_binding(self.ctx.session,
 243                                                            port_id)
 244         self.assertEqual(port_id, port.id)
 245         self.assertEqual(port_id, binding.port_id)
 246
 247     def test_get_locked_port_and_binding_result_not_found(self):
 248         port_id = uuidutils.generate_uuid()
 249
 250         port, binding = ml2_db.get_locked_port_and_binding(self.ctx.session,
 251                                                            port_id)
 252         self.assertIsNone(port)
 253         self.assertIsNone(binding)
 254
 255
 256 class Ml2DvrDBTestCase(testlib_api.SqlTestCase):
 257
 258     def setUp(self):
 259         super(Ml2DvrDBTestCase, self).setUp()
 260         self.ctx = context.get_admin_context()
 261
 262     def _setup_neutron_network(self, network_id, port_ids):
 263         with self.ctx.session.begin(subtransactions=True):
 264             self.ctx.session.add(models_v2.Network(id=network_id))
 265             ports = []
 266             for port_id in port_ids:
 267                 mac_address = (db_base_plugin_v2.NeutronDbPluginV2.
 268                                _generate_mac())
 269                 port = models_v2.Port(id=port_id,
 270                                       network_id=network_id,
 271                                       mac_address=mac_address,
 272                                       admin_state_up=True,
 273                                       status='ACTIVE',
 274                                       device_id='',
 275                                       device_owner='')
 276                 self.ctx.session.add(port)
 277                 ports.append(port)
 278             return ports
 279
 280     def _setup_neutron_router(self):
 281         with self.ctx.session.begin(subtransactions=True):
 282             router = l3_db.Router()
 283             self.ctx.session.add(router)
 284             return router
 285
 286     def _setup_dvr_binding(self, network_id, port_id, router_id, host_id):
 287         with self.ctx.session.begin(subtransactions=True):
 288             record = models.DVRPortBinding(
 289                 port_id=port_id,
 290                 host=host_id,
 291                 router_id=router_id,
 292                 vif_type=portbindings.VIF_TYPE_UNBOUND,
 293                 vnic_type=portbindings.VNIC_NORMAL,
 294                 status='DOWN')
 295             self.ctx.session.add(record)
 296             return record
 297
 298     def test_ensure_dvr_port_binding_deals_with_db_duplicate(self):
 299         network_id = 'foo_network_id'
 300         port_id = 'foo_port_id'
 301         router_id = 'foo_router_id'
 302         host_id = 'foo_host_id'
 303         self._setup_neutron_network(network_id, [port_id])
 304         self._setup_dvr_binding(network_id, port_id, router_id, host_id)
 305         with mock.patch.object(query.Query, 'first') as query_first:
 306             query_first.return_value = []
 307             with mock.patch.object(ml2_db.LOG, 'debug') as log_trace:
 308                 binding = ml2_db.ensure_dvr_port_binding(
 309                     self.ctx.session, port_id, host_id, router_id)
 310         self.assertTrue(query_first.called)
 311         self.assertTrue(log_trace.called)
 312         self.assertEqual(port_id, binding.port_id)
 313
 314     def test_ensure_dvr_port_binding(self):
 315         network_id = 'foo_network_id'
 316         port_id = 'foo_port_id'
 317         self._setup_neutron_network(network_id, [port_id])
 318         router = self._setup_neutron_router()
 319         ml2_db.ensure_dvr_port_binding(
 320             self.ctx.session, port_id, 'foo_host', router.id)
 321         expected = (self.ctx.session.query(models.DVRPortBinding).
 322                     filter_by(port_id=port_id).one())
 323         self.assertEqual(expected.port_id, port_id)
 324
 325     def test_ensure_dvr_port_binding_multiple_bindings(self):
 326         network_id = 'foo_network_id'
 327         port_id = 'foo_port_id'
 328         self._setup_neutron_network(network_id, [port_id])
 329         router = self._setup_neutron_router()
 330         ml2_db.ensure_dvr_port_binding(
 331             self.ctx.session, port_id, 'foo_host_1', router.id)
 332         ml2_db.ensure_dvr_port_binding(
 333             self.ctx.session, port_id, 'foo_host_2', router.id)
 334         bindings = (self.ctx.session.query(models.DVRPortBinding).
 335                     filter_by(port_id=port_id).all())
 336         self.assertEqual(2, len(bindings))
 337
 338     def test_delete_dvr_port_binding(self):
 339         network_id = 'foo_network_id'
 340         port_id = 'foo_port_id'
 341         self._setup_neutron_network(network_id, [port_id])
 342         router = self._setup_neutron_router()
 343         binding = self._setup_dvr_binding(
 344             network_id, port_id, router.id, 'foo_host_id')
 345         ml2_db.delete_dvr_port_binding(
 346             self.ctx.session, port_id, 'foo_host_id')
 347         count = (self.ctx.session.query(models.DVRPortBinding).
 348             filter_by(port_id=binding.port_id).count())
 349         self.assertFalse(count)
 350
 351     def test_delete_dvr_port_binding_not_found(self):
 352         ml2_db.delete_dvr_port_binding(
 353             self.ctx.session, 'foo_port_id', 'foo_host')
 354
 355     def test_delete_dvr_port_binding_if_stale(self):
 356         network_id = 'foo_network_id'
 357         port_id = 'foo_port_id'
 358         self._setup_neutron_network(network_id, [port_id])
 359         binding = self._setup_dvr_binding(
 360             network_id, port_id, None, 'foo_host_id')
 361
 362         ml2_db.delete_dvr_port_binding_if_stale(self.ctx.session, binding)
 363         count = (self.ctx.session.query(models.DVRPortBinding).
 364             filter_by(port_id=binding.port_id).count())
 365         self.assertFalse(count)
 366
 367     def test_get_dvr_port_binding_by_host_not_found(self):
 368         port = ml2_db.get_dvr_port_binding_by_host(
 369             self.ctx.session, 'foo_port_id', 'foo_host_id')
 370         self.assertIsNone(port)
 371
 372     def test_get_dvr_port_bindings_not_found(self):
 373         port = ml2_db.get_dvr_port_bindings(self.ctx.session, 'foo_port_id')
 374         self.assertFalse(len(port))
 375
 376     def test_get_dvr_port_bindings(self):
 377         network_id = 'foo_network_id'
 378         port_id_1 = 'foo_port_id_1'
 379         port_id_2 = 'foo_port_id_2'
 380         self._setup_neutron_network(network_id, [port_id_1, port_id_2])
 381         router = self._setup_neutron_router()
 382         self._setup_dvr_binding(
 383             network_id, port_id_1, router.id, 'foo_host_id_1')
 384         self._setup_dvr_binding(
 385             network_id, port_id_1, router.id, 'foo_host_id_2')
 386         ports = ml2_db.get_dvr_port_bindings(self.ctx.session, 'foo_port_id')
 387         self.assertEqual(2, len(ports))