¡@
OpenStack Study: test_db.py**** CubicPower OpenStack Study ****
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright 2012 NEC Corporation. All rights reserved. # # 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: Ryota MIBU import contextlib import random from neutron.common import constants as q_const from neutron.openstack.common import uuidutils from neutron.plugins.nec.common import exceptions as nexc from neutron.plugins.nec.db import api as ndb from neutron.plugins.nec.db import models as nmodels # noqa from neutron.tests.unit.nec import test_nec_plugin
**** CubicPower OpenStack Study **** class NECPluginV2DBTestBase(test_nec_plugin.NecPluginV2TestCase): """Class conisting of NECPluginV2 DB unit tests."""
**** CubicPower OpenStack Study **** def setUp(self): """Setup for tests.""" super(NECPluginV2DBTestBase, self).setUp() self.session = self.context.session
**** CubicPower OpenStack Study **** def get_ofc_item_random_params(self): """create random parameters for ofc_item test.""" ofc_id = uuidutils.generate_uuid() neutron_id = uuidutils.generate_uuid() none = uuidutils.generate_uuid() return ofc_id, neutron_id, none @contextlib.contextmanager
**** CubicPower OpenStack Study **** def portinfo_random_params(self): with self.port() as port: params = {'port_id': port['port']['id'], 'datapath_id': hex(random.randint(0, 0xffffffff)), 'port_no': random.randint(1, 100), 'vlan_id': random.randint(q_const.MIN_VLAN_TAG, q_const.MAX_VLAN_TAG), 'mac': ':'.join(["%02x" % random.randint(0, 0xff) for x in range(6)]) } yield params
**** CubicPower OpenStack Study **** class NECPluginV2DBOfcMappingTest(NECPluginV2DBTestBase):
**** CubicPower OpenStack Study **** def test_add_ofc_item(self): """test add OFC item.""" o, q, n = self.get_ofc_item_random_params() tenant = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) self.assertEqual(tenant.ofc_id, o) self.assertEqual(tenant.neutron_id, q)
**** CubicPower OpenStack Study **** def test_add_ofc_item_duplicate_entry(self): o, q, n = self.get_ofc_item_random_params() ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) self.assertRaises(nexc.NECDBException, ndb.add_ofc_item, self.session, 'ofc_tenant', q, o)
**** CubicPower OpenStack Study **** def test_get_ofc_item(self): o, q, n = self.get_ofc_item_random_params() ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) tenant = ndb.get_ofc_item(self.session, 'ofc_tenant', q) self.assertEqual(tenant.ofc_id, o) self.assertEqual(tenant.neutron_id, q)
**** CubicPower OpenStack Study **** def test_get_ofc_item_for_nonexisting_entry(self): self.assertIsNone( ndb.get_ofc_item(self.session, 'ofc_tenant', 'non-exist-id'))
**** CubicPower OpenStack Study **** def test_get_ofc_id(self): o, q, n = self.get_ofc_item_random_params() ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) tenant_id = ndb.get_ofc_id(self.session, 'ofc_tenant', q) self.assertEqual(tenant_id, o)
**** CubicPower OpenStack Study **** def test_get_ofc_id_for_nonexisting_entry(self): self.assertRaises(nexc.OFCMappingNotFound, ndb.get_ofc_id, self.session, 'ofc_tenant', 'non-exist-id')
**** CubicPower OpenStack Study **** def test_exists_ofc_item(self): o, q, n = self.get_ofc_item_random_params() self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q)) ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) self.assertTrue(ndb.exists_ofc_item(self.session, 'ofc_tenant', q)) ndb.del_ofc_item(self.session, 'ofc_tenant', q) self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
**** CubicPower OpenStack Study **** def test_find_ofc_item(self): o, q, n = self.get_ofc_item_random_params() ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) tenant = ndb.find_ofc_item(self.session, 'ofc_tenant', o) self.assertEqual(tenant.ofc_id, o) self.assertEqual(tenant.neutron_id, q)
**** CubicPower OpenStack Study **** def test_find_ofc_item_for_nonexisting_entry(self): self.assertIsNone( ndb.find_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
**** CubicPower OpenStack Study **** def test_del_ofc_item(self): o, q, n = self.get_ofc_item_random_params() ndb.add_ofc_item(self.session, 'ofc_tenant', q, o) self.assertTrue(ndb.del_ofc_item(self.session, 'ofc_tenant', q)) self.assertIsNone(ndb.get_ofc_item(self.session, 'ofc_tenant', q)) self.assertIsNone(ndb.find_ofc_item(self.session, 'ofc_tenant', o))
**** CubicPower OpenStack Study **** def test_del_ofc_item_for_nonexisting_entry(self): self.assertFalse( ndb.del_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
**** CubicPower OpenStack Study **** class NECPluginV2DBPortInfoTest(NECPluginV2DBTestBase):
**** CubicPower OpenStack Study **** def _compare_portinfo(self, portinfo, expected): self.assertEqual(portinfo.id, expected['port_id']) self.assertEqual(portinfo.datapath_id, expected['datapath_id']) self.assertEqual(portinfo.port_no, expected['port_no']) self.assertEqual(portinfo.vlan_id, expected['vlan_id']) self.assertEqual(portinfo.mac, expected['mac'])
**** CubicPower OpenStack Study **** def _add_portinfo(self, session, params): return ndb.add_portinfo(session, params['port_id'], params['datapath_id'], params['port_no'], params['vlan_id'], params['mac'])
**** CubicPower OpenStack Study **** def testd_add_portinfo(self): """test add portinfo.""" with self.portinfo_random_params() as params: portinfo = self._add_portinfo(self.session, params) self._compare_portinfo(portinfo, params) exception_raised = False try: self._add_portinfo(self.session, params) except nexc.NECDBException: exception_raised = True self.assertTrue(exception_raised)
**** CubicPower OpenStack Study **** def teste_get_portinfo(self): """test get portinfo.""" with self.portinfo_random_params() as params: self._add_portinfo(self.session, params) portinfo = ndb.get_portinfo(self.session, params['port_id']) self._compare_portinfo(portinfo, params) nonexist_id = uuidutils.generate_uuid() portinfo_none = ndb.get_portinfo(self.session, nonexist_id) self.assertIsNone(portinfo_none)
**** CubicPower OpenStack Study **** def testf_del_portinfo(self): """test delete portinfo.""" with self.portinfo_random_params() as params: self._add_portinfo(self.session, params) portinfo = ndb.get_portinfo(self.session, params['port_id']) self.assertEqual(portinfo.id, params['port_id']) ndb.del_portinfo(self.session, params['port_id']) portinfo_none = ndb.get_portinfo(self.session, params['port_id']) self.assertIsNone(portinfo_none)
|