¡@

Home 

OpenStack Study: test_ovs_lib.py

OpenStack Index

**** CubicPower OpenStack Study ****

# Copyright 2012, VMware, 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.

try:

from collections import OrderedDict

except ImportError:

from ordereddict import OrderedDict

import mock

from oslo.config import cfg

import testtools

from neutron.agent.linux import ovs_lib

from neutron.agent.linux import utils

from neutron.common import exceptions

from neutron.openstack.common import jsonutils

from neutron.openstack.common import uuidutils

from neutron.plugins.openvswitch.common import constants

from neutron.tests import base

from neutron.tests import tools

OVS_LINUX_KERN_VERS_WITHOUT_VXLAN = "3.12.0"

**** CubicPower OpenStack Study ****

class TestBaseOVS(base.BaseTestCase):

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(TestBaseOVS, self).setUp()

        self.root_helper = 'sudo'

        self.ovs = ovs_lib.BaseOVS(self.root_helper)

        self.br_name = 'bridge1'

**** CubicPower OpenStack Study ****

    def test_add_bridge(self):

        with mock.patch.object(self.ovs, 'run_vsctl') as mock_vsctl:

            self.ovs.add_bridge(self.br_name)

        mock_vsctl.assert_called_with(["--", "--may-exist",

                                       "add-br", self.br_name])

**** CubicPower OpenStack Study ****

    def test_delete_bridge(self):

        with mock.patch.object(self.ovs, 'run_vsctl') as mock_vsctl:

            self.ovs.delete_bridge(self.br_name)

        mock_vsctl.assert_called_with(["--", "--if-exists", "del-br",

                                       self.br_name])

**** CubicPower OpenStack Study ****

    def test_bridge_exists_returns_true(self):

        with mock.patch.object(self.ovs, 'run_vsctl') as mock_vsctl:

            self.assertTrue(self.ovs.bridge_exists(self.br_name))

        mock_vsctl.assert_called_with(['br-exists', self.br_name],

                                      check_error=True)

**** CubicPower OpenStack Study ****

    def test_bridge_exists_returns_false_for_exit_code_2(self):

        with mock.patch.object(self.ovs, 'run_vsctl',

                               side_effect=RuntimeError('Exit code: 2\n')):

            self.assertFalse(self.ovs.bridge_exists('bridge1'))

**** CubicPower OpenStack Study ****

    def test_bridge_exists_raises_unknown_exception(self):

        with mock.patch.object(self.ovs, 'run_vsctl',

                               side_effect=RuntimeError()):

            with testtools.ExpectedException(RuntimeError):

                self.ovs.bridge_exists('bridge1')

**** CubicPower OpenStack Study ****

    def test_get_bridge_name_for_port_name_returns_bridge_for_valid_port(self):

        port_name = 'bar'

        with mock.patch.object(self.ovs, 'run_vsctl',

                               return_value=self.br_name) as mock_vsctl:

            bridge = self.ovs.get_bridge_name_for_port_name(port_name)

        self.assertEqual(bridge, self.br_name)

        mock_vsctl.assert_called_with(['port-to-br', port_name],

                                      check_error=True)

**** CubicPower OpenStack Study ****

    def test_get_bridge_name_for_port_name_returns_none_for_exit_code_1(self):

        with mock.patch.object(self.ovs, 'run_vsctl',

                               side_effect=RuntimeError('Exit code: 1\n')):

            self.assertFalse(self.ovs.get_bridge_name_for_port_name('bridge1'))

**** CubicPower OpenStack Study ****

    def test_get_bridge_name_for_port_name_raises_unknown_exception(self):

        with mock.patch.object(self.ovs, 'run_vsctl',

                               side_effect=RuntimeError()):

            with testtools.ExpectedException(RuntimeError):

                self.ovs.get_bridge_name_for_port_name('bridge1')

**** CubicPower OpenStack Study ****

    def _test_port_exists(self, br_name, result):

        with mock.patch.object(self.ovs,

                               'get_bridge_name_for_port_name',

                               return_value=br_name):

            self.assertEqual(self.ovs.port_exists('bar'), result)

**** CubicPower OpenStack Study ****

    def test_port_exists_returns_true_for_bridge_name(self):

        self._test_port_exists(self.br_name, True)

