openstack
/
neutron
Code Issues Proposed changes

Merge "Refactor type_gre.vxlan tests to reduce duplicate code"

This commit is contained in:
Jenkins 2015-06-02 00:41:12 +00:00 committed by Gerrit Code Review
parent 25bcbfdbd4 b7ffbf96ff
commit 88068ee94d
3 changed files with 77 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py
View File

@ -33,6 +33,7 @@ UPDATED_TUNNEL_RANGES = [(TUN_MIN + 5, TUN_MAX + 5)]
class TunnelTypeTestMixin(object):
DRIVER_MODULE = None
DRIVER_CLASS = None
TYPE = None
@ -195,6 +196,53 @@ class TunnelTypeTestMixin(object):
segment[api.SEGMENTATION_ID] = tunnel_id
self.driver.release_segment(self.session, segment)
def add_endpoint(self, ip=TUNNEL_IP_ONE, host=HOST_ONE):
return self.driver.add_endpoint(ip, host)
def test_add_endpoint(self):
endpoint = self.add_endpoint()
self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
self.assertEqual(HOST_ONE, endpoint.host)
return endpoint
def test_add_endpoint_for_existing_tunnel_ip(self):
self.add_endpoint()
log = getattr(self.DRIVER_MODULE, 'LOG')
with mock.patch.object(log, 'warning') as log_warn:
self.add_endpoint()
log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
def test_get_endpoint_by_host(self):
self.add_endpoint()
host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
return host_endpoint
def test_get_endpoint_by_host_for_not_existing_host(self):
ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
self.assertIsNone(ip_endpoint)
def test_get_endpoint_by_ip(self):
self.add_endpoint()
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
self.assertEqual(HOST_ONE, ip_endpoint.host)
return ip_endpoint
def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
self.assertIsNone(ip_endpoint)
def test_delete_endpoint(self):
self.add_endpoint()
self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
# Get all the endpoints and verify its empty
endpoints = self.driver.get_endpoints()
self.assertNotIn(TUNNEL_IP_ONE, endpoints)
class TunnelTypeMultiRangeTestMixin(object):
DRIVER_CLASS = None

54
neutron/tests/unit/plugins/ml2/drivers/test_type_gre.py
View File

