← Back to branch summary

~james-page/charms/trusty/neutron-openvswitch/lp1531102

« back to all changes in this revision

Viewing changes to unit_tests/test_neutron_ovs_context.py

Deploy from source

Show diffs side-by-side

added added

removed removed

Lines of Context:
unit_tests/test_neutron_ovs_context.py
1
1
 
2
2
from test_utils import CharmTestCase
 
3
from test_utils import patch_open
3
4
from mock import patch
4
5
import neutron_ovs_context as context
5
6
import charmhelpers
 
7
 
6
8
TO_PATCH = [
7
 
    'relation_get',
8
 
    'relation_ids',
9
 
    'related_units',
 
9
    'resolve_address',
10
10
    'config',
11
11
    'unit_get',
12
 
    'add_bridge',
13
 
    'add_bridge_port',
14
 
    'service_running',
15
 
    'service_start',
16
12
    'get_host_ip',
17
13
]
18
14
 
19
15
 
 
16
def fake_context(settings):
 
17
    def outer():
 
18
        def inner():
 
19
            return settings
 
20
        return inner
 
21
    return outer
 
22
 
 
23
 
20
24
class OVSPluginContextTest(CharmTestCase):
21
25
 
22
26
    def setUp(self):
23
27
        super(OVSPluginContextTest, self).setUp(context, TO_PATCH)
24
 
        self.relation_get.side_effect = self.test_relation.get
25
28
        self.config.side_effect = self.test_config.get
26
29
        self.test_config.set('debug', True)
27
30
        self.test_config.set('verbose', True)
30
33
    def tearDown(self):
31
34
        super(OVSPluginContextTest, self).tearDown()
32
35
 
 
36
    @patch('charmhelpers.contrib.openstack.context.config')
33
37
    @patch('charmhelpers.contrib.openstack.context.NeutronPortContext.'
34
38
           'resolve_ports')
35
 
    def test_data_port_name(self, mock_resolve_ports):
 
39
    def test_data_port_name(self, mock_resolve_ports, config):
36
40
        self.test_config.set('data-port', 'br-data:em1')
 
41
        config.side_effect = self.test_config.get
37
42
        mock_resolve_ports.side_effect = lambda ports: ports
38
 
        self.assertEquals(context.DataPortContext()(),
39
 
                          {'br-data': 'em1'})
 
43
        self.assertEquals(
 
44
            charmhelpers.contrib.openstack.context.DataPortContext()(),
 
45
            {'br-data': 'em1'}
 
46
        )
40
47
 
41
 
    @patch.object(context, 'get_nic_hwaddr')
 
48
    @patch('charmhelpers.contrib.openstack.context.config')
42
49
    @patch('charmhelpers.contrib.openstack.context.get_nic_hwaddr')
43
50
    @patch('charmhelpers.contrib.openstack.context.list_nics')
44
 
    def test_data_port_mac(self, list_nics, get_nic_hwaddr, get_nic_hwaddr2):
 
51
    def test_data_port_mac(self, list_nics, get_nic_hwaddr, config):
45
52
        machine_machs = {
46
53
            'em1': 'aa:aa:aa:aa:aa:aa',
47
54
            'eth0': 'bb:bb:bb:bb:bb:bb',
48
55
        }
49
 
        get_nic_hwaddr2.side_effect = lambda nic: machine_machs[nic]
50
56
        absent_mac = "cc:cc:cc:cc:cc:cc"
51
57
        config_macs = ("br-d1:%s br-d2:%s" %
52
58
                       (absent_mac, machine_machs['em1']))
53
59
        self.test_config.set('data-port', config_macs)
 
60
        config.side_effect = self.test_config.get
54
61
        list_nics.return_value = machine_machs.keys()
55
62
        get_nic_hwaddr.side_effect = lambda nic: machine_machs[nic]
56
 
        self.assertEquals(context.DataPortContext()(),
57
 
                          {'br-d2': 'em1'})
58
 
 
59
 
    @patch('charmhelpers.contrib.openstack.context.NeutronPortContext.'
60
 
           'resolve_ports')
61
 
    def test_ensure_bridge_data_port_present(self, mock_resolve_ports):
62
 
        self.test_config.set('data-port', 'br-data:em1')
63
 
        self.test_config.set('bridge-mappings', 'phybr1:br-data')
64
 
 
65
 
        def add_port(bridge, port, promisc):
66
 
            if bridge == 'br-data' and port == 'em1' and promisc is True:
67
 
                self.bridge_added = True
68
 
                return
69
 
            self.bridge_added = False
70
 
 
71
 
        mock_resolve_ports.side_effect = lambda ports: ports
72
 
        self.add_bridge_port.side_effect = add_port
73
 
        context.OVSPluginContext()._ensure_bridge()
74
 
        self.assertEquals(self.bridge_added, True)
75
 
 
 