**** CubicPower OpenStack Study ****

    def test_port_exists_returns_false_for_none(self):

        self._test_port_exists(None, False)

**** CubicPower OpenStack Study ****

class OVS_Lib_Test(base.BaseTestCase):

"""A test suite to exercise the OVS libraries shared by Neutron agents.

Note: these tests do not actually execute ovs-* utilities, and thus

can run on any system. That does, however, limit their scope.

"""

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(OVS_Lib_Test, self).setUp()

        self.BR_NAME = "br-int"

        self.TO = "--timeout=10"

        self.root_helper = 'sudo'

        self.br = ovs_lib.OVSBridge(self.BR_NAME, self.root_helper)

        self.execute = mock.patch.object(

            utils, "execute", spec=utils.execute).start()

**** CubicPower OpenStack Study ****

    def test_vifport(self):

        """Create and stringify vif port, confirm no exceptions."""

        pname = "vif1.0"

        ofport = 5

        vif_id = uuidutils.generate_uuid()

        mac = "ca:fe:de:ad:be:ef"

        # test __init__

        port = ovs_lib.VifPort(pname, ofport, vif_id, mac, self.br)

        self.assertEqual(port.port_name, pname)

        self.assertEqual(port.ofport, ofport)

        self.assertEqual(port.vif_id, vif_id)

        self.assertEqual(port.vif_mac, mac)

        self.assertEqual(port.switch.br_name, self.BR_NAME)

        # test __str__

        str(port)

**** CubicPower OpenStack Study ****

    def test_set_controller(self):

        controller_names = ['tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555']

        self.br.set_controller(controller_names)

        self.execute.assert_called_once_with(

            ['ovs-vsctl', self.TO, '--', 'set-controller', self.BR_NAME,

             'tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555'],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_del_controller(self):

        self.br.del_controller()

        self.execute.assert_called_once_with(

            ['ovs-vsctl', self.TO, '--', 'del-controller', self.BR_NAME],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_get_controller(self):

        self.execute.return_value = 'tcp:127.0.0.1:6633\ntcp:172.17.16.10:5555'

        names = self.br.get_controller()

        self.assertEqual(names,

                         ['tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555'])

        self.execute.assert_called_once_with(

            ['ovs-vsctl', self.TO, '--', 'get-controller', self.BR_NAME],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_set_protocols(self):

        protocols = 'OpenFlow13'

        self.br.set_protocols(protocols)

        self.execute.assert_called_once_with(

            ['ovs-vsctl', self.TO, '--', 'set', 'bridge', self.BR_NAME,

             "protocols=%s" % protocols],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_create(self):

        self.br.add_bridge(self.BR_NAME)

        self.br.create()

**** CubicPower OpenStack Study ****

    def test_destroy(self):

        self.br.delete_bridge(self.BR_NAME)

        self.br.destroy()

**** CubicPower OpenStack Study ****

    def test_reset_bridge(self):

        self.br.destroy()

        self.br.create()

        self.br.reset_bridge()

**** CubicPower OpenStack Study ****

    def _build_timeout_opt(self, exp_timeout):

        return "--timeout=%d" % exp_timeout if exp_timeout else self.TO

**** CubicPower OpenStack Study ****

    def _test_delete_port(self, exp_timeout=None):

        exp_timeout_str = self._build_timeout_opt(exp_timeout)

        pname = "tap5"

        self.br.delete_port(pname)

        self.execute.assert_called_once_with(

            ["ovs-vsctl", exp_timeout_str, "--", "--if-exists",

             "del-port", self.BR_NAME, pname],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_delete_port(self):

        self._test_delete_port()

**** CubicPower OpenStack Study ****

    def test_call_command_non_default_timeput(self):

        # This test is only for verifying a non-default timeout

        # is correctly applied. Does not need to be repeated for

        # every ovs_lib method

        new_timeout = 5

        self.br.vsctl_timeout = new_timeout

        self._test_delete_port(new_timeout)

**** CubicPower OpenStack Study ****

    def test_add_flow(self):

        ofport = "99"

        vid = 4000

        lsw_id = 18

        cidr = '192.168.1.0/24'

        flow_dict_1 = OrderedDict([('priority', 2),

                                   ('dl_src', 'ca:fe:de:ad:be:ef'),

                                   ('actions', 'strip_vlan,output:0')])

        flow_dict_2 = OrderedDict([('priority', 1),

                                   ('actions', 'normal')])

        flow_dict_3 = OrderedDict([('priority', 2),

                                   ('actions', 'drop')])

        flow_dict_4 = OrderedDict([('priority', 2),

                                   ('in_port', ofport),

                                   ('actions', 'drop')])

        flow_dict_5 = OrderedDict([

            ('priority', 4),

            ('in_port', ofport),

            ('dl_vlan', vid),

            ('actions', "strip_vlan,set_tunnel:%s,normal" % (lsw_id))])

        flow_dict_6 = OrderedDict([

            ('priority', 3),

            ('tun_id', lsw_id),

            ('actions', "mod_vlan_vid:%s,output:%s" % (vid, ofport))])

        flow_dict_7 = OrderedDict([

            ('priority', 4),

            ('nw_src', cidr),

            ('proto', 'arp'),

            ('actions', 'drop')])

        self.br.add_flow(**flow_dict_1)

        self.br.add_flow(**flow_dict_2)

        self.br.add_flow(**flow_dict_3)

        self.br.add_flow(**flow_dict_4)

        self.br.add_flow(**flow_dict_5)

        self.br.add_flow(**flow_dict_6)

        self.br.add_flow(**flow_dict_7)

        expected_calls = [

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=2,dl_src=ca:fe:de:ad:be:ef"

                       ",actions=strip_vlan,output:0"],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=1,actions=normal"],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=2,actions=drop"],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=2,in_port=%s,actions=drop" % ofport],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=4,dl_vlan=%s,in_port=%s,"

                       "actions=strip_vlan,set_tunnel:%s,normal"

                       % (vid, ofport, lsw_id)],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=3,tun_id=%s,actions="

                       "mod_vlan_vid:%s,output:%s"

                       % (lsw_id, vid, ofport)],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "add-flow", self.BR_NAME,

                       "hard_timeout=0,idle_timeout=0,"

                       "priority=4,nw_src=%s,arp,actions=drop" % cidr],

                      process_input=None, root_helper=self.root_helper),

        ]

        self.execute.assert_has_calls(expected_calls)

