← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-updates

« back to all changes in this revision

Viewing changes to neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2015-03-30 11:17:19 UTC
  • mfrom: (1.1.21)
  • Revision ID: package-import@ubuntu.com-20150330111719-h0gx7233p4jkkgfh
Tags: 1:2015.1~b3-0ubuntu1
https://launchpad.net/bugs/1263539
* New upstream milestone release:
  - d/control: Align version requirements with upstream.
  - d/control: Add new dependency on oslo-log.
  - d/p/*: Rebase.
  - d/control,d/neutron-plugin-hyperv*: Dropped, decomposed into
    separate project upstream.
  - d/control,d/neutron-plugin-openflow*: Dropped, decomposed into
    separate project upstream.
  - d/neutron-common.install: Add neutron-rootwrap-daemon and 
    neutron-keepalived-state-change binaries.
  - d/rules: Ignore neutron-hyperv-agent when installing; only for Windows.
  - d/neutron-plugin-cisco.install: Drop neutron-cisco-cfg-agent as
    decomposed into separate project upstream.
  - d/neutron-plugin-vmware.install: Drop neutron-check-nsx-config and
    neutron-nsx-manage as decomposed into separate project upstream.
  - d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent.
* d/pydist-overrides: Add overrides for oslo packages.
* d/control: Fixup type in package description (LP: #1263539).
* d/p/fixup-driver-test-execution.patch: Cherry pick fix from upstream VCS
  to support unit test exection in out-of-tree vendor drivers.
* d/neutron-common.postinst: Allow general access to /etc/neutron but limit
  access to root/neutron to /etc/neutron/neutron.conf to support execution
  of unit tests in decomposed vendor drivers.
* d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent
  package.

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/linuxbridge/test_lb_neutron_agent.py
16
16
import os
17
17
 
18
18
import mock
19
 
from oslo.config import cfg
 
19
from oslo_config import cfg
20
20
 
21
21
from neutron.agent.linux import ip_lib
22
22
from neutron.agent.linux import utils
47
47
        self.skipTest("udev not consistently available in Ubuntu buildds")
48
48
        super(TestLinuxBridge, self).setUp()
49
49
        interface_mappings = {'physnet1': 'eth1'}
50
 
        root_helper = cfg.CONF.AGENT.root_helper
51
50
 
52
 
        self.linux_bridge = linuxbridge_neutron_agent.LinuxBridgeManager(
53
 
            interface_mappings, root_helper)
 
51
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
 
52
                               'get_interface_by_ip', return_value=None):
 
53
            self.linux_bridge = linuxbridge_neutron_agent.LinuxBridgeManager(
 
54
                interface_mappings)
54
55
 
55
56
    def test_ensure_physical_in_bridge_invalid(self):
56
57
        result = self.linux_bridge.ensure_physical_in_bridge('network_id',
107
108
                                    'get_interface_mac')
108
109
        self.get_mac = self.get_mac_p.start()
109
110
        self.get_mac.return_value = '00:00:00:00:00:01'
110
 
        self.agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
111
 
                                                                          0,
112
 
                                                                          None)
 
111
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
 
112
                               'get_interface_by_ip', return_value=None):
 
113
            self.agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC(
 
114
                {}, 0)
113
115
 
114
116
    def test_treat_devices_removed_with_existed_device(self):
115
 
        agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
116
 
                                                                     0,
117
 
                                                                     None)
 
117
        agent = self.agent
118
118
        devices = [DEVICE_1]
119
119
        with contextlib.nested(
120
120
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
131
131
                self.assertTrue(fn_rdf.called)
132
132
 
133
133
    def test_treat_devices_removed_with_not_existed_device(self):
134
 
        agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
135
 
                                                                     0,
136
 
                                                                     None)
 
134
        agent = self.agent
137
135
        devices = [DEVICE_1]
138
136
        with contextlib.nested(
139
137
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
150
148
                self.assertTrue(fn_rdf.called)
151
149
 
152
150
    def test_treat_devices_removed_failed(self):
153
 
        agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC({},
154
 
                                                                     0,
155
 
                                                                     None)
 
151
        agent = self.agent
156
152
        devices = [DEVICE_1]
157
153
        with contextlib.nested(
158
154
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
349
345
        self.skipTest("udev not consistently available in Ubuntu buildds")
350
346
        super(TestLinuxBridgeManager, self).setUp()
351
347
        self.interface_mappings = {'physnet1': 'eth1'}
352
 
        self.root_helper = cfg.CONF.AGENT.root_helper
353
348
 
354
 
        self.lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
355
 
            self.interface_mappings, self.root_helper)
 
349
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
 
350
                               'get_interface_by_ip', return_value=None):
 
351
            self.lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
 
352
                self.interface_mappings)
356
353
 
357
354
    def test_interface_exists_on_bridge(self):
358
355
        with mock.patch.object(os, 'listdir') as listdir_fn:
450
447
            exists_fn.return_value = True
451
448
            self.assertTrue(self.lbm.is_device_on_bridge("tap1"))
452
449
            exists_fn.assert_called_with(
453
 
                "/sys/devices/virtual/net/tap1/brport"
 
450
                "/sys/class/net/tap1/brport"
454
451
            )
455
452
 
456
453
    def test_get_interface_details(self):
625
622
            self.lbm.ensure_bridge("br0", "eth0")
626
623
            expected = [
627
624
                mock.call(['brctl', 'delif', 'br1', 'eth0'],
628
 
                          root_helper=self.root_helper),
 
625
                          run_as_root=True),
629
626
                mock.call(['brctl', 'addif', 'br0', 'eth0'],
630
 
                          root_helper=self.root_helper),
 
627
                          run_as_root=True),
631
628
            ]
632
629
            exec_fn.assert_has_calls(expected)
633
630
 
764
761
 
765
762
    def test_delete_vxlan_bridge_no_int_mappings(self):
766
763
        interface_mappings = {}
767
 
        lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
768
 
            interface_mappings, self.root_helper)
 
764
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
 
765
                               'get_interface_by_ip', return_value=None):
 
766
            lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
 
767
                interface_mappings)
769
768
 
770
769
        with contextlib.nested(
771
770
            mock.patch.object(ip_lib, "device_exists"),
928
927
            def __init__(self):
929
928
                self.agent_id = 1
930
929
                self.br_mgr = (linuxbridge_neutron_agent.
931
 
                               LinuxBridgeManager({'physnet1': 'eth1'},
932
 
                                                  cfg.CONF.AGENT.root_helper))
 
930
                               LinuxBridgeManager({'physnet1': 'eth1'}))
933
931
 
934
932
                self.br_mgr.vxlan_mode = lconst.VXLAN_UCAST
935
933
                segment = mock.Mock()
943
941
            object()
944
942
        )
945
943
 
946
 
        self.root_helper = cfg.CONF.AGENT.root_helper
947
 
 
948
944
    def test_network_delete(self):
949
945
        with contextlib.nested(
950
946
            mock.patch.object(self.lb_rpc.agent.br_mgr, "get_bridge_name"),
969
965
 
970
966
            expected = [
971
967
                mock.call(['bridge', 'fdb', 'show', 'dev', 'vxlan-1'],
972
 
                          root_helper=self.root_helper),
 
968
                          run_as_root=True),
973
969
                mock.call(['bridge', 'fdb', 'add',
974
970
                           constants.FLOODING_ENTRY[0],
975
971
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
976
 
                          root_helper=self.root_helper,
 
972
                          run_as_root=True,
977
973
                          check_exit_code=False),
978
974
                mock.call(['ip', 'neigh', 'replace', 'port_ip', 'lladdr',
979
975
                           'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
980
 
                          root_helper=self.root_helper,
 
976
                          run_as_root=True,
981
977
                          check_exit_code=False),
982
978
                mock.call(['bridge', 'fdb', 'add', 'port_mac', 'dev',
983
979
                           'vxlan-1', 'dst', 'agent_ip'],
984
 
                          root_helper=self.root_helper,
 
980
                          run_as_root=True,
985
981
                          check_exit_code=False),
986
982
            ]
987
983
            execute_fn.assert_has_calls(expected)
1031
1027
                mock.call(['bridge', 'fdb', 'del',
1032
1028
                           constants.FLOODING_ENTRY[0],
1033
1029
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
1034
 
                          root_helper=self.root_helper,
 
1030
                          run_as_root=True,
1035
1031
                          check_exit_code=False),
1036
1032
                mock.call(['ip', 'neigh', 'del', 'port_ip', 'lladdr',
1037
1033
                           'port_mac', 'dev', 'vxlan-1'],
1038
 
                          root_helper=self.root_helper,
 
1034
                          run_as_root=True,
1039
1035
                          check_exit_code=False),
1040
1036
                mock.call(['bridge', 'fdb', 'del', 'port_mac',
1041
1037
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
1042
 
                          root_helper=self.root_helper,
 
1038
                          run_as_root=True,
1043
1039
                          check_exit_code=False),
1044
1040
            ]
1045
1041
            execute_fn.assert_has_calls(expected)
1058
1054
            expected = [
1059
1055
                mock.call(['ip', 'neigh', 'replace', 'port_ip_2', 'lladdr',
1060
1056
                           'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
1061
 
                          root_helper=self.root_helper,
 
1057
                          run_as_root=True,
1062
1058
                          check_exit_code=False),
1063
1059
                mock.call(['ip', 'neigh', 'del', 'port_ip_1', 'lladdr',
1064
1060
                           'port_mac', 'dev', 'vxlan-1'],
1065
 
                          root_helper=self.root_helper,
 
1061
                          run_as_root=True,
1066
1062
                          check_exit_code=False)
1067
1063
            ]
1068
1064
            execute_fn.assert_has_calls(expected)