← Back to branch summary

~ubuntu-branches/ubuntu/raring/cinder/raring-updates

« back to all changes in this revision

Viewing changes to cinder/tests/api/fakes.py

Tags: upstream-2013.1~g2
Import upstream version 2013.1~g2

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/api/fakes.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2010 OpenStack LLC.
 
4
# All Rights Reserved.
 
5
#
 
6
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
7
#    not use this file except in compliance with the License. You may obtain
 
8
#    a copy of the License at
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
12
#    Unless required by applicable law or agreed to in writing, software
 
13
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
14
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
15
#    License for the specific language governing permissions and limitations
 
16
#    under the License.
 
17
 
 
18
import uuid
 
19
 
 
20
import routes
 
21
import webob
 
22
import webob.dec
 
23
import webob.request
 
24
 
 
25
from cinder.api.middleware import auth
 
26
from cinder.api.middleware import fault
 
27
from cinder.api.openstack import wsgi as os_wsgi
 
28
from cinder.api import urlmap
 
29
from cinder.api.v2 import limits
 
30
from cinder.api.v2 import router
 
31
from cinder.api import versions
 
32
from cinder import context
 
33
from cinder.openstack.common import timeutils
 
34
from cinder import wsgi
 
35
 
 
36
 
 
37
FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
 
38
FAKE_UUIDS = {}
 
39
 
 
40
 
 
41
class Context(object):
 
42
    pass
 
43
 
 
44
 
 
45
class FakeRouter(wsgi.Router):
 
46
    def __init__(self, ext_mgr=None):
 
47
        pass
 
48
 
 
49
    @webob.dec.wsgify
 
50
    def __call__(self, req):
 
51
        res = webob.Response()
 
52
        res.status = '200'
 
53
        res.headers['X-Test-Success'] = 'True'
 
54
        return res
 
55
 
 
56
 
 
57
@webob.dec.wsgify
 
58
def fake_wsgi(self, req):
 
59
    return self.application
 
60
 
 
61
 
 
62
def wsgi_app(inner_app_v2=None, fake_auth=True, fake_auth_context=None,
 
63
             use_no_auth=False, ext_mgr=None):
 
64
    if not inner_app_v2:
 
65
        inner_app_v2 = router.APIRouter(ext_mgr)
 
66
 
 
67
    if fake_auth:
 
68
        if fake_auth_context is not None:
 
69
            ctxt = fake_auth_context
 
70
        else:
 
71
            ctxt = context.RequestContext('fake', 'fake', auth_token=True)
 
72
        api_v2 = fault.FaultWrapper(auth.InjectContext(ctxt,
 
73
                                                       inner_app_v2))
 
74
    elif use_no_auth:
 
75
        api_v2 = fault.FaultWrapper(auth.NoAuthMiddleware(
 
76
            limits.RateLimitingMiddleware(inner_app_v2)))
 
77
    else:
 
78
        api_v2 = fault.FaultWrapper(auth.AuthMiddleware(
 
79
            limits.RateLimitingMiddleware(inner_app_v2)))
 
80
 
 
81
    mapper = urlmap.URLMap()
 
82
    mapper['/v2'] = api_v2
 
83
    mapper['/'] = fault.FaultWrapper(versions.Versions())
 
84
    return mapper
 
85
 
 
86
 
 
87
def stub_out_rate_limiting(stubs):
 
88
    def fake_rate_init(self, app):
 
89
        # super(limits.RateLimitingMiddleware, self).__init__(app)
 
90
        self.application = app
 
91
 
 
92
    # FIXME(ja): unsure about limits in volumes
 
93
    # stubs.Set(cinder.api.openstack.compute.limits.RateLimitingMiddleware,
 
94
    #     '__init__', fake_rate_init)
 
95
 
 
96
    # stubs.Set(cinder.api.openstack.compute.limits.RateLimitingMiddleware,
 
97
    #     '__call__', fake_wsgi)
 
98
 
 
99
 
 
100
class FakeToken(object):
 
101
    id_count = 0
 
102
 
 
103
    def __getitem__(self, key):
 
104
        return getattr(self, key)
 
105
 
 
106
    def __init__(self, **kwargs):
 
107
        FakeToken.id_count += 1
 
108
        self.id = FakeToken.id_count
 
109
        for k, v in kwargs.iteritems():
 
110
            setattr(self, k, v)
 
111
 
 
112
 
 
113
class FakeRequestContext(context.RequestContext):
 
114
    def __init__(self, *args, **kwargs):
 
115
        kwargs['auth_token'] = kwargs.get('auth_token', 'fake_auth_token')
 
116
        return super(FakeRequestContext, self).__init__(*args, **kwargs)
 
117
 
 
118
 
 
119
class HTTPRequest(webob.Request):
 
120
 
 
121
    @classmethod
 
122
    def blank(cls, *args, **kwargs):
 
123
        kwargs['base_url'] = 'http://localhost/v1'
 
124
        use_admin_context = kwargs.pop('use_admin_context', False)
 
125
        out = webob.Request.blank(*args, **kwargs)
 
126
        out.environ['cinder.context'] = FakeRequestContext(
 
127
            'fake_user',
 
128
            'fake',
 
129
            is_admin=use_admin_context)
 
130
        return out
 
131
 
 
132
 
 
133
class TestRouter(wsgi.Router):
 
134
    def __init__(self, controller):
 
135
        mapper = routes.Mapper()
 
136
        mapper.resource("test", "tests",
 
137
                        controller=os_wsgi.Resource(controller))
 
138
        super(TestRouter, self).__init__(mapper)
 
139
 
 
140
 
 
141
class FakeAuthDatabase(object):
 
142
    data = {}
 
143
 
 
144
    @staticmethod
 
145
    def auth_token_get(context, token_hash):
 
146
        return FakeAuthDatabase.data.get(token_hash, None)
 
147
 
 
148
    @staticmethod
 
149
    def auth_token_create(context, token):
 
150
        fake_token = FakeToken(created_at=timeutils.utcnow(), **token)
 
151
        FakeAuthDatabase.data[fake_token.token_hash] = fake_token
 
152
        FakeAuthDatabase.data['id_%i' % fake_token.id] = fake_token
 
153
        return fake_token
 
154
 
 
155
    @staticmethod
 
156
    def auth_token_destroy(context, token_id):
 
157
        token = FakeAuthDatabase.data.get('id_%i' % token_id)
 
158
        if token and token.token_hash in FakeAuthDatabase.data:
 
159
            del FakeAuthDatabase.data[token.token_hash]
 
160
            del FakeAuthDatabase.data['id_%i' % token_id]
 
161
 
 
162
 
 
163
class FakeRateLimiter(object):
 
164
    def __init__(self, application):
 
165
        self.application = application
 
166
 
 
167
    @webob.dec.wsgify
 
168
    def __call__(self, req):
 
169
        return self.application
 
170
 
 
171
 
 
172
def get_fake_uuid(token=0):
 
173
    if not token in FAKE_UUIDS:
 
174
        FAKE_UUIDS[token] = str(uuid.uuid4())
 
175
    return FAKE_UUIDS[token]