**** CubicPower OpenStack Study ****

    def test_add_flow_timeout_set(self):

        flow_dict = OrderedDict([('priority', 1),

                                 ('hard_timeout', 1000),

                                 ('idle_timeout', 2000),

                                 ('actions', 'normal')])

        self.br.add_flow(**flow_dict)

        self.execute.assert_called_once_with(

            ["ovs-ofctl", "add-flow", self.BR_NAME,

             "hard_timeout=1000,idle_timeout=2000,priority=1,actions=normal"],

            process_input=None,

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_add_flow_default_priority(self):

        flow_dict = OrderedDict([('actions', 'normal')])

        self.br.add_flow(**flow_dict)

        self.execute.assert_called_once_with(

            ["ovs-ofctl", "add-flow", self.BR_NAME,

             "hard_timeout=0,idle_timeout=0,priority=1,actions=normal"],

            process_input=None,

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_get_port_ofport(self):

        pname = "tap99"

        ofport = "6"

        self.execute.return_value = ofport

        self.assertEqual(self.br.get_port_ofport(pname), ofport)

        self.execute.assert_called_once_with(

            ["ovs-vsctl", self.TO, "get", "Interface", pname, "ofport"],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_get_datapath_id(self):

        datapath_id = '"0000b67f4fbcc149"'

        self.execute.return_value = datapath_id

        self.assertEqual(self.br.get_datapath_id(), datapath_id.strip('"'))

        self.execute.assert_called_once_with(

            ["ovs-vsctl", self.TO, "get",

             "Bridge", self.BR_NAME, "datapath_id"],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def test_count_flows(self):

        self.execute.return_value = 'ignore\nflow-1\n'

        # counts the number of flows as total lines of output - 2

        self.assertEqual(self.br.count_flows(), 1)

        self.execute.assert_called_once_with(

            ["ovs-ofctl", "dump-flows", self.BR_NAME],

            root_helper=self.root_helper,

            process_input=None)

**** CubicPower OpenStack Study ****

    def test_delete_flow(self):

        ofport = "5"

        lsw_id = 40

        vid = 39

        self.br.delete_flows(in_port=ofport)

        self.br.delete_flows(tun_id=lsw_id)

        self.br.delete_flows(dl_vlan=vid)

        expected_calls = [

            mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,

                       "in_port=" + ofport],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,

                       "tun_id=%s" % lsw_id],

                      process_input=None, root_helper=self.root_helper),

            mock.call(["ovs-ofctl", "del-flows", self.BR_NAME,

                       "dl_vlan=%s" % vid],

                      process_input=None, root_helper=self.root_helper),

        ]

        self.execute.assert_has_calls(expected_calls)

**** CubicPower OpenStack Study ****

    def test_delete_flow_with_priority_set(self):

        params = {'in_port': '1',

                  'priority': '1'}

        self.assertRaises(exceptions.InvalidInput,

                          self.br.delete_flows,

                          **params)

**** CubicPower OpenStack Study ****

    def test_mod_flow_with_priority_set(self):

        params = {'in_port': '1',

                  'priority': '1'}

        self.assertRaises(exceptions.InvalidInput,

                          self.br.mod_flow,

                          **params)

**** CubicPower OpenStack Study ****

    def test_mod_flow_no_actions_set(self):

        params = {'in_port': '1'}

        self.assertRaises(exceptions.InvalidInput,

                          self.br.mod_flow,

                          **params)

**** CubicPower OpenStack Study ****

    def test_defer_apply_flows(self):

        flow_expr = mock.patch.object(ovs_lib, '_build_flow_expr_str').start()

        flow_expr.side_effect = ['added_flow_1', 'added_flow_2',

                                 'deleted_flow_1']

        run_ofctl = mock.patch.object(self.br, 'run_ofctl').start()

        self.br.defer_apply_on()

        self.br.add_flow(flow='add_flow_1')

        self.br.defer_apply_on()

        self.br.add_flow(flow='add_flow_2')

        self.br.delete_flows(flow='delete_flow_1')

        self.br.defer_apply_off()

        flow_expr.assert_has_calls([

            mock.call({'flow': 'add_flow_1'}, 'add'),

            mock.call({'flow': 'add_flow_2'}, 'add'),

            mock.call({'flow': 'delete_flow_1'}, 'del')

        ])

        run_ofctl.assert_has_calls([

            mock.call('add-flows', ['-'], 'added_flow_1\nadded_flow_2\n'),

            mock.call('del-flows', ['-'], 'deleted_flow_1\n')

        ])

**** CubicPower OpenStack Study ****

    def test_add_tunnel_port(self):

        pname = "tap99"

        local_ip = "1.1.1.1"

        remote_ip = "9.9.9.9"

        ofport = "6"

        command = ["ovs-vsctl", self.TO, '--', "--may-exist", "add-port",

                   self.BR_NAME, pname]

        command.extend(["--", "set", "Interface", pname])

        command.extend(["type=gre", "options:remote_ip=" + remote_ip,

                        "options:local_ip=" + local_ip,

                        "options:in_key=flow",

                        "options:out_key=flow"])

        # Each element is a tuple of (expected mock call, return_value)

        expected_calls_and_values = [

            (mock.call(command, root_helper=self.root_helper), None),

            (mock.call(["ovs-vsctl", self.TO, "get",

                        "Interface", pname, "ofport"],

                       root_helper=self.root_helper),

             ofport),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertEqual(

            self.br.add_tunnel_port(pname, remote_ip, local_ip),

            ofport)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def test_add_patch_port(self):

        pname = "tap99"

        peer = "bar10"

        ofport = "6"

        # Each element is a tuple of (expected mock call, return_value)

        command = ["ovs-vsctl", self.TO, "add-port", self.BR_NAME, pname]

        command.extend(["--", "set", "Interface", pname])

        command.extend(["type=patch", "options:peer=" + peer])

        expected_calls_and_values = [

            (mock.call(command, root_helper=self.root_helper),

             None),

            (mock.call(["ovs-vsctl", self.TO, "get",

                        "Interface", pname, "ofport"],

                       root_helper=self.root_helper),

             ofport)

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertEqual(self.br.add_patch_port(pname, peer), ofport)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def _test_get_vif_ports(self, is_xen=False):

        pname = "tap99"

        ofport = "6"

        vif_id = uuidutils.generate_uuid()

        mac = "ca:fe:de:ad:be:ef"

        if is_xen:

            external_ids = ('{xs-vif-uuid="%s", attached-mac="%s"}'

                            % (vif_id, mac))

        else:

            external_ids = ('{iface-id="%s", attached-mac="%s"}'

                            % (vif_id, mac))

        # Each element is a tuple of (expected mock call, return_value)

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             "%s\n" % pname),

            (mock.call(["ovs-vsctl", self.TO, "get",

                        "Interface", pname, "external_ids"],

                       root_helper=self.root_helper),

             external_ids),

            (mock.call(["ovs-vsctl", self.TO, "get",

                        "Interface", pname, "ofport"],

                       root_helper=self.root_helper),

             ofport),

        ]

        if is_xen:

            expected_calls_and_values.append(

                (mock.call(["xe", "vif-param-get", "param-name=other-config",

                            "param-key=nicira-iface-id", "uuid=" + vif_id],

                           root_helper=self.root_helper),

                 vif_id)

            )

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        ports = self.br.get_vif_ports()

        self.assertEqual(1, len(ports))

        self.assertEqual(ports[0].port_name, pname)

        self.assertEqual(ports[0].ofport, ofport)

        self.assertEqual(ports[0].vif_id, vif_id)

        self.assertEqual(ports[0].vif_mac, mac)

        self.assertEqual(ports[0].switch.br_name, self.BR_NAME)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def _encode_ovs_json(self, headings, data):

        # See man ovs-vsctl(8) for the encoding details.

        r = {"data": [],

             "headings": headings}

        for row in data:

            ovs_row = []

            r["data"].append(ovs_row)

            for cell in row:

                if isinstance(cell, (str, int, list)):

                    ovs_row.append(cell)

                elif isinstance(cell, dict):

                    ovs_row.append(["map", cell.items()])

                elif isinstance(cell, set):

                    ovs_row.append(["set", cell])

                else:

                    raise TypeError('%r not int, str, list, set or dict' %

                                    type(cell))

        return jsonutils.dumps(r)

**** CubicPower OpenStack Study ****

    def _test_get_vif_port_set(self, is_xen):

        if is_xen:

            id_key = 'xs-vif-uuid'

        else:

            id_key = 'iface-id'

        headings = ['name', 'external_ids']

        data = [

            # A vif port on this bridge:

            ['tap99', {id_key: 'tap99id', 'attached-mac': 'tap99mac'}, 1],

            # A vif port on this bridge not yet configured

            ['tap98', {id_key: 'tap98id', 'attached-mac': 'tap98mac'}, []],

            # Another vif port on this bridge not yet configured

            ['tap97', {id_key: 'tap97id', 'attached-mac': 'tap97mac'},

             ['set', []]],

            # A vif port on another bridge:

            ['tap88', {id_key: 'tap88id', 'attached-mac': 'tap88id'}, 1],

            # Non-vif port on this bridge:

            ['tun22', {}, 2],

        ]

        # Each element is a tuple of (expected mock call, return_value)

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             'tap99\ntun22'),

            (mock.call(["ovs-vsctl", self.TO, "--format=json",

                        "--", "--columns=name,external_ids,ofport",

                        "list", "Interface"],

                       root_helper=self.root_helper),

             self._encode_ovs_json(headings, data)),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        if is_xen:

            get_xapi_iface_id = mock.patch.object(self.br,

                                                  'get_xapi_iface_id').start()

            get_xapi_iface_id.return_value = 'tap99id'

        port_set = self.br.get_vif_port_set()

        self.assertEqual(set(['tap99id']), port_set)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

        if is_xen:

            get_xapi_iface_id.assert_called_once_with('tap99id')

