Revision f90a1ab5 test/ganeti.rapi.client_unittest.py
b/test/ganeti.rapi.client_unittest.py | ||
---|---|---|
22 | 22 |
"""Script for unittesting the RAPI client module""" |
23 | 23 |
|
24 | 24 |
|
25 |
import re |
|
26 | 25 |
import unittest |
27 | 26 |
import warnings |
28 | 27 |
import pycurl |
... | ... | |
33 | 32 |
from ganeti import utils |
34 | 33 |
from ganeti import query |
35 | 34 |
from ganeti import objects |
35 |
from ganeti import rapi |
|
36 | 36 |
|
37 |
import ganeti.rapi.testutils |
|
37 | 38 |
from ganeti.rapi import connector |
38 | 39 |
from ganeti.rapi import rlib2 |
39 | 40 |
from ganeti.rapi import client |
... | ... | |
41 | 42 |
import testutils |
42 | 43 |
|
43 | 44 |
|
44 |
_URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)") |
|
45 |
|
|
46 | 45 |
# List of resource handlers which aren't used by the RAPI client |
47 | 46 |
_KNOWN_UNUSED = set([ |
48 | 47 |
rlib2.R_root, |
... | ... | |
53 | 52 |
_used_handlers = None |
54 | 53 |
|
55 | 54 |
|
56 |
def _GetPathFromUri(uri): |
|
57 |
"""Gets the path and query from a URI. |
|
58 |
|
|
59 |
""" |
|
60 |
match = _URI_RE.match(uri) |
|
61 |
if match: |
|
62 |
return match.groupdict()["path"] |
|
63 |
else: |
|
64 |
return None |
|
65 |
|
|
66 |
|
|
67 |
class FakeCurl: |
|
68 |
def __init__(self, rapi): |
|
69 |
self._rapi = rapi |
|
70 |
self._opts = {} |
|
71 |
self._info = {} |
|
72 |
|
|
73 |
def setopt(self, opt, value): |
|
74 |
self._opts[opt] = value |
|
75 |
|
|
76 |
def getopt(self, opt): |
|
77 |
return self._opts.get(opt) |
|
78 |
|
|
79 |
def unsetopt(self, opt): |
|
80 |
self._opts.pop(opt, None) |
|
81 |
|
|
82 |
def getinfo(self, info): |
|
83 |
return self._info[info] |
|
84 |
|
|
85 |
def perform(self): |
|
86 |
method = self._opts[pycurl.CUSTOMREQUEST] |
|
87 |
url = self._opts[pycurl.URL] |
|
88 |
request_body = self._opts[pycurl.POSTFIELDS] |
|
89 |
writefn = self._opts[pycurl.WRITEFUNCTION] |
|
90 |
|
|
91 |
path = _GetPathFromUri(url) |
|
92 |
(code, resp_body) = self._rapi.FetchResponse(path, method, request_body) |
|
93 |
|
|
94 |
self._info[pycurl.RESPONSE_CODE] = code |
|
95 |
if resp_body is not None: |
|
96 |
writefn(resp_body) |
|
97 |
|
|
98 |
|
|
99 | 55 |
class RapiMock(object): |
100 | 56 |
def __init__(self): |
101 | 57 |
self._mapper = connector.Mapper() |
... | ... | |
216 | 172 |
class TestExtendedConfig(unittest.TestCase): |
217 | 173 |
def testAuth(self): |
218 | 174 |
cl = client.GanetiRapiClient("master.example.com", |
219 |
username="user", password="pw",
|
|
220 |
curl_factory=lambda: FakeCurl(RapiMock()))
|
|
175 |
username="user", password="pw", |
|
176 |
curl_factory=lambda: rapi.testutils.FakeCurl(RapiMock()))
|
|
221 | 177 |
|
222 | 178 |
curl = cl._CreateCurl() |
223 | 179 |
self.assertEqual(curl.getopt(pycurl.HTTPAUTH), pycurl.HTTPAUTH_BASIC) |
... | ... | |
254 | 210 |
verify_hostname=verify_hostname, |
255 | 211 |
_pycurl_version_fn=pcverfn) |
256 | 212 |
|
257 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
213 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
258 | 214 |
cl = client.GanetiRapiClient("master.example.com", |
259 | 215 |
curl_config_fn=cfgfn, |
260 | 216 |
curl_factory=curl_factory) |
... | ... | |
271 | 227 |
def testNoCertVerify(self): |
272 | 228 |
cfgfn = client.GenericCurlConfig() |
273 | 229 |
|
274 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
230 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
275 | 231 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
276 | 232 |
curl_factory=curl_factory) |
277 | 233 |
|
... | ... | |
283 | 239 |
def testCertVerifyCurlBundle(self): |
284 | 240 |
cfgfn = client.GenericCurlConfig(use_curl_cabundle=True) |
285 | 241 |
|
286 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
242 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
287 | 243 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
288 | 244 |
curl_factory=curl_factory) |
289 | 245 |
|
... | ... | |
296 | 252 |
mycert = "/tmp/some/UNUSED/cert/file.pem" |
297 | 253 |
cfgfn = client.GenericCurlConfig(cafile=mycert) |
298 | 254 |
|
299 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
255 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
300 | 256 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
301 | 257 |
curl_factory=curl_factory) |
302 | 258 |
|
... | ... | |
311 | 267 |
cfgfn = client.GenericCurlConfig(capath=certdir, |
312 | 268 |
_pycurl_version_fn=pcverfn) |
313 | 269 |
|
314 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
270 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
315 | 271 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
316 | 272 |
curl_factory=curl_factory) |
317 | 273 |
|
... | ... | |
326 | 282 |
cfgfn = client.GenericCurlConfig(capath=certdir, |
327 | 283 |
_pycurl_version_fn=pcverfn) |
328 | 284 |
|
329 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
285 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
330 | 286 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
331 | 287 |
curl_factory=curl_factory) |
332 | 288 |
|
... | ... | |
338 | 294 |
cfgfn = client.GenericCurlConfig(capath=certdir, |
339 | 295 |
_pycurl_version_fn=pcverfn) |
340 | 296 |
|
341 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
297 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
342 | 298 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
343 | 299 |
curl_factory=curl_factory) |
344 | 300 |
|
... | ... | |
350 | 306 |
cfgfn = client.GenericCurlConfig(capath=certdir, |
351 | 307 |
_pycurl_version_fn=pcverfn) |
352 | 308 |
|
353 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
309 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
354 | 310 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
355 | 311 |
curl_factory=curl_factory) |
356 | 312 |
|
... | ... | |
362 | 318 |
cfgfn = client.GenericCurlConfig(connect_timeout=connect_timeout, |
363 | 319 |
timeout=timeout) |
364 | 320 |
|
365 |
curl_factory = lambda: FakeCurl(RapiMock()) |
|
321 |
curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
|
|
366 | 322 |
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn, |
367 | 323 |
curl_factory=curl_factory) |
368 | 324 |
|
... | ... | |
376 | 332 |
testutils.GanetiTestCase.setUp(self) |
377 | 333 |
|
378 | 334 |
self.rapi = RapiMock() |
379 |
self.curl = FakeCurl(self.rapi) |
|
335 |
self.curl = rapi.testutils.FakeCurl(self.rapi)
|
|
380 | 336 |
self.client = client.GanetiRapiClient("master.example.com", |
381 | 337 |
curl_factory=lambda: self.curl) |
382 | 338 |
|
Also available in: Unified diff