x/vmware-nsx
Code Issues Proposed changes
vmware-nsx/neutron/tests/unit/cisco/n1kv/test_n1kv_db.py
Abhishek Raut 2f1cd3eb55 Add support for the Nexus 1000V into the Cisco Plugin. 
This will enable the Cisco Nexus 1000V to integrate with the Cisco plugin
and be used to drive the realization of Neutron constructs.
Network profile and Policy profile are introduced as extended neutron
resources, while n1kv:profile_id is introduced as an extended attribute
for network and port objects. Necessary changes to the Cisco plugin are
made to accomodate Nexus 1000V as a configurable vswitch plugin.

Implements: blueprint cisco-plugin-n1k-support
Change-Id: I951e10c57d74c935fca8754c0e21e1ac9df35704
2013-08-09 16:56:54 -07:00

673 lines
31 KiB
Python
Raw Blame History

# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cisco Systems, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
# @author: Juergen Brendel, Cisco Systems Inc.
# @author: Abhishek Raut, Cisco Systems Inc.
from sqlalchemy.orm import exc as s_exc
from testtools import matchers
from neutron.common import exceptions as q_exc
from neutron import context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.plugins.cisco.common import cisco_constants
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.db import n1kv_db_v2
from neutron.plugins.cisco.db import n1kv_models_v2
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
PHYS_NET = 'physnet1'
PHYS_NET_2 = 'physnet2'
VLAN_MIN = 10
VLAN_MAX = 19
VLAN_RANGES = {PHYS_NET: [(VLAN_MIN, VLAN_MAX)]}
UPDATED_VLAN_RANGES = {PHYS_NET: [(VLAN_MIN + 20, VLAN_MAX + 20)],
PHYS_NET_2: [(VLAN_MIN + 40, VLAN_MAX + 40)]}
VXLAN_MIN = 5000
VXLAN_MAX = 5009
VXLAN_RANGES = [(VXLAN_MIN, VXLAN_MAX)]
UPDATED_VXLAN_RANGES = [(VXLAN_MIN + 20, VXLAN_MAX + 20)]
SEGMENT_RANGE = '200-220'
SEGMENT_RANGE_MIN_OVERLAP = '210-230'
SEGMENT_RANGE_MAX_OVERLAP = '190-209'
SEGMENT_RANGE_OVERLAP = '190-230'
TEST_NETWORK_ID = 'abcdefghijklmnopqrstuvwxyz'
TEST_NETWORK_PROFILE = {'name': 'test_profile',
'segment_type': 'vlan',
'physical_network': 'physnet1',
'segment_range': '10-19'}
TEST_NETWORK_PROFILE_2 = {'name': 'test_profile_2',
'segment_type': 'vlan',
'physical_network': 'physnet1',
'segment_range': SEGMENT_RANGE}
TEST_NETWORK_PROFILE_VXLAN = {'name': 'test_profile',
'segment_type': 'vxlan',
'segment_range': '5000-5009',
'multicast_ip_range': '239.0.0.70-239.0.0.80'}
TEST_POLICY_PROFILE = {'id': '4a417990-76fb-11e2-bcfd-0800200c9a66',
'name': 'test_policy_profile'}
def _create_test_network_profile_if_not_there(session,
profile=TEST_NETWORK_PROFILE):
try:
_profile = session.query(n1kv_models_v2.NetworkProfile).filter_by(
name=profile['name']).one()
except s_exc.NoResultFound:
_profile = n1kv_db_v2.create_network_profile(session, profile)
return _profile
def _create_test_policy_profile_if_not_there(session,
profile=TEST_POLICY_PROFILE):
try:
_profile = session.query(n1kv_models_v2.PolicyProfile).filter_by(
name=profile['name']).one()
except s_exc.NoResultFound:
_profile = n1kv_db_v2.create_policy_profile(profile)
return _profile
class VlanAllocationsTest(base.BaseTestCase):
def setUp(self):
super(VlanAllocationsTest, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
n1kv_db_v2.sync_vlan_allocations(self.session, VLAN_RANGES)
def tearDown(self):
super(VlanAllocationsTest, self).tearDown()
def test_sync_vlan_allocations_outside_segment_range(self):
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
VLAN_MIN - 1)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
VLAN_MAX + 1)
n1kv_db_v2.sync_vlan_allocations(self.session, UPDATED_VLAN_RANGES)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
VLAN_MIN + 20 - 1)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
VLAN_MAX + 20 + 1)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET_2,
VLAN_MIN + 40 - 1)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET_2,
VLAN_MAX + 40 + 1)
n1kv_db_v2.sync_vlan_allocations(self.session, VLAN_RANGES)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET_2,
VLAN_MIN + 20)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET_2,
VLAN_MIN + 20)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET_2,
VLAN_MAX + 20)
def test_sync_vlan_allocations_unallocated_vlans(self):
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MIN).allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MIN + 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MAX - 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MAX).allocated)
n1kv_db_v2.sync_vlan_allocations(self.session, UPDATED_VLAN_RANGES)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MIN + 20).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MIN + 20 + 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
VLAN_MAX + 20 - 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session, PHYS_NET,
VLAN_MAX + 20).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET_2,
VLAN_MIN + 40).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET_2,
VLAN_MIN + 40 + 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET_2,
VLAN_MAX + 40 - 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET_2,
VLAN_MAX + 40).
allocated)
def test_vlan_pool(self):
vlan_ids = set()
p = _create_test_network_profile_if_not_there(self.session)
for x in xrange(VLAN_MIN, VLAN_MAX + 1):
(physical_network, seg_type,
vlan_id, m_ip) = n1kv_db_v2.reserve_vlan(self.session, p)
self.assertEqual(physical_network, PHYS_NET)
self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1))
self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1))
vlan_ids.add(vlan_id)
self.assertRaises(q_exc.NoNetworkAvailable,
n1kv_db_v2.reserve_vlan,
self.session,
p)
n1kv_db_v2.release_vlan(self.session, PHYS_NET, vlan_ids.pop(),
VLAN_RANGES)
physical_network, seg_type, vlan_id, m_ip = (n1kv_db_v2.reserve_vlan(
self.session, p))
self.assertEqual(physical_network, PHYS_NET)
self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1))
self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1))
vlan_ids.add(vlan_id)
for vlan_id in vlan_ids:
n1kv_db_v2.release_vlan(self.session, PHYS_NET, vlan_id,
VLAN_RANGES)
def test_specific_vlan_inside_pool(self):
vlan_id = VLAN_MIN + 5
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
vlan_id).allocated)
n1kv_db_v2.reserve_specific_vlan(self.session, PHYS_NET, vlan_id)
self.assertTrue(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
vlan_id).allocated)
self.assertRaises(q_exc.VlanIdInUse,
n1kv_db_v2.reserve_specific_vlan,
self.session,
PHYS_NET,
vlan_id)
n1kv_db_v2.release_vlan(self.session, PHYS_NET, vlan_id, VLAN_RANGES)
self.assertFalse(n1kv_db_v2.get_vlan_allocation(self.session,
PHYS_NET,
vlan_id).allocated)
def test_specific_vlan_outside_pool(self):
vlan_id = VLAN_MAX + 5
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
vlan_id)
n1kv_db_v2.reserve_specific_vlan(self.session, PHYS_NET, vlan_id)
self.assertTrue(n1kv_db_v2.get_vlan_allocation(self.session, PHYS_NET,
vlan_id).allocated)
self.assertRaises(q_exc.VlanIdInUse,
n1kv_db_v2.reserve_specific_vlan,
self.session,
PHYS_NET,
vlan_id)
n1kv_db_v2.release_vlan(self.session, PHYS_NET, vlan_id, VLAN_RANGES)
self.assertRaises(c_exc.VlanIDNotFound,
n1kv_db_v2.get_vlan_allocation,
self.session,
PHYS_NET,
vlan_id)
class VxlanAllocationsTest(base.BaseTestCase,
n1kv_db_v2.NetworkProfile_db_mixin):
def setUp(self):
super(VxlanAllocationsTest, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
n1kv_db_v2.sync_vxlan_allocations(self.session, VXLAN_RANGES)
def tearDown(self):
super(VxlanAllocationsTest, self).tearDown()
def test_sync_vxlan_allocations_outside_segment_range(self):
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN - 1))
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX + 1))
n1kv_db_v2.sync_vxlan_allocations(self.session, UPDATED_VXLAN_RANGES)
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN + 20 - 1))
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX + 20 + 1))
def test_sync_vxlan_allocations_unallocated_vxlans(self):
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN).allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN + 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX - 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX).allocated)
n1kv_db_v2.sync_vxlan_allocations(self.session, UPDATED_VXLAN_RANGES)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN + 20).
allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MIN + 20 + 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX + 20 - 1).
allocated)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
VXLAN_MAX + 20).
allocated)
def test_vxlan_pool(self):
vxlan_ids = set()
profile = n1kv_db_v2.create_network_profile(self.session,
TEST_NETWORK_PROFILE_VXLAN)
for x in xrange(VXLAN_MIN, VXLAN_MAX + 1):
vxlan = n1kv_db_v2.reserve_vxlan(self.session, profile)
vxlan_id = vxlan[2]
self.assertThat(vxlan_id, matchers.GreaterThan(VXLAN_MIN - 1))
self.assertThat(vxlan_id, matchers.LessThan(VXLAN_MAX + 1))
vxlan_ids.add(vxlan_id)
self.assertRaises(q_exc.NoNetworkAvailable,
n1kv_db_v2.reserve_vxlan,
self.session,
profile)
n1kv_db_v2.release_vxlan(self.session, vxlan_ids.pop(), VXLAN_RANGES)
vxlan = n1kv_db_v2.reserve_vxlan(self.session, profile)
vxlan_id = vxlan[2]
self.assertThat(vxlan_id, matchers.GreaterThan(VXLAN_MIN - 1))
self.assertThat(vxlan_id, matchers.LessThan(VXLAN_MAX + 1))
vxlan_ids.add(vxlan_id)
for vxlan_id in vxlan_ids:
n1kv_db_v2.release_vxlan(self.session, vxlan_id, VXLAN_RANGES)
n1kv_db_v2.delete_network_profile(self.session, profile.id)
def test_specific_vxlan_inside_pool(self):
vxlan_id = VXLAN_MIN + 5
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id).allocated)
n1kv_db_v2.reserve_specific_vxlan(self.session, vxlan_id)
self.assertTrue(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id).allocated)
self.assertRaises(c_exc.VxlanIdInUse,
n1kv_db_v2.reserve_specific_vxlan,
self.session,
vxlan_id)
n1kv_db_v2.release_vxlan(self.session, vxlan_id, VXLAN_RANGES)
self.assertFalse(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id).allocated)
def test_specific_vxlan_outside_pool(self):
vxlan_id = VXLAN_MAX + 5
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id))
n1kv_db_v2.reserve_specific_vxlan(self.session, vxlan_id)
self.assertTrue(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id).allocated)
self.assertRaises(c_exc.VxlanIdInUse,
n1kv_db_v2.reserve_specific_vxlan,
self.session,
vxlan_id)
n1kv_db_v2.release_vxlan(self.session, vxlan_id, VXLAN_RANGES)
self.assertIsNone(n1kv_db_v2.get_vxlan_allocation(self.session,
vxlan_id))
class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
super(NetworkBindingsTest, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
def tearDown(self):
super(NetworkBindingsTest, self).tearDown()
def test_add_network_binding(self):
with self.network() as network:
TEST_NETWORK_ID = network['network']['id']
self.assertRaises(c_exc.NetworkBindingNotFound,
n1kv_db_v2.get_network_binding,
self.session,
TEST_NETWORK_ID)
p = _create_test_network_profile_if_not_there(self.session)
n1kv_db_v2.add_network_binding(
self.session, TEST_NETWORK_ID, 'vlan',
PHYS_NET, 1234, '0.0.0.0', p.id)
binding = n1kv_db_v2.get_network_binding(
self.session, TEST_NETWORK_ID)
self.assertIsNotNone(binding)
self.assertEqual(binding.network_id, TEST_NETWORK_ID)
self.assertEqual(binding.network_type, 'vlan')
self.assertEqual(binding.physical_network, PHYS_NET)
self.assertEqual(binding.segmentation_id, 1234)
class NetworkProfileTests(base.BaseTestCase,
n1kv_db_v2.NetworkProfile_db_mixin):
def setUp(self):
super(NetworkProfileTests, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
def tearDown(self):
super(NetworkProfileTests, self).tearDown()
def test_create_network_profile(self):
_db_profile = n1kv_db_v2.create_network_profile(self.session,
TEST_NETWORK_PROFILE)
self.assertIsNotNone(_db_profile)
db_profile = (self.session.query(n1kv_models_v2.NetworkProfile).
filter_by(name=TEST_NETWORK_PROFILE['name']).one())
self.assertIsNotNone(db_profile)
self.assertEqual(_db_profile.id, db_profile.id)
self.assertEqual(_db_profile.name, db_profile.name)
self.assertEqual(_db_profile.segment_type, db_profile.segment_type)
self.assertEqual(_db_profile.segment_range, db_profile.segment_range)
self.assertEqual(_db_profile.multicast_ip_index,
db_profile.multicast_ip_index)
self.assertEqual(_db_profile.multicast_ip_range,
db_profile.multicast_ip_range)
n1kv_db_v2.delete_network_profile(self.session, _db_profile.id)
def test_create_network_profile_overlap(self):
_db_profile = n1kv_db_v2.create_network_profile(self.session,
TEST_NETWORK_PROFILE_2)
ctx = context.get_admin_context()
TEST_NETWORK_PROFILE_2['name'] = 'net-profile-min-overlap'
TEST_NETWORK_PROFILE_2['segment_range'] = SEGMENT_RANGE_MIN_OVERLAP
test_net_profile = {'network_profile': TEST_NETWORK_PROFILE_2}
self.assertRaises(q_exc.InvalidInput,
self.create_network_profile,
ctx,
test_net_profile)
TEST_NETWORK_PROFILE_2['name'] = 'net-profile-max-overlap'
TEST_NETWORK_PROFILE_2['segment_range'] = SEGMENT_RANGE_MAX_OVERLAP
test_net_profile = {'network_profile': TEST_NETWORK_PROFILE_2}
self.assertRaises(q_exc.InvalidInput,
self.create_network_profile,
ctx,
test_net_profile)
TEST_NETWORK_PROFILE_2['name'] = 'net-profile-overlap'
TEST_NETWORK_PROFILE_2['segment_range'] = SEGMENT_RANGE_OVERLAP
test_net_profile = {'network_profile': TEST_NETWORK_PROFILE_2}
self.assertRaises(q_exc.InvalidInput,
self.create_network_profile,
ctx,
test_net_profile)
n1kv_db_v2.delete_network_profile(self.session, _db_profile.id)
def test_delete_network_profile(self):
try:
profile = (self.session.query(n1kv_models_v2.NetworkProfile).
filter_by(name=TEST_NETWORK_PROFILE['name']).one())
except s_exc.NoResultFound:
profile = n1kv_db_v2.create_network_profile(self.session,
TEST_NETWORK_PROFILE)
n1kv_db_v2.delete_network_profile(self.session, profile.id)
try:
self.session.query(n1kv_models_v2.NetworkProfile).filter_by(
name=TEST_NETWORK_PROFILE['name']).one()
except s_exc.NoResultFound:
pass
else:
self.fail("Network Profile (%s) was not deleted" %
TEST_NETWORK_PROFILE['name'])
def test_update_network_profile(self):
TEST_PROFILE_1 = {'name': 'test_profile_1'}
profile = _create_test_network_profile_if_not_there(self.session)
updated_profile = n1kv_db_v2.update_network_profile(self.session,
profile.id,
TEST_PROFILE_1)
self.assertEqual(updated_profile.name, TEST_PROFILE_1['name'])
n1kv_db_v2.delete_network_profile(self.session, profile.id)
def test_get_network_profile(self):
profile = n1kv_db_v2.create_network_profile(self.session,
TEST_NETWORK_PROFILE)
got_profile = n1kv_db_v2.get_network_profile(self.session, profile.id)
self.assertEqual(profile.id, got_profile.id)
self.assertEqual(profile.name, got_profile.name)
n1kv_db_v2.delete_network_profile(self.session, profile.id)
def test_get_network_profiles(self):
test_profiles = [{'name': 'test_profile1',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '200-210'},
{'name': 'test_profile2',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '211-220'},
{'name': 'test_profile3',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '221-230'},
{'name': 'test_profile4',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '231-240'},
{'name': 'test_profile5',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '241-250'},
{'name': 'test_profile6',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '251-260'},
{'name': 'test_profile7',
'segment_type': 'vlan',
'physical_network': 'phys1',
'segment_range': '261-270'}]
[n1kv_db_v2.create_network_profile(self.session, p)
for p in test_profiles]
# TODO(abhraut): Fix this test to work with real tenant_td
profiles = n1kv_db_v2._get_network_profiles()
self.assertEqual(len(test_profiles), len(list(profiles)))
class PolicyProfileTests(base.BaseTestCase):
def setUp(self):
super(PolicyProfileTests, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
def tearDown(self):
super(PolicyProfileTests, self).tearDown()
def test_create_policy_profile(self):
_db_profile = n1kv_db_v2.create_policy_profile(TEST_POLICY_PROFILE)
self.assertIsNotNone(_db_profile)
db_profile = (self.session.query(n1kv_models_v2.PolicyProfile).
filter_by(name=TEST_POLICY_PROFILE['name']).one)()
self.assertIsNotNone(db_profile)
self.assertTrue(_db_profile.id == db_profile.id)
self.assertTrue(_db_profile.name == db_profile.name)
def test_delete_policy_profile(self):
profile = _create_test_policy_profile_if_not_there(self.session)
n1kv_db_v2.delete_policy_profile(profile.id)
try:
self.session.query(n1kv_models_v2.PolicyProfile).filter_by(
name=TEST_POLICY_PROFILE['name']).one()
except s_exc.NoResultFound:
pass
else:
self.fail("Policy Profile (%s) was not deleted" %
TEST_POLICY_PROFILE['name'])
def test_update_policy_profile(self):
TEST_PROFILE_1 = {'name': 'test_profile_1'}
profile = _create_test_policy_profile_if_not_there(self.session)
updated_profile = n1kv_db_v2.update_policy_profile(self.session,
profile.id,
TEST_PROFILE_1)
self.assertEqual(updated_profile.name, TEST_PROFILE_1['name'])
def test_get_policy_profile(self):
profile = _create_test_policy_profile_if_not_there(self.session)
got_profile = n1kv_db_v2.get_policy_profile(self.session, profile.id)
self.assertEqual(profile.id, got_profile.id)
self.assertEqual(profile.name, got_profile.name)
class ProfileBindingTests(base.BaseTestCase,
n1kv_db_v2.NetworkProfile_db_mixin,
db_base_plugin_v2.CommonDbMixin):
def setUp(self):
super(ProfileBindingTests, self).setUp()
n1kv_db_v2.initialize()
self.session = db.get_session()
def tearDown(self):
super(ProfileBindingTests, self).tearDown()
def _create_test_binding_if_not_there(self, tenant_id, profile_id,
profile_type):
try:
_binding = (self.session.query(n1kv_models_v2.ProfileBinding).
filter_by(profile_type=profile_type,
tenant_id=tenant_id,
profile_id=profile_id).one())
except s_exc.NoResultFound:
_binding = n1kv_db_v2.create_profile_binding(tenant_id,
profile_id,
profile_type)
return _binding
def test_create_profile_binding(self):
test_tenant_id = "d434dd90-76ec-11e2-bcfd-0800200c9a66"
test_profile_id = "dd7b9741-76ec-11e2-bcfd-0800200c9a66"
test_profile_type = "network"
n1kv_db_v2.create_profile_binding(test_tenant_id, test_profile_id,
test_profile_type)
try:
self.session.query(n1kv_models_v2.ProfileBinding).filter_by(
profile_type=test_profile_type,
tenant_id=test_tenant_id,
profile_id=test_profile_id).one()
except s_exc.MultipleResultsFound:
self.fail("Bindings must be unique")
except s_exc.NoResultFound:
self.fail("Could not create Profile Binding")
def test_get_profile_binding(self):
test_tenant_id = "d434dd90-76ec-11e2-bcfd-0800200c9a66"
test_profile_id = "dd7b9741-76ec-11e2-bcfd-0800200c9a66"
test_profile_type = "network"
self._create_test_binding_if_not_there(test_tenant_id,
test_profile_id,
test_profile_type)
binding = n1kv_db_v2.get_profile_binding(test_tenant_id,
test_profile_id)
self.assertEqual(binding.tenant_id, test_tenant_id)
self.assertEqual(binding.profile_id, test_profile_id)
self.assertEqual(binding.profile_type, test_profile_type)
def test_delete_profile_binding(self):
test_tenant_id = "d434dd90-76ec-11e2-bcfd-0800200c9a66"
test_profile_id = "dd7b9741-76ec-11e2-bcfd-0800200c9a66"
test_profile_type = "network"
self._create_test_binding_if_not_there(test_tenant_id,
test_profile_id,
test_profile_type)
n1kv_db_v2.delete_profile_binding(test_tenant_id, test_profile_id)
q = (self.session.query(n1kv_models_v2.ProfileBinding).filter_by(
profile_type=test_profile_type,
tenant_id=test_tenant_id,
profile_id=test_profile_id))
self.assertFalse(q.count())
def test_default_tenant_replace(self):
ctx = context.get_admin_context()
ctx.tenant_id = "d434dd90-76ec-11e2-bcfd-0800200c9a66"
test_profile_id = "AAAAAAAA-76ec-11e2-bcfd-0800200c9a66"
test_profile_type = "policy"
n1kv_db_v2.create_profile_binding(cisco_constants.TENANT_ID_NOT_SET,
test_profile_id,
test_profile_type)
network_profile = {"network_profile": TEST_NETWORK_PROFILE}
test_network_profile = self.create_network_profile(ctx,
network_profile)
binding = n1kv_db_v2.get_profile_binding(ctx.tenant_id,
test_profile_id)
self.assertIsNone(n1kv_db_v2.get_profile_binding(
cisco_constants.TENANT_ID_NOT_SET,
test_profile_id))
self.assertNotEqual(binding.tenant_id,
cisco_constants.TENANT_ID_NOT_SET)
n1kv_db_v2.delete_network_profile(self.session,
test_network_profile['id'])
View Git Blame Copy Permalink