**** CubicPower OpenStack Study ****

    def test_get_vif_ports_nonxen(self):

        self._test_get_vif_ports(is_xen=False)

**** CubicPower OpenStack Study ****

    def test_get_vif_ports_xen(self):

        self._test_get_vif_ports(is_xen=True)

**** CubicPower OpenStack Study ****

    def test_get_vif_port_set_nonxen(self):

        self._test_get_vif_port_set(False)

**** CubicPower OpenStack Study ****

    def test_get_vif_port_set_xen(self):

        self._test_get_vif_port_set(True)

**** CubicPower OpenStack Study ****

    def test_get_vif_ports_list_ports_error(self):

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             RuntimeError()),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertRaises(RuntimeError, self.br.get_vif_ports)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def test_get_vif_port_set_list_ports_error(self):

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             RuntimeError()),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertRaises(RuntimeError, self.br.get_vif_port_set)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def test_get_vif_port_set_list_interface_error(self):

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             'tap99\n'),

            (mock.call(["ovs-vsctl", self.TO, "--format=json",

                        "--", "--columns=name,external_ids,ofport",

                        "list", "Interface"],

                       root_helper=self.root_helper),

             RuntimeError()),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertRaises(RuntimeError, self.br.get_vif_port_set)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def test_get_port_tag_dict(self):

        headings = ['name', 'tag']

        data = [

            ['int-br-eth2', set()],

            ['patch-tun', set()],

            ['qr-76d9e6b6-21', 1],

            ['tapce5318ff-78', 1],

            ['tape1400310-e6', 1],

        ]

        # Each element is a tuple of (expected mock call, return_value)

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             '\n'.join((iface for iface, tag in data))),

            (mock.call(["ovs-vsctl", self.TO, "--format=json",

                        "--", "--columns=name,tag",

                        "list", "Port"],

                       root_helper=self.root_helper),

             self._encode_ovs_json(headings, data)),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        port_tags = self.br.get_port_tag_dict()

        self.assertEqual(

            port_tags,

            {u'int-br-eth2': [],

             u'patch-tun': [],

             u'qr-76d9e6b6-21': 1,

             u'tapce5318ff-78': 1,

             u'tape1400310-e6': 1}

        )

