← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/ml2/test_extension_driver_api.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/ml2/test_extension_driver_api.py
10
10
#    License for the specific language governing permissions and limitations
11
11
#    under the License.
12
12
 
13
 
from neutron.api import extensions
 
13
import contextlib
 
14
import mock
 
15
 
 
16
from neutron import context
 
17
from neutron import manager
14
18
from neutron.plugins.ml2 import config
15
 
from neutron.plugins.ml2 import driver_api as api
16
 
from neutron.tests.unit.ml2 import extensions as test_extensions
 
19
from neutron.tests.unit.ml2.drivers import ext_test
17
20
from neutron.tests.unit.ml2 import test_ml2_plugin
18
21
 
19
22
 
26
29
                                     self._extension_drivers,
27
30
                                     group='ml2')
28
31
        super(ExtensionDriverTestCase, self).setUp()
 
32
        self._plugin = manager.NeutronManager.get_plugin()
 
33
        self._ctxt = context.get_admin_context()
29
34
 
30
35
    def test_network_attr(self):
31
36
        with self.network() as network:
 
37
            # Test create network
32
38
            ent = network['network'].get('network_extension')
33
39
            self.assertIsNotNone(ent)
34
40
 
 
41
            # Test list networks
 
42
            res = self._list('networks')
 
43
            val = res['networks'][0].get('network_extension')
 
44
            self.assertEqual('Test_Network_Extension_extend', val)
 
45
 
 
46
            # Test network update
 
47
            data = {'network':
 
48
                    {'network_extension': 'Test_Network_Extension_Update'}}
 
49
            res = self._update('networks', network['network']['id'], data)
 
50
            val = res['network'].get('network_extension')
 
51
            self.assertEqual('Test_Network_Extension_Update_update', val)
 
52
 
35
53
    def test_subnet_attr(self):
36
54
        with self.subnet() as subnet:
 
55
            # Test create subnet
37
56
            ent = subnet['subnet'].get('subnet_extension')
38
57
            self.assertIsNotNone(ent)
39
58
 
 
59
            # Test list subnets
 
60
            res = self._list('subnets')
 
61
            val = res['subnets'][0].get('subnet_extension')
 
62
            self.assertEqual('Test_Subnet_Extension_extend', val)
 
63
 
 
64
            # Test subnet update
 
65
            data = {'subnet':
 
66
                    {'subnet_extension': 'Test_Subnet_Extension_Update'}}
 
67
            res = self._update('subnets', subnet['subnet']['id'], data)
 
68
            val = res['subnet'].get('subnet_extension')
 
69
            self.assertEqual('Test_Subnet_Extension_Update_update', val)
 
70
 
40
71
    def test_port_attr(self):
41
72
        with self.port() as port:
 
73
            # Test create port
42
74
            ent = port['port'].get('port_extension')
43
75
            self.assertIsNotNone(ent)
44
76
 
45
 
 
46
 
class TestExtensionDriver(api.ExtensionDriver):
47
 
    _supported_extension_alias = 'test_extension'
48
 
 
49
 
    def initialize(self):
50
 
        self.network_extension = 'Test_Network_Extension'
51
 
        self.subnet_extension = 'Test_Subnet_Extension'
52
 
        self.port_extension = 'Test_Port_Extension'
53
 
        extensions.append_api_extensions_path(test_extensions.__path__)
54
 
 
55
 
    @property
56
 
    def extension_alias(self):
57
 
        return self._supported_extension_alias
58
 
 
59
 
    def process_create_network(self, session, data, result):
60
 
        result['network_extension'] = self.network_extension
61
 
 
62
 
    def process_create_subnet(self, session, data, result):
63
 
        result['subnet_extension'] = self.subnet_extension
64
 
 
65
 
    def process_create_port(self, session, data, result):
66
 
        result['port_extension'] = self.port_extension
 
77
            # Test list ports
 
78
            res = self._list('ports')
 
79
            val = res['ports'][0].get('port_extension')
 
80
            self.assertEqual('Test_Port_Extension_extend', val)
 
81
 
 
82
            # Test port update
 
