← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/test_extension_extended_attribute.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2015-04-15 13:59:07 UTC
  • mfrom: (1.1.22)
  • Revision ID: package-import@ubuntu.com-20150415135907-z10fr18evag1ozq3
Tags: 1:2015.1~rc1-0ubuntu1
* New upstream milestone release:
  - debian/control: Update dependencies. 
  - debian/patches/disable-udev-tests.patch: Dropped no longer needed.
  - debian/patches/fixup-driver-test-execution.patch: Dropped no longer needed.
  - debian/patches/skip-iptest.patch: Skip failing test
  - debian/neutron-plugin-openvswitch-agent.install: Added neutron-ovsvapp-agent binary.
  - debian/neutron-plugin-cisco.install: Added neutron-cisco-apic-service-agent and 
    neutron-cisco-apic-host-agent

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/test_extension_extended_attribute.py
1
 
# Copyright 2013 VMware, Inc
2
 
# All Rights Reserved.
3
 
#
4
 
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
5
 
#    not use this file except in compliance with the License. You may obtain
6
 
#    a copy of the License at
7
 
#
8
 
#         http://www.apache.org/licenses/LICENSE-2.0
9
 
#
10
 
#    Unless required by applicable law or agreed to in writing, software
11
 
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12
 
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13
 
#    License for the specific language governing permissions and limitations
14
 
#    under the License.
15
 
 
16
 
"""
17
 
Unit tests for extension extended attribute
18
 
"""
19
 
 
20
 
from oslo_config import cfg
21
 
import webob.exc as webexc
22
 
 
23
 
import neutron
24
 
from neutron.api import extensions
25
 
from neutron.api.v2 import attributes
26
 
from neutron.common import config
27
 
from neutron import manager
28
 
from neutron.plugins.common import constants
29
 
from neutron.plugins.ml2 import plugin as ml2_plugin
30
 
from neutron import quota
31
 
from neutron.tests import base
32
 
from neutron.tests.unit.extensions import extendedattribute as extattr
33
 
from neutron.tests.unit import test_api_v2
34
 
from neutron.tests.unit import testlib_api
35
 
from neutron.tests.unit import testlib_plugin
36
 
from neutron import wsgi
37
 
 
38
 
_uuid = test_api_v2._uuid
39
 
_get_path = test_api_v2._get_path
40
 
extensions_path = ':'.join(neutron.tests.unit.extensions.__path__)
41
 
 
42
 
 
43
 
class ExtensionExtendedAttributeTestPlugin(
44
 
    ml2_plugin.Ml2Plugin):
45
 
 
46
 
    supported_extension_aliases = [
47
 
        'ext-obj-test', "extended-ext-attr"
48
 
    ]
49
 
 
50
 
    def __init__(self, configfile=None):
51
 
        super(ExtensionExtendedAttributeTestPlugin, self)
52
 
        self.objs = []
53
 
        self.objh = {}
54
 
 
55
 
    def create_ext_test_resource(self, context, ext_test_resource):
56
 
        obj = ext_test_resource['ext_test_resource']
57
 
        id = _uuid()
58
 
        obj['id'] = id
59
 
        self.objs.append(obj)
60
 
        self.objh.update({id: obj})
61
 
        return obj
62
 
 
63
 
    def get_ext_test_resources(self, context, filters=None, fields=None):
64
 
        return self.objs
65
 
 
66
 
    def get_ext_test_resource(self, context, id, fields=None):
67
 
        return self.objh[id]
68
 
 
69
 
 
70
 
class ExtensionExtendedAttributeTestCase(base.BaseTestCase,
71
 
                                         testlib_plugin.PluginSetupHelper):
72
 
    def setUp(self):
73
 
        super(ExtensionExtendedAttributeTestCase, self).setUp()
74
 
        plugin = (
75
 
            "neutron.tests.unit.test_extension_extended_attribute."
76
 
            "ExtensionExtendedAttributeTestPlugin"
77
 
        )
78
 
 
79
 
        # point config file to: neutron/tests/etc/neutron.conf.test
80
 
        self.config_parse()
81
 
 
82
 
        self.setup_coreplugin(plugin)
83
 
 
84
 
        ext_mgr = extensions.PluginAwareExtensionManager(
85
 
            extensions_path,
86
 
            {constants.CORE: ExtensionExtendedAttributeTestPlugin}
87
 
        )
88
 
        ext_mgr.extend_resources("2.0", {})
89
 
        extensions.PluginAwareExtensionManager._instance = ext_mgr
90
 
 
91
 
        app = config.load_paste_app('extensions_test_app')
92
 
        self._api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
93
 
 
94
 
        self._tenant_id = "8c70909f-b081-452d-872b-df48e6c355d1"
95
 
        # Save the global RESOURCE_ATTRIBUTE_MAP
96
 
        self.saved_attr_map = {}
97
 
        for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
98
 
            self.saved_attr_map[resource] = attrs.copy()
99
 
        # Add the resources to the global attribute map
100
 
        # This is done here as the setup process won't
101
 
        # initialize the main API router which extends
102
 
        # the global attribute map
103
 
        attributes.RESOURCE_ATTRIBUTE_MAP.update(
104
 
            extattr.EXTENDED_ATTRIBUTES_2_0)
105
 
        self.agentscheduler_dbMinxin = manager.NeutronManager.get_plugin()
106
 
        self.addCleanup(self.restore_attribute_map)
107
 
 
108
 
        quota.QUOTAS._driver = None
109
 
        cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
110
 
                              group='QUOTAS')
111
 
 
112
 
    def restore_attribute_map(self):
113
 
        # Restore the original RESOURCE_ATTRIBUTE_MAP
114
 
        attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
115
 
 
116
 
    def _do_request(self, method, path, data=None, params=None, action=None):
117
 
        content_type = 'application/json'
118
 
        body = None
119
 
        if data is not None:  # empty dict is valid
120
 
            body = wsgi.Serializer().serialize(data, content_type)
121
 
 
122
 
        req = testlib_api.create_request(
123
 
            path, body, content_type,
124
 
            method, query_string=params)
125
 
        res = req.get_response(self._api)
126
 
        if res.status_code >= 400:
127
 
            raise webexc.HTTPClientError(detail=res.body, code=res.status_code)
128
 
        if res.status_code != webexc.HTTPNoContent.code:
129
 
            return res.json
130
 
 
131
 
    def _ext_test_resource_create(self, attr=None):
132
 
        data = {
133
 
            "ext_test_resource": {
134
 
                "tenant_id": self._tenant_id,
135
 
                "name": "test",
136
 
                extattr.EXTENDED_ATTRIBUTE: attr
137
 
            }
138
 
        }
139
 
 
140
 
        res = self._do_request('POST', _get_path('ext_test_resources'), data)
141
 
        return res['ext_test_resource']
142
 
 
143
 
    def test_ext_test_resource_create(self):
144
 
        ext_test_resource = self._ext_test_resource_create()
145
 
        attr = _uuid()
146
 
        ext_test_resource = self._ext_test_resource_create(attr)
147
 
        self.assertEqual(ext_test_resource[extattr.EXTENDED_ATTRIBUTE], attr)
148
 
 
149
 
    def test_ext_test_resource_get(self):
150
 
        attr = _uuid()
151
 
        obj = self._ext_test_resource_create(attr)
152
 
        obj_id = obj['id']
153
 
        res = self._do_request('GET', _get_path(
154
 
            'ext_test_resources/{0}'.format(obj_id)))
155
 
        obj2 = res['ext_test_resource']
156
 
        self.assertEqual(obj2[extattr.EXTENDED_ATTRIBUTE], attr)