**** CubicPower OpenStack Study ****

    def test_clear_db_attribute(self):

        pname = "tap77"

        self.br.clear_db_attribute("Port", pname, "tag")

        self.execute.assert_called_once_with(

            ["ovs-vsctl", self.TO, "clear", "Port", pname, "tag"],

            root_helper=self.root_helper)

**** CubicPower OpenStack Study ****

    def _test_iface_to_br(self, exp_timeout=None):

        iface = 'tap0'

        br = 'br-int'

        root_helper = 'sudo'

        self.execute.return_value = 'br-int'

        exp_timeout_str = self._build_timeout_opt(exp_timeout)

        self.assertEqual(ovs_lib.get_bridge_for_iface(root_helper, iface), br)

        self.execute.assert_called_once_with(

            ["ovs-vsctl", exp_timeout_str, "iface-to-br", iface],

            root_helper=root_helper)

**** CubicPower OpenStack Study ****

    def test_iface_to_br(self):

        self._test_iface_to_br()

**** CubicPower OpenStack Study ****

    def test_iface_to_br_non_default_timeout(self):

        new_timeout = 5

        cfg.CONF.set_override('ovs_vsctl_timeout', new_timeout)

        self._test_iface_to_br(new_timeout)

**** CubicPower OpenStack Study ****

    def test_iface_to_br_handles_ovs_vsctl_exception(self):

        iface = 'tap0'

        root_helper = 'sudo'

        self.execute.side_effect = Exception

        self.assertIsNone(ovs_lib.get_bridge_for_iface(root_helper, iface))

        self.execute.assert_called_once_with(

            ["ovs-vsctl", self.TO, "iface-to-br", iface],

            root_helper=root_helper)