83
            data = {'port': {'port_extension': 'Test_Port_Extension_Update'}}
 
84
            res = self._update('ports', port['port']['id'], data)
 
85
            val = res['port'].get('port_extension')
 
86
            self.assertEqual('Test_Port_Extension_Update_update', val)
 
87
 
 
88
    def test_extend_network_dict(self):
 
89
        with contextlib.nested(
 
90
            mock.patch.object(ext_test.TestExtensionDriver,
 
91
                              'process_update_network'),
 
92
            mock.patch.object(ext_test.TestExtensionDriver,
 
93
                              'extend_network_dict'),
 
94
            self.network()
 
95
        ) as (ext_update_net, ext_net_dict, network):
 
96
            net_id = network['network']['id']
 
97
            net_data = {'network': {'id': net_id}}
 
98
            self._plugin.update_network(self._ctxt, net_id, net_data)
 
99
            self.assertTrue(ext_update_net.called)
 
100
            self.assertTrue(ext_net_dict.called)
 
101
 
 
102
    def test_extend_subnet_dict(self):
 
103
        with contextlib.nested(
 
104
            mock.patch.object(ext_test.TestExtensionDriver,
 
105
                              'process_update_subnet'),
 
106
            mock.patch.object(ext_test.TestExtensionDriver,
 
107
                              'extend_subnet_dict'),
 
108
            self.subnet()
 
109
        ) as (ext_update_subnet, ext_subnet_dict, subnet):
 
110
            subnet_id = subnet['subnet']['id']
 
111
            subnet_data = {'subnet': {'id': subnet_id}}
 
112
            self._plugin.update_subnet(self._ctxt, subnet_id, subnet_data)
 
113
            self.assertTrue(ext_update_subnet.called)
 
114
            self.assertTrue(ext_subnet_dict.called)
 
115
 
 
116
    def test_extend_port_dict(self):
 
117
        with contextlib.nested(
 
118
            mock.patch.object(ext_test.TestExtensionDriver,
 
119
                              'process_update_port'),
 
120
            mock.patch.object(ext_test.TestExtensionDriver,
 
121
                              'extend_port_dict'),
 
122
            self.port()
 
123
        ) as (ext_update_port, ext_port_dict, port):
 
124
            port_id = port['port']['id']
 
125
            port_data = {'port': {'id': port_id}}
 
126
            self._plugin.update_port(self._ctxt, port_id, port_data)
 
127
            self.assertTrue(ext_update_port.called)
 
128
            self.assertTrue(ext_port_dict.called)
 
129
 
 
130
 
 
131
class DBExtensionDriverTestCase(test_ml2_plugin.Ml2PluginV2TestCase):
 
132
    _extension_drivers = ['testdb']
 
133
 
 
134
    def setUp(self):
 
135
        config.cfg.CONF.set_override('extension_drivers',
 
136
                                     self._extension_drivers,
 
137
                                     group='ml2')
 
138
        super(DBExtensionDriverTestCase, self).setUp()
 
139
        self._plugin = manager.NeutronManager.get_plugin()
 
140
        self._ctxt = context.get_admin_context()
 
141
 
 
142
    def test_network_attr(self):
 
143
        with self.network() as network:
 
144
            # Test create with default value.
 
145
            net_id = network['network']['id']
 
146
            val = network['network']['network_extension']
 
147
            self.assertEqual("", val)
 
148
            res = self._show('networks', net_id)
 
149
            val = res['network']['network_extension']
 
150
            self.assertEqual("", val)
 
151
 
 
152
            # Test list.
 
153
            res = self._list('networks')
 
154
            val = res['networks'][0]['network_extension']
 
155
            self.assertEqual("", val)
 
156
 
 
157
        # Test create with explict value.
 
158
        res = self._create_network(self.fmt,
 
159
                                   'test-network', True,
 
160
                                   arg_list=('network_extension', ),
 
161
                                   network_extension="abc")
 
162
        network = self.deserialize(self.fmt, res)
 
163
        net_id = network['network']['id']
 
164
        val = network['network']['network_extension']
 
