openstack
/
kuryr
Code Issues Proposed changes

Merge "Add unit test for veth binding driver"

This commit is contained in:
Jenkins 2017-03-04 03:20:15 +00:00 committed by Gerrit Code Review
parent a0d2e379a8 e6ed953784
commit 895c4b4549
2 changed files with 38 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
kuryr/tests/unit/base.py
View File

@ -89,7 +89,9 @@ class TestCase(base.BaseTestCase):
neutron_subnet_v4_id=None, neutron_subnet_v4_id=None,
neutron_subnet_v6_id=None, neutron_subnet_v6_id=None,
neutron_subnet_v4_address="192.168.1.2", neutron_subnet_v4_address="192.168.1.2",
neutron_subnet_v6_address="fe80::f816:3eff:fe20:57c4"): neutron_subnet_v6_address="fe80::f816:3eff:fe20:57c4",
vif_details=None,
vif_type=None):
# The following fake response is retrieved from the Neutron doc: # The following fake response is retrieved from the Neutron doc:
# http://developer.openstack.org/api-ref-networking-v2.html#createPort # noqa # http://developer.openstack.org/api-ref-networking-v2.html#createPort # noqa
fake_port = { fake_port = {
@ -105,7 +107,9 @@ class TestCase(base.BaseTestCase):
"fixed_ips": [], "fixed_ips": [],
"id": neutron_port_id, "id": neutron_port_id,
"security_groups": [], "security_groups": [],
"device_id": "" "device_id": "",
"binding:vif_details": vif_details,
"binding:vif_type": vif_type
} }
} }

57
kuryr/tests/unit/binding/test_package.py → kuryr/tests/unit/binding/drivers/test_veth.py
View File

@ -1,31 +1,31 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import ddt
import mock import mock
from oslo_utils import uuidutils from oslo_utils import uuidutils
from kuryr.lib import binding from kuryr.lib.binding.drivers import veth
from kuryr.lib import constants from kuryr.lib import constants
from kuryr.lib import utils from kuryr.lib import utils
from kuryr.tests.unit import base from kuryr.tests.unit import base
from mock import call
mock_create = mock.MagicMock() mock_create = mock.MagicMock()
mock_interface = mock.MagicMock() mock_interface = mock.MagicMock()
@ddt.ddt class TestVethDriver(base.TestCase):
class BindingTest(base.TestCase): """Unit tests for veth driver"""
"""Unit tests for binding."""
@mock.patch('os.path.exists', return_value=True) @mock.patch('os.path.exists', return_value=True)
@mock.patch('oslo_concurrency.processutils.execute', @mock.patch('oslo_concurrency.processutils.execute',
@ -37,27 +37,31 @@ class BindingTest(base.TestCase):
def test_port_bind(self, mock_getitem, mock_getattribute, def test_port_bind(self, mock_getitem, mock_getattribute,
mock_execute, mock_path_exists): mock_execute, mock_path_exists):
fake_mtu = 1450 fake_mtu = 1450
fake_docker_network_id = utils.get_hash()
fake_docker_endpoint_id = utils.get_hash() fake_docker_endpoint_id = utils.get_hash()
fake_docker_network_id = utils.get_hash()
fake_port_id = uuidutils.generate_uuid() fake_port_id = uuidutils.generate_uuid()
fake_neutron_v4_subnet_id = uuidutils.generate_uuid() fake_neutron_v4_subnet_id = uuidutils.generate_uuid()
fake_neutron_v6_subnet_id = uuidutils.generate_uuid() fake_neutron_v6_subnet_id = uuidutils.generate_uuid()
fake_vif_details = {"port_filter": True, "ovs_hybrid_plug": False}
fake_vif_type = "ovs"
fake_port = self._get_fake_port( fake_port = self._get_fake_port(
fake_docker_endpoint_id, fake_docker_network_id, fake_docker_endpoint_id, fake_docker_network_id,
fake_port_id, constants.PORT_STATUS_ACTIVE, fake_port_id, constants.PORT_STATUS_ACTIVE,
fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id) fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id,
vif_details=fake_vif_details, vif_type=fake_vif_type)
fake_subnets = self._get_fake_subnets( fake_subnets = self._get_fake_subnets(
fake_docker_endpoint_id, fake_docker_network_id, fake_docker_endpoint_id, fake_docker_network_id,
fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id) fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id)
fake_network = self._get_fake_networks(fake_docker_network_id) fake_network = self._get_fake_networks(fake_docker_network_id)
fake_network['networks'][0]['mtu'] = fake_mtu fake_network['networks'][0]['mtu'] = fake_mtu
binding.port_bind(fake_docker_endpoint_id, fake_port['port'], veth.port_bind(fake_docker_endpoint_id,
fake_subnets['subnets'], fake_port['port'],
fake_network['networks'][0]) fake_subnets['subnets'],
fake_network['networks'][0])
expect_calls = [call.__enter__().set_mtu(fake_mtu), expect_calls = [mock.call.__enter__().set_mtu(fake_mtu),
call.__enter__().up()] mock.call.__enter__().up()]
mock_interface.assert_has_calls(expect_calls, any_order=True) mock_interface.assert_has_calls(expect_calls, any_order=True)
mock_path_exists.assert_called_once() mock_path_exists.assert_called_once()
mock_execute.assert_called_once() mock_execute.assert_called_once()
@ -66,15 +70,18 @@ class BindingTest(base.TestCase):
@mock.patch('oslo_concurrency.processutils.execute', @mock.patch('oslo_concurrency.processutils.execute',
return_value=('fake_stdout', 'fake_stderr')) return_value=('fake_stdout', 'fake_stderr'))
def test_port_unbind(self, mock_execute, mock_remove_device): def test_port_unbind(self, mock_execute, mock_remove_device):
fake_docker_network_id = utils.get_hash()
fake_docker_endpoint_id = utils.get_hash() fake_docker_endpoint_id = utils.get_hash()
fake_docker_network_id = utils.get_hash()
fake_port_id = uuidutils.generate_uuid() fake_port_id = uuidutils.generate_uuid()
fake_neutron_v4_subnet_id = uuidutils.generate_uuid() fake_neutron_v4_subnet_id = uuidutils.generate_uuid()
fake_neutron_v6_subnet_id = uuidutils.generate_uuid() fake_neutron_v6_subnet_id = uuidutils.generate_uuid()
fake_vif_details = {"port_filter": True, "ovs_hybrid_plug": False}
fake_vif_type = "ovs"
fake_port = self._get_fake_port( fake_port = self._get_fake_port(
fake_docker_endpoint_id, fake_docker_network_id, fake_docker_endpoint_id, fake_docker_network_id,
fake_port_id, constants.PORT_STATUS_ACTIVE, fake_port_id, constants.PORT_STATUS_ACTIVE,
fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id) fake_neutron_v4_subnet_id, fake_neutron_v6_subnet_id,
binding.port_unbind(fake_docker_endpoint_id, fake_port['port']) vif_details=fake_vif_details, vif_type=fake_vif_type)
veth.port_unbind(fake_docker_endpoint_id, fake_port['port'])
mock_execute.assert_called_once() mock_execute.assert_called_once()
mock_remove_device.assert_called_once() mock_remove_device.assert_called_once()