**** CubicPower OpenStack Study ****

    def test_delete_all_ports(self):

        with mock.patch.object(self.br, 'get_port_name_list',

                               return_value=['port1']) as get_port:

            with mock.patch.object(self.br, 'delete_port') as delete_port:

                self.br.delete_ports(all_ports=True)

        get_port.assert_called_once_with()

        delete_port.assert_called_once_with('port1')

**** CubicPower OpenStack Study ****

    def test_delete_neutron_ports(self):

        port1 = ovs_lib.VifPort('tap1234', 1, uuidutils.generate_uuid(),

                                'ca:fe:de:ad:be:ef', 'br')

        port2 = ovs_lib.VifPort('tap5678', 2, uuidutils.generate_uuid(),

                                'ca:ee:de:ad:be:ef', 'br')

        with mock.patch.object(self.br, 'get_vif_ports',

                               return_value=[port1, port2]) as get_ports:

            with mock.patch.object(self.br, 'delete_port') as delete_port:

                self.br.delete_ports(all_ports=False)

        get_ports.assert_called_once_with()

        delete_port.assert_has_calls([

            mock.call('tap1234'),

            mock.call('tap5678')

        ])

**** CubicPower OpenStack Study ****

    def test_delete_neutron_ports_list_error(self):

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],

                       root_helper=self.root_helper),

             RuntimeError()),

        ]

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        self.assertRaises(RuntimeError, self.br.delete_ports, all_ports=False)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