@ -46,59 +46,21 @@ def _get_allocation(session, gre_id):
class GreTypeTest(base_type_tunnel.TunnelTypeTestMixin,
testlib_api.SqlTestCase):
DRIVER_MODULE = type_gre
DRIVER_CLASS = type_gre.GreTypeDriver
TYPE = p_const.TYPE_GRE
def test_add_endpoint(self):
endpoint = self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
self.assertEqual(HOST_ONE, endpoint.host)
def test_add_endpoint_for_existing_tunnel_ip(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
with mock.patch.object(type_gre.LOG, 'warning') as log_warn:
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
def test_get_endpoint_by_host(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
def test_get_endpoint_by_host_for_not_existing_host(self):
ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
self.assertIsNone(ip_endpoint)
def test_get_endpoint_by_ip(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
self.assertEqual(HOST_ONE, ip_endpoint.host)
def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
self.assertIsNone(ip_endpoint)
def test_get_endpoints(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
self.driver.add_endpoint(TUNNEL_IP_TWO, HOST_TWO)
self.add_endpoint()
self.add_endpoint(
base_type_tunnel.TUNNEL_IP_TWO, base_type_tunnel.HOST_TWO)
endpoints = self.driver.get_endpoints()
for endpoint in endpoints:
if endpoint['ip_address'] == TUNNEL_IP_ONE:
self.assertEqual(HOST_ONE, endpoint['host'])
elif endpoint['ip_address'] == TUNNEL_IP_TWO:
self.assertEqual(HOST_TWO, endpoint['host'])
def test_delete_endpoint(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
# Get all the endpoints and verify its empty
endpoints = self.driver.get_endpoints()
self.assertNotIn(TUNNEL_IP_ONE, endpoints)
if endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_ONE:
self.assertEqual(base_type_tunnel.HOST_ONE, endpoint['host'])
elif endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_TWO:
self.assertEqual(base_type_tunnel.HOST_TWO, endpoint['host'])
def test_sync_allocations_entry_added_during_session(self):
with mock.patch.object(self.driver, '_add_allocation',

71
neutron/tests/unit/plugins/ml2/drivers/test_type_vxlan.py
View File

@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import config
from neutron.plugins.ml2.drivers import type_vxlan
@ -23,76 +21,49 @@ from neutron.tests.unit.plugins.ml2 import test_rpc
from neutron.tests.unit import testlib_api
TUNNEL_IP_ONE = "10.10.10.10"
TUNNEL_IP_TWO = "10.10.10.20"
HOST_ONE = 'fake_host_one'
HOST_TWO = 'fake_host_two'
VXLAN_UDP_PORT_ONE = 9999
VXLAN_UDP_PORT_TWO = 8888
class VxlanTypeTest(base_type_tunnel.TunnelTypeTestMixin,
testlib_api.SqlTestCase):
DRIVER_MODULE = type_vxlan
DRIVER_CLASS = type_vxlan.VxlanTypeDriver
TYPE = p_const.TYPE_VXLAN
def add_endpoint(self, ip=base_type_tunnel.TUNNEL_IP_ONE,
host=base_type_tunnel.HOST_ONE):
if ip == base_type_tunnel.TUNNEL_IP_ONE:
port = VXLAN_UDP_PORT_ONE
else:
port = VXLAN_UDP_PORT_TWO
return self.driver.add_endpoint(ip, host, port)
def test_add_endpoint(self):
endpoint = self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE,
VXLAN_UDP_PORT_ONE)
self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
endpoint = super(VxlanTypeTest, self).test_add_endpoint()
self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
self.assertEqual(HOST_ONE, endpoint.host)
def test_add_endpoint_for_existing_tunnel_ip(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
with mock.patch.object(type_vxlan.LOG, 'warning') as log_warn:
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE,
VXLAN_UDP_PORT_ONE)
log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
def test_get_endpoint_by_host(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
self.assertEqual(VXLAN_UDP_PORT_ONE, host_endpoint.udp_port)
def test_get_endpoint_by_host_for_not_existing_host(self):
ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
self.assertIsNone(ip_endpoint)
endpoint = super(VxlanTypeTest, self).test_get_endpoint_by_host()
self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
def test_get_endpoint_by_ip(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
self.assertEqual(HOST_ONE, ip_endpoint.host)
self.assertEqual(VXLAN_UDP_PORT_ONE, ip_endpoint.udp_port)
def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
self.assertIsNone(ip_endpoint)
endpoint = super(VxlanTypeTest, self).test_get_endpoint_by_ip()
self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
def test_get_endpoints(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
self.driver.add_endpoint(TUNNEL_IP_TWO, HOST_TWO, VXLAN_UDP_PORT_TWO)
self.add_endpoint()
self.add_endpoint(base_type_tunnel.TUNNEL_IP_TWO,
base_type_tunnel.HOST_TWO)
endpoints = self.driver.get_endpoints()
for endpoint in endpoints:
if endpoint['ip_address'] == TUNNEL_IP_ONE:
if endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_ONE:
self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint['udp_port'])
self.assertEqual(HOST_ONE, endpoint['host'])
elif endpoint['ip_address'] == TUNNEL_IP_TWO:
self.assertEqual(base_type_tunnel.HOST_ONE, endpoint['host'])
elif endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_TWO:
self.assertEqual(VXLAN_UDP_PORT_TWO, endpoint['udp_port'])
self.assertEqual(HOST_TWO, endpoint['host'])
def test_delete_endpoint(self):
self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
# Get all the endpoints and verify its empty
endpoints = self.driver.get_endpoints()
self.assertNotIn(TUNNEL_IP_ONE, endpoints)
self.assertEqual(base_type_tunnel.HOST_TWO, endpoint['host'])
def test_get_mtu(self):
config.cfg.CONF.set_override('segment_mtu', 1500, group='ml2')