63
        self.assertEquals(
 
64
            charmhelpers.contrib.openstack.context.DataPortContext()(),
 
65
            {'br-d2': 'em1'}
 
66
        )
 
67
 
 
68
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_get')
 
69
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids')
 
70
    @patch.object(charmhelpers.contrib.openstack.context, 'related_units')
76
71
    @patch.object(charmhelpers.contrib.openstack.context, 'config')
77
72
    @patch.object(charmhelpers.contrib.openstack.context, 'unit_get')
78
73
    @patch.object(charmhelpers.contrib.openstack.context, 'is_clustered')
84
79
    @patch.object(charmhelpers.contrib.openstack.context, 'unit_private_ip')
85
80
    def test_neutroncc_context_api_rel(self, _unit_priv_ip, _npa, _ens_pkgs,
86
81
                                       _save_ff, _https, _is_clus, _unit_get,
87
 
                                       _config):
 
82
                                       _config, _runits, _rids, _rget):
88
83
        def mock_npa(plugin, section, manager):
89
84
            if section == "driver":
90
85
                return "neutron.randomdriver"
95
90
        _unit_get.return_value = '127.0.0.13'
96
91
        _unit_priv_ip.return_value = '127.0.0.14'
97
92
        _is_clus.return_value = False
98
 
        self.related_units.return_value = ['unit1']
99
 
        self.relation_ids.return_value = ['rid2']
100
 
        self.test_relation.set({'neutron-security-groups': 'True',
101
 
                                'l2-population': 'True',
102
 
                                'network-device-mtu': 1500,
103
 
                                'overlay-network-type': 'gre',
104
 
                                })
 
93
        _runits.return_value = ['unit1']
 
94
        _rids.return_value = ['rid2']
 
95
        rdata = {
 
96
            'neutron-security-groups': 'True',
 
97
            'l2-population': 'True',
 
98
            'network-device-mtu': 1500,
 
99
            'overlay-network-type': 'gre',
 
100
            'enable-dvr': 'True',
 
101
        }
 
102
        _rget.side_effect = lambda *args, **kwargs: rdata
105
103
        self.get_host_ip.return_value = '127.0.0.15'
106
 
        self.service_running.return_value = False
107
104
        napi_ctxt = context.OVSPluginContext()
108
105
        expect = {
109
106
            'neutron_alchemy_flags': {},
110
107
            'neutron_security_groups': True,
 
108
            'distributed_routing': True,
111
109
            'verbose': True,
112
110
            'local_ip': '127.0.0.15',
113
111
            'network_device_mtu': 1500,
126
124
            'vlan_ranges': 'physnet1:1000:2000',
127
125
        }
128
126
        self.assertEquals(expect, napi_ctxt())
129
 
        self.service_start.assertCalled()
130
127
 
 
128
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_get')
 
129
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids')
 
130
    @patch.object(charmhelpers.contrib.openstack.context, 'related_units')
131
131
    @patch.object(charmhelpers.contrib.openstack.context, 'config')
132
132
    @patch.object(charmhelpers.contrib.openstack.context, 'unit_get')
133
133
    @patch.object(charmhelpers.contrib.openstack.context, 'is_clustered')
142
142
                                                        _ens_pkgs, _save_ff,
143
143
                                                        _https, _is_clus,
144
144
                                                        _unit_get,
145
 
                                                        _config):
 
145
                                                        _config, _runits,
 
146
                                                        _rids, _rget):
146
147
        def mock_npa(plugin, section, manager):
147
148
            if section == "driver":
148
149
                return "neutron.randomdriver"
155
156
        _unit_priv_ip.return_value = '127.0.0.14'
156
157
        _is_clus.return_value = False
157
158
        self.test_config.set('disable-security-groups', True)
158
 
        self.related_units.return_value = ['unit1']
159
 
        self.relation_ids.return_value = ['rid2']
160
 
        self.test_relation.set({'neutron-security-groups': 'True',
161
 
                                'l2-population': 'True',
162
 
                                'network-device-mtu': 1500,
163
 
                                'overlay-network-type': 'gre',
164
 
                                })
 
159
        _runits.return_value = ['unit1']
 
160
        _rids.return_value = ['rid2']
 
161
        rdata = {
 
162
            'neutron-security-groups': 'True',
 
163
            'l2-population': 'True',
 
164
            'network-device-mtu': 1500,
 
165
            'overlay-network-type': 'gre',
 
166
        }
 
167
        _rget.side_effect = lambda *args, **kwargs: rdata
165
168
        self.get_host_ip.return_value = '127.0.0.15'
166
 
        self.service_running.return_value = False
167
169
        napi_ctxt = context.OVSPluginContext()
168
170
        expect = {
 
171
            'distributed_routing': False,
169
172
            'neutron_alchemy_flags': {},
170
173
            'neutron_security_groups': False,
171
174
            'verbose': True,
186
189
            'vlan_ranges': 'physnet1:1000:2000',
187
190
        }