**** CubicPower OpenStack Study ****

    def _test_get_bridges(self, exp_timeout=None):

        bridges = ['br-int', 'br-ex']

        root_helper = 'sudo'

        self.execute.return_value = 'br-int\nbr-ex\n'

        timeout_str = self._build_timeout_opt(exp_timeout)

        self.assertEqual(ovs_lib.get_bridges(root_helper), bridges)

        self.execute.assert_called_once_with(

            ["ovs-vsctl", timeout_str, "list-br"],

            root_helper=root_helper)

**** CubicPower OpenStack Study ****

    def test_get_bridges(self):

        self._test_get_bridges()

**** CubicPower OpenStack Study ****

    def test_get_bridges_not_default_timeout(self):

        new_timeout = 5

        cfg.CONF.set_override('ovs_vsctl_timeout', new_timeout)

        self._test_get_bridges(new_timeout)

**** CubicPower OpenStack Study ****

    def test_get_local_port_mac_succeeds(self):

        with mock.patch('neutron.agent.linux.ip_lib.IpLinkCommand',

                        return_value=mock.Mock(address='foo')):

            self.assertEqual('foo', self.br.get_local_port_mac())

**** CubicPower OpenStack Study ****

    def test_get_local_port_mac_raises_exception_for_missing_mac(self):

        with mock.patch('neutron.agent.linux.ip_lib.IpLinkCommand',

                        return_value=mock.Mock(address=None)):

            with testtools.ExpectedException(Exception):

                self.br.get_local_port_mac()

**** CubicPower OpenStack Study ****

    def _test_get_vif_port_by_id(self, iface_id, data, br_name=None):

        headings = ['external_ids', 'name', 'ofport']

        # Each element is a tuple of (expected mock call, return_value)

        expected_calls_and_values = [

            (mock.call(["ovs-vsctl", self.TO, "--format=json",

                        "--", "--columns=external_ids,name,ofport",

                        "find", "Interface",

                        'external_ids:iface-id="%s"' % iface_id],

                       root_helper=self.root_helper),

             self._encode_ovs_json(headings, data))]

        if data:

            if not br_name:

                br_name = self.BR_NAME

            expected_calls_and_values.append(

                (mock.call(["ovs-vsctl", self.TO,

                            "iface-to-br", data[0][headings.index('name')]],

                           root_helper=self.root_helper),

                 br_name))

        tools.setup_mock_calls(self.execute, expected_calls_and_values)

        vif_port = self.br.get_vif_port_by_id(iface_id)

        tools.verify_mock_calls(self.execute, expected_calls_and_values)

        return vif_port