165
        self.assertEqual("abc", val)
 
166
        res = self._show('networks', net_id)
 
167
        val = res['network']['network_extension']
 
168
        self.assertEqual("abc", val)
 
169
 
 
170
        # Test update.
 
171
        data = {'network': {'network_extension': "def"}}
 
172
        res = self._update('networks', net_id, data)
 
173
        val = res['network']['network_extension']
 
174
        self.assertEqual("def", val)
 
175
        res = self._show('networks', net_id)
 
176
        val = res['network']['network_extension']
 
177
        self.assertEqual("def", val)
 
178
 
 
179
    def test_subnet_attr(self):
 
180
        with self.subnet() as subnet:
 
181
            # Test create with default value.
 
182
            net_id = subnet['subnet']['id']
 
183
            val = subnet['subnet']['subnet_extension']
 
184
            self.assertEqual("", val)
 
185
            res = self._show('subnets', net_id)
 
186
            val = res['subnet']['subnet_extension']
 
187
            self.assertEqual("", val)
 
188
 
 
189
            # Test list.
 
190
            res = self._list('subnets')
 
191
            val = res['subnets'][0]['subnet_extension']
 
192
            self.assertEqual("", val)
 
193
 
 
194
        with self.network() as network:
 
195
            # Test create with explict value.
 
196
            data = {'subnet':
 
197
                    {'network_id': network['network']['id'],
 
198
                     'cidr': '10.1.0.0/24',
 
199
                     'ip_version': '4',
 
200
                     'tenant_id': self._tenant_id,
 
201
                     'subnet_extension': 'abc'}}
 
202
            req = self.new_create_request('subnets', data, self.fmt)
 
203
            res = req.get_response(self.api)
 
204
            subnet = self.deserialize(self.fmt, res)
 
205
            subnet_id = subnet['subnet']['id']
 
206
            val = subnet['subnet']['subnet_extension']
 
207
            self.assertEqual("abc", val)
 
208
            res = self._show('subnets', subnet_id)
 
209
            val = res['subnet']['subnet_extension']
 
210
            self.assertEqual("abc", val)
 
211
 
 
212
            # Test update.
 
213
            data = {'subnet': {'subnet_extension': "def"}}
 
214
            res = self._update('subnets', subnet_id, data)
 
215
            val = res['subnet']['subnet_extension']
 
216
            self.assertEqual("def", val)
 
217
            res = self._show('subnets', subnet_id)
 
218
            val = res['subnet']['subnet_extension']
 
219
            self.assertEqual("def", val)
 
220
 
 
221
    def test_port_attr(self):
 
222
        with self.port() as port:
 
223
            # Test create with default value.
 
224
            net_id = port['port']['id']
 
225
            val = port['port']['port_extension']
 
226
            self.assertEqual("", val)
 
227
            res = self._show('ports', net_id)
 
228
            val = res['port']['port_extension']
 
229
            self.assertEqual("", val)
 
230
 
 
231
            # Test list.
 
232
            res = self._list('ports')
 
233
            val = res['ports'][0]['port_extension']
 
234
            self.assertEqual("", val)
 
235
 
 
236
        with self.network() as network:
 
237
            # Test create with explict value.
 
238
            res = self._create_port(self.fmt,
 
239
                                    network['network']['id'],
 
240
                                    arg_list=('port_extension', ),
 
241
                                    port_extension="abc")
 
242
            port = self.deserialize(self.fmt, res)
 
243
            port_id = port['port']['id']
 
244
            val = port['port']['port_extension']
 
245
            self.assertEqual("abc", val)
 
246
            res = self._show('ports', port_id)
 
247
            val = res['port']['port_extension']
 
248
            self.assertEqual("abc", val)
 
249
 
 
250
            # Test update.
 
251
            data = {'port': {'port_extension': "def"}}
 
252
            res = self._update('ports', port_id, data)
 
253
            val = res['port']['port_extension']
 
254
            self.assertEqual("def", val)
 
255
            res = self._show('ports', port_id)
 
256
            val = res['port']['port_extension']
 
257
            self.assertEqual("def", val)