openstack/neutron
Code Issues Proposed changes

Fix update_device_up method of linuxbridge plugin

Browse Source 
Also add unit tests covering update_device_up and update_device_down
methods

Change-Id: I97f2f9249b684aa5350b3f0621754543e80bec70
Closes-Bug: #1241602
This commit is contained in:
Eugene Nikanorov 2013-10-19 16:06:03 +04:00
parent 84a847429a
commit 01e45596fa
2 changed files with 51 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
neutron/plugins/linuxbridge/lb_neutron_plugin.py
View File

@ -144,7 +144,7 @@ class LinuxBridgeRpcCallbacks(dhcp_rpc_base.DhcpRpcCallbackMixin,
agent_id = kwargs.get('agent_id')
device = kwargs.get('device')
host = kwargs.get('host')
port = self.get_port_from_device.get_port(device)
port = self.get_port_from_device(device)
LOG.debug(_("Device %(device)s up on %(agent_id)s"),
{'device': device, 'agent_id': agent_id})
plugin = manager.NeutronManager.get_plugin()

51
neutron/tests/unit/linuxbridge/test_linuxbridge_plugin.py
View File

@ -13,12 +13,18 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import contextlib
import mock
from neutron.common import constants as q_const
from neutron.extensions import portbindings
from neutron import manager
from neutron.plugins.linuxbridge import lb_neutron_plugin
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import test_security_groups_rpc as test_sg_rpc
PLUGIN_NAME = ('neutron.plugins.linuxbridge.'
'lb_neutron_plugin.LinuxBridgePluginV2')
@ -75,3 +81,46 @@ class TestLinuxBridgePortBindingHost(
LinuxBridgePluginV2TestCase,
test_bindings.PortBindingsHostTestCaseMixin):
pass
class TestLinuxBridgePluginRpcCallbacks(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
super(TestLinuxBridgePluginRpcCallbacks, self).setUp(PLUGIN_NAME)
self.callbacks = lb_neutron_plugin.LinuxBridgeRpcCallbacks()
def test_update_device_down(self):
with contextlib.nested(
mock.patch.object(self.callbacks, "get_port_from_device",
return_value=None),
mock.patch.object(manager.NeutronManager, "get_plugin")
) as (gpfd, gp):
self.assertEqual(
self.callbacks.update_device_down("fake_context",
agent_id="123",
device="device",
host="host"),
{'device': 'device', 'exists': False}
)
gpfd.return_value = {'id': 'fakeid',
'status': q_const.PORT_STATUS_ACTIVE}
self.assertEqual(
self.callbacks.update_device_down("fake_context",
agent_id="123",
device="device",
host="host"),
{'device': 'device', 'exists': True}
)
def test_update_device_up(self):
with contextlib.nested(
mock.patch.object(self.callbacks, "get_port_from_device",
return_value=None),
mock.patch.object(manager.NeutronManager, "get_plugin")
) as (gpfd, gp):
gpfd.return_value = {'id': 'fakeid',
'status': q_const.PORT_STATUS_ACTIVE}
self.callbacks.update_device_up("fake_context",
agent_id="123",
device="device",
host="host")
gpfd.assert_called_once_with('device')