**** CubicPower OpenStack Study ****

    def _test_get_vif_port_by_id_with_data(self, ofport=None, mac=None):

        external_ids = [["iface-id", "tap99id"],

                        ["iface-status", "active"]]

        if mac:

            external_ids.append(["attached-mac", mac])

        data = [[["map", external_ids], "tap99",

                 ofport if ofport else '["set",[]]']]

        vif_port = self._test_get_vif_port_by_id('tap99id', data)

        if not ofport or ofport == -1 or not mac:

            self.assertIsNone(vif_port)

            return

        self.assertEqual(vif_port.vif_id, 'tap99id')

        self.assertEqual(vif_port.vif_mac, 'aa:bb:cc:dd:ee:ff')

        self.assertEqual(vif_port.port_name, 'tap99')

        self.assertEqual(vif_port.ofport, ofport)

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_with_ofport(self):

        self._test_get_vif_port_by_id_with_data(

            ofport=1, mac="aa:bb:cc:dd:ee:ff")

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_without_ofport(self):

        self._test_get_vif_port_by_id_with_data(mac="aa:bb:cc:dd:ee:ff")

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_with_invalid_ofport(self):

        self._test_get_vif_port_by_id_with_data(

            ofport=-1, mac="aa:bb:cc:dd:ee:ff")

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_without_mac(self):

        self._test_get_vif_port_by_id_with_data(ofport=1)

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_with_no_data(self):

        self.assertIsNone(self._test_get_vif_port_by_id('whatever', []))

**** CubicPower OpenStack Study ****

    def test_get_vif_by_port_id_different_bridge(self):

        external_ids = [["iface-id", "tap99id"],

                        ["iface-status", "active"]]

        data = [[["map", external_ids], "tap99", 1]]

        self.assertIsNone(self._test_get_vif_port_by_id('tap99id', data,

                                                        "br-ext"))

**** CubicPower OpenStack Study ****

    def _check_ovs_vxlan_version(self, installed_usr_version,

                                 installed_klm_version,

                                 installed_kernel_version,

                                 expecting_ok):

        with mock.patch(

                'neutron.agent.linux.ovs_lib.get_installed_ovs_klm_version'

        ) as klm_cmd:

            with mock.patch(

                'neutron.agent.linux.ovs_lib.get_installed_ovs_usr_version'

            ) as usr_cmd:

                with mock.patch(

                    'neutron.agent.linux.ovs_lib.get_installed_kernel_version'

                ) as kernel_cmd:

                    try:

                        klm_cmd.return_value = installed_klm_version

                        usr_cmd.return_value = installed_usr_version

                        kernel_cmd.return_value = installed_kernel_version

                        ovs_lib.check_ovs_vxlan_version(root_helper='sudo')

                        version_ok = True

                    except SystemError:

                        version_ok = False

                self.assertEqual(version_ok, expecting_ok)

**** CubicPower OpenStack Study ****

    def test_check_minimum_version(self):

        min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION

        min_kernel_ver = constants.MINIMUM_LINUX_KERNEL_OVS_VXLAN

        self._check_ovs_vxlan_version(min_vxlan_ver, min_vxlan_ver,

                                      min_kernel_ver, expecting_ok=True)

**** CubicPower OpenStack Study ****

    def test_check_future_version(self):

        install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) + 0.01)

        min_kernel_ver = constants.MINIMUM_LINUX_KERNEL_OVS_VXLAN

        self._check_ovs_vxlan_version(install_ver, install_ver,

                                      min_kernel_ver, expecting_ok=True)

**** CubicPower OpenStack Study ****

    def test_check_fail_version(self):

        install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) - 0.01)

        min_kernel_ver = constants.MINIMUM_LINUX_KERNEL_OVS_VXLAN

        self._check_ovs_vxlan_version(install_ver, install_ver,

                                      min_kernel_ver, expecting_ok=False)

**** CubicPower OpenStack Study ****

    def test_check_fail_no_version(self):

        min_kernel_ver = constants.MINIMUM_LINUX_KERNEL_OVS_VXLAN

        self._check_ovs_vxlan_version(None, None,

                                      min_kernel_ver,

                                      expecting_ok=False)

**** CubicPower OpenStack Study ****

    def test_check_fail_klm_version(self):

        min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION

        min_kernel_ver = OVS_LINUX_KERN_VERS_WITHOUT_VXLAN

        install_ver = str(float(min_vxlan_ver) - 0.01)

        self._check_ovs_vxlan_version(min_vxlan_ver,

                                      install_ver,

                                      min_kernel_ver,

                                      expecting_ok=False)

**** CubicPower OpenStack Study ****

    def test_check_pass_kernel_version(self):

        min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION

        min_kernel_ver = constants.MINIMUM_LINUX_KERNEL_OVS_VXLAN

        self._check_ovs_vxlan_version(min_vxlan_ver, min_vxlan_ver,

                                      min_kernel_ver, expecting_ok=True)