← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/agent/l3/test_l3_router.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/agent/l3/test_l3_router.py
 
1
# Copyright (c) 2015 Openstack Foundation
 
2
#
 
3
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
4
#    not use this file except in compliance with the License. You may obtain
 
5
#    a copy of the License at
 
6
#
 
7
#         http://www.apache.org/licenses/LICENSE-2.0
 
8
#
 
9
#    Unless required by applicable law or agreed to in writing, software
 
10
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
11
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
12
#    License for the specific language governing permissions and limitations
 
13
#    under the License.
 
14
 
 
15
import mock
 
16
 
 
17
from neutron.agent.l3 import router_info
 
18
from neutron.agent.linux import ip_lib
 
19
from neutron.common import constants as l3_constants
 
20
from neutron.common import exceptions as n_exc
 
21
from neutron.openstack.common import uuidutils
 
22
from neutron.tests import base
 
23
 
 
24
_uuid = uuidutils.generate_uuid
 
25
 
 
26
 
 
27
class BasicRouterTestCaseFramework(base.BaseTestCase):
 
28
    def _create_router(self, router=None, **kwargs):
 
29
        if not router:
 
30
            router = mock.MagicMock()
 
31
        self.agent_conf = mock.Mock()
 
32
        # NOTE The use_namespaces config will soon be deprecated
 
33
        self.agent_conf.use_namespaces = True
 
34
        self.router_id = _uuid()
 
35
        return router_info.RouterInfo(self.router_id,
 
36
                                      router,
 
37
                                      self.agent_conf,
 
38
                                      mock.sentinel.interface_driver,
 
39
                                      **kwargs)
 
40
 
 
41
 
 
42
class TestBasicRouterOperations(BasicRouterTestCaseFramework):
 
43
 
 
44
    def test_get_floating_ips(self):
 
45
        router = mock.MagicMock()
 
46
        router.get.return_value = [mock.sentinel.floating_ip]
 
47
        ri = self._create_router(router)
 
48
 
 
49
        fips = ri.get_floating_ips()
 
50
 
 
51
        self.assertEqual([mock.sentinel.floating_ip], fips)
 
52
 
 
53
    def test_process_floating_ip_nat_rules(self):
 
54
        ri = self._create_router()
 
55
        fips = [{'fixed_ip_address': mock.sentinel.ip,
 
56
                 'floating_ip_address': mock.sentinel.fip}]
 
57
        ri.get_floating_ips = mock.Mock(return_value=fips)
 
58
        ri.iptables_manager = mock.MagicMock()
 
59
        ipv4_nat = ri.iptables_manager.ipv4['nat']
 
60
        ri.floating_forward_rules = mock.Mock(
 
61
            return_value=[(mock.sentinel.chain, mock.sentinel.rule)])
 
62
 
 
63
        ri.process_floating_ip_nat_rules()
 
64
 
 
65
        # Be sure that the rules are cleared first and apply is called last
 
66
        self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
 
67
                         ipv4_nat.mock_calls[0])
 
68
        self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
 
69
 
 
70
        # Be sure that add_rule is called somewhere in the middle
 
71
        ipv4_nat.add_rule.assert_called_once_with(mock.sentinel.chain,
 
72
                                                  mock.sentinel.rule,
 
73
                                                  tag='floating_ip')
 
74
 
 
75
    def test_process_floating_ip_nat_rules_removed(self):
 
76
        ri = self._create_router()
 
77
        ri.get_floating_ips = mock.Mock(return_value=[])
 
78
        ri.iptables_manager = mock.MagicMock()
 
79
        ipv4_nat = ri.iptables_manager.ipv4['nat']
 
80
 
 
81
        ri.process_floating_ip_nat_rules()
 
82
 
 
83
        # Be sure that the rules are cleared first and apply is called last
 
84
        self.assertEqual(mock.call.clear_rules_by_tag('floating_ip'),
 
85
                         ipv4_nat.mock_calls[0])
 
86
        self.assertEqual(mock.call.apply(), ri.iptables_manager.mock_calls[-1])
 
87
 
 
88
        # Be sure that add_rule is called somewhere in the middle
 
89
        self.assertFalse(ipv4_nat.add_rule.called)
 
90
 
 
91
    def _test_add_fip_addr_to_device_error(self, device):
 
92
        ri = self._create_router()
 
93
        ip = '15.1.2.3'
 
94
 
 
95
        result = ri._add_fip_addr_to_device(
 
96
            {'id': mock.sentinel.id, 'floating_ip_address': ip}, device)
 
97
 
 
98
        device.addr.add.assert_called_with(ip + '/32')
 
99
        return result
 
100
 
 
101
    def test__add_fip_addr_to_device(self):
 
102
        result = self._test_add_fip_addr_to_device_error(mock.Mock())
 
103
        self.assertTrue(result)
 
104
 
 
105
    def test__add_fip_addr_to_device_error(self):
 
106
        device = mock.Mock()
 
107
        device.addr.add.side_effect = RuntimeError
 
108
        result = self._test_add_fip_addr_to_device_error(device)
 
109
        self.assertFalse(result)
 
110
 
 
111
    def test_process_snat_dnat_for_fip(self):
 
112
        ri = self._create_router()
 
113
        ri.process_floating_ip_nat_rules = mock.Mock(side_effect=Exception)
 
114
 
 
115
        self.assertRaises(n_exc.FloatingIpSetupException,
 
116
                          ri.process_snat_dnat_for_fip)
 