188
191
        self.assertEquals(expect, napi_ctxt())
189
 
        self.service_start.assertCalled()
 
192
 
 
193
 
 
194
class L3AgentContextTest(CharmTestCase):
 
195
 
 
196
    def setUp(self):
 
197
        super(L3AgentContextTest, self).setUp(context, TO_PATCH)
 
198
        self.config.side_effect = self.test_config.get
 
199
 
 
200
    def tearDown(self):
 
201
        super(L3AgentContextTest, self).tearDown()
 
202
 
 
203
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_get')
 
204
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids')
 
205
    @patch.object(charmhelpers.contrib.openstack.context, 'related_units')
 
206
    def test_dvr_enabled(self, _runits, _rids, _rget):
 
207
        _runits.return_value = ['unit1']
 
208
        _rids.return_value = ['rid2']
 
209
        rdata = {
 
210
            'neutron-security-groups': 'True',
 
211
            'enable-dvr': 'True',
 
212
            'l2-population': 'True',
 
213
            'overlay-network-type': 'vxlan',
 
214
            'network-device-mtu': 1500,
 
215
        }
 
216
        _rget.side_effect = lambda *args, **kwargs: rdata
 
217
        self.assertEquals(context.L3AgentContext()(), {'agent_mode': 'dvr'})
 
218
 
 
219
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_get')
 
220
    @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids')
 
221
    @patch.object(charmhelpers.contrib.openstack.context, 'related_units')
 
222
    def test_dvr_disabled(self, _runits, _rids, _rget):
 
223
        _runits.return_value = ['unit1']
 
224
        _rids.return_value = ['rid2']
 
225
        rdata = {
 
226
            'neutron-security-groups': 'True',
 
227
            'enable-dvr': 'False',
 
228
            'l2-population': 'True',
 
229
            'overlay-network-type': 'vxlan',
 
230
            'network-device-mtu': 1500,
 
231
        }
 
232
        _rget.side_effect = lambda *args, **kwargs: rdata
 
233
        self.assertEquals(context.L3AgentContext()(), {'agent_mode': 'legacy'})
 
234
 
 
235
 
 
236
class DVRSharedSecretContext(CharmTestCase):
 
237
 
 
238
    def setUp(self):
 
239
        super(DVRSharedSecretContext, self).setUp(context,
 
240
                                                  TO_PATCH)
 
241
        self.config.side_effect = self.test_config.get
 
242
 
 
243
    @patch('os.path')
 
244
    @patch('uuid.uuid4')
 
245
    def test_secret_created_stored(self, _uuid4, _path):
 
246
        _path.exists.return_value = False
 
247
        _uuid4.return_value = 'secret_thing'
 
248
        with patch_open() as (_open, _file):
 
249
            self.assertEquals(context.get_shared_secret(),
 
250
                              'secret_thing')
 
251
            _open.assert_called_with(
 
252
                context.SHARED_SECRET.format('quantum'), 'w')
 
253
            _file.write.assert_called_with('secret_thing')
 
254
 
 
255
    @patch('os.path')
 
256
    def test_secret_retrieved(self, _path):
 
257
        _path.exists.return_value = True
 
258
        with patch_open() as (_open, _file):
 
259
            _file.read.return_value = 'secret_thing\n'
 
260
            self.assertEquals(context.get_shared_secret(),
 
261
                              'secret_thing')
 
262
            _open.assert_called_with(
 
263
                context.SHARED_SECRET.format('quantum'), 'r')
 
264
 
 
265
    @patch.object(context, 'NeutronAPIContext')
 
266
    @patch.object(context, 'get_shared_secret')
 
267
    def test_shared_secretcontext_dvr(self, _shared_secret,
 
268
                                      _NeutronAPIContext):
 
269
        _NeutronAPIContext.side_effect = fake_context({'enable_dvr': True})
 
270
        _shared_secret.return_value = 'secret_thing'
 
271
        self.resolve_address.return_value = '10.0.0.10'
 
272
        self.assertEquals(context.DVRSharedSecretContext()(),
 
273
                          {'shared_secret': 'secret_thing',
 
274
                           'local_ip': '10.0.0.10'})
 
275
 
 
276
    @patch.object(context, 'NeutronAPIContext')
 
277
    @patch.object(context, 'get_shared_secret')
 
278
    def test_shared_secretcontext_nodvr(self, _shared_secret,
 
279
                                        _NeutronAPIContext):
 
280
        _NeutronAPIContext.side_effect = fake_context({'enable_dvr': False})
 
281
        _shared_secret.return_value = 'secret_thing'
 
282
        self.resolve_address.return_value = '10.0.0.10'
 
283
        self.assertEquals(context.DVRSharedSecretContext()(), {})