1
# Copyright 2013 OpenStack Foundation
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
8
# http://www.apache.org/licenses/LICENSE-2.0
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
17
A fake server that "responds" to API methods with pre-canned responses.
19
All of these responses come from the spec, so if for some reason the spec's
20
wrong the tests might raise AssertionError. I've indicated in comments the
21
places where actual behavior differs from the spec.
24
# W0102: Dangerous default value %s as argument
25
# pylint: disable=W0102
31
from ceilometerclient.openstack.common.apiclient import client
32
from ceilometerclient.openstack.common.py3kcompat import urlutils
35
def assert_has_keys(dct, required=[], optional=[]):
39
except AssertionError:
40
extra_keys = set(dct.keys()).difference(set(required + optional))
41
raise AssertionError("found unexpected keys: %s" %
45
class TestResponse(requests.Response):
46
"""Wrap requests.Response and provide a convenient initialization.
49
def __init__(self, data):
50
super(TestResponse, self).__init__()
51
self._content_consumed = True
52
if isinstance(data, dict):
53
self.status_code = data.get('status_code', 200)
54
# Fake the text attribute to streamline Response creation
55
text = data.get('text', "")
56
if isinstance(text, (dict, list)):
57
self._content = json.dumps(text)
59
"Content-Type": "application/json",
64
self.headers = data.get('headers') or default_headers
66
self.status_code = data
68
def __eq__(self, other):
69
return (self.status_code == other.status_code and
70
self.headers == other.headers and
71
self._content == other._content)
74
class FakeHTTPClient(client.HTTPClient):
76
def __init__(self, *args, **kwargs):
78
self.fixtures = kwargs.pop("fixtures", None) or {}
79
if not args and not "auth_plugin" in kwargs:
81
super(FakeHTTPClient, self).__init__(*args, **kwargs)
83
def assert_called(self, method, url, body=None, pos=-1):
84
"""Assert than an API method was just called.
86
expected = (method, url)
87
called = self.callstack[pos][0:2]
88
assert self.callstack, \
89
"Expected %s %s but no calls were made." % expected
91
assert expected == called, 'Expected %s %s; got %s %s' % \
95
if self.callstack[pos][3] != body:
96
raise AssertionError('%r != %r' %
97
(self.callstack[pos][3], body))
99
def assert_called_anytime(self, method, url, body=None):
100
"""Assert than an API method was called anytime in the test.
102
expected = (method, url)
104
assert self.callstack, \
105
"Expected %s %s but no calls were made." % expected
109
for entry in self.callstack:
110
if expected == entry[0:2]:
114
assert found, 'Expected %s %s; got %s' % \
115
(method, url, self.callstack)
117
assert entry[3] == body, "%s != %s" % (entry[3], body)
121
def clear_callstack(self):
124
def authenticate(self):
127
def client_request(self, client, method, url, **kwargs):
128
# Check that certain things are called correctly
129
if method in ["GET", "DELETE"]:
130
assert "json" not in kwargs
133
self.callstack.append(
136
kwargs.get("headers") or {},
137
kwargs.get("json") or kwargs.get("data")))
139
fixture = self.fixtures[url][method]
143
return TestResponse({"headers": fixture[0],
147
args = urlutils.parse_qsl(urlutils.urlparse(url)[4])
149
munged_url = url.rsplit('?', 1)[0]
150
munged_url = munged_url.strip('/').replace('/', '_').replace('.', '_')
151
munged_url = munged_url.replace('-', '_')
153
callback = "%s_%s" % (method.lower(), munged_url)
155
if not hasattr(self, callback):
156
raise AssertionError('Called unknown API method: %s %s, '
157
'expected fakes method name: %s' %
158
(method, url, callback))
160
resp = getattr(self, callback)(**kwargs)
162
status, headers, body = resp
166
return TestResponse({
167
"status_code": status,
added
removed
ceilometerclient/openstack/common/apiclient/fake_client.py