117
 
 
118
        ri.process_floating_ip_nat_rules.assert_called_once_with()
 
119
 
 
120
    def test_put_fips_in_error_state(self):
 
121
        ri = self._create_router()
 
122
        ri.router = mock.Mock()
 
123
        ri.router.get.return_value = [{'id': mock.sentinel.id1},
 
124
                                      {'id': mock.sentinel.id2}]
 
125
 
 
126
        statuses = ri.put_fips_in_error_state()
 
127
 
 
128
        expected = [{mock.sentinel.id1: l3_constants.FLOATINGIP_STATUS_ERROR,
 
129
                     mock.sentinel.id2: l3_constants.FLOATINGIP_STATUS_ERROR}]
 
130
        self.assertNotEqual(expected, statuses)
 
131
 
 
132
    def test_configure_fip_addresses(self):
 
133
        ri = self._create_router()
 
134
        ri.process_floating_ip_addresses = mock.Mock(
 
135
            side_effect=Exception)
 
136
 
 
137
        self.assertRaises(n_exc.FloatingIpSetupException,
 
138
                          ri.configure_fip_addresses,
 
139
                          mock.sentinel.interface_name)
 
140
 
 
141
        ri.process_floating_ip_addresses.assert_called_once_with(
 
142
            mock.sentinel.interface_name)
 
143
 
 
144
    def test_get_router_cidrs_returns_cidrs(self):
 
145
        ri = self._create_router()
 
146
        addresses = ['15.1.2.2/24', '15.1.2.3/32']
 
147
        device = mock.MagicMock()
 
148
        device.addr.list.return_value = [{'cidr': addresses[0]},
 
149
                                         {'cidr': addresses[1]}]
 
150
        self.assertEqual(set(addresses), ri.get_router_cidrs(device))
 
151
 
 
152
 
 
153
@mock.patch.object(ip_lib, 'IPDevice')
 
154
class TestFloatingIpWithMockDevice(BasicRouterTestCaseFramework):
 
155
 
 
156
    def test_process_floating_ip_addresses_remap(self, IPDevice):
 
157
        fip_id = _uuid()
 
158
        fip = {
 
159
            'id': fip_id, 'port_id': _uuid(),
 
160
            'floating_ip_address': '15.1.2.3',
 
161
            'fixed_ip_address': '192.168.0.2'
 
162
        }
 
163
 
 
164
        IPDevice.return_value = device = mock.Mock()
 
165
        device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
 
166
        ri = self._create_router()
 
167
        ri.get_floating_ips = mock.Mock(return_value=[fip])
 
168
 
 
169
        fip_statuses = ri.process_floating_ip_addresses(
 
170
            mock.sentinel.interface_name)
 
171
        self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ACTIVE},
 
172
                         fip_statuses)
 
173
 
 
174
        self.assertFalse(device.addr.add.called)
 
175
        self.assertFalse(device.addr.delete.called)
 
176
 
 
177
    def test_process_router_with_disabled_floating_ip(self, IPDevice):
 
178
        fip_id = _uuid()
 
179
        fip = {
 
180
            'id': fip_id, 'port_id': _uuid(),
 
181
            'floating_ip_address': '15.1.2.3',
 
182
            'fixed_ip_address': '192.168.0.2'
 
183
        }
 
184
 
 
185
        ri = self._create_router()
 
186
        ri.floating_ips = [fip]
 
187
        ri.get_floating_ips = mock.Mock(return_value=[])
 
188
 
 
189
        fip_statuses = ri.process_floating_ip_addresses(
 
190
            mock.sentinel.interface_name)
 
191
 
 
192
        self.assertIsNone(fip_statuses.get(fip_id))
 
193
 
 
194
    def test_process_router_floating_ip_with_device_add_error(self, IPDevice):
 
195
        IPDevice.return_value = device = mock.Mock(side_effect=RuntimeError)
 
196
        device.addr.list.return_value = []
 
197
        fip_id = _uuid()
 
198
        fip = {
 
199
            'id': fip_id, 'port_id': _uuid(),
 
200
            'floating_ip_address': '15.1.2.3',
 
201
            'fixed_ip_address': '192.168.0.2'
 
202
        }
 
203
        ri = self._create_router()
 
204
        ri.add_floating_ip = mock.Mock(
 
205
            return_value=l3_constants.FLOATINGIP_STATUS_ERROR)
 
206
        ri.get_floating_ips = mock.Mock(return_value=[fip])
 
207
 
 
208
        fip_statuses = ri.process_floating_ip_addresses(
 
209
            mock.sentinel.interface_name)
 
210
 
 
211
        self.assertEqual({fip_id: l3_constants.FLOATINGIP_STATUS_ERROR},
 
212
                         fip_statuses)
 
213
 
 
214
    # TODO(mrsmith): refactor for DVR cases
 
215
    def test_process_floating_ip_addresses_remove(self, IPDevice):
 
216
        IPDevice.return_value = device = mock.Mock()
 
217
        device.addr.list.return_value = [{'cidr': '15.1.2.3/32'}]
 
218
 
 
219
        ri = self._create_router()
 
220
        ri.remove_floating_ip = mock.Mock()
 
221
        ri.router.get = mock.Mock(return_value=[])
 
222
 
 
223
        fip_statuses = ri.process_floating_ip_addresses(
 
224
            mock.sentinel.interface_name)
 
225
        self.assertEqual({}, fip_statuses)
 
226
        ri.remove_floating_ip.assert_called_once_with(device, '15.1.2.3/32')