def test_create_server_metadata(self, SP):
vm_id = vm_recv['server']['id']
metadata = dict(m1='v1', m2='v2', m3='v3')
- FR.json = dict(metadata=vm_recv['server'])
+ FR.json = dict(meta=vm_recv['server'])
for k, v in metadata.items():
r = self.client.create_server_metadata(vm_id, k, v)
self.assert_dicts_are_equal(r, vm_recv['server'])
self.assertEqual(SP.mock_calls[-1], call(
vm_id, 'metadata/%s' % k,
- json_data=dict(metadata={k: v}), success=201))
+ json_data=dict(meta={k: v}), success=201))
@patch('%s.servers_get' % compute_pkg, return_value=FR())
def test_get_server_metadata(self, SG):
metadata = dict(m1='v1', m2='v2', m3='v3')
FR.json = dict(metadata=metadata)
r = self.client.get_server_metadata(vm_id)
+ FR.json = dict(meta=metadata)
SG.assert_called_once_with(vm_id, '/metadata')
self.assert_dicts_are_equal(r, metadata)
for k, v in metadata.items():
- FR.json = dict(metadata={k: v})
+ FR.json = dict(meta={k: v})
r = self.client.get_server_metadata(vm_id, k)
self.assert_dicts_are_equal(r, {k: v})
self.assertEqual(
@patch('%s.images_get' % compute_pkg, return_value=FR())
def test_get_image_metadata(self, IG):
for key in ('', '50m3k3y'):
- FR.json = dict(metadata=img_recv['image']) if (
+ FR.json = dict(meta=img_recv['image']) if (
key) else dict(metadata=img_recv['image'])
r = self.client.get_image_metadata(img_ref, key)
self.assertEqual(IG.mock_calls[-1], call(
@patch('%s.images_put' % compute_pkg, return_value=FR())
def test_create_image_metadata(self, IP):
(key, val) = ('k1', 'v1')
- FR.json = dict(metadata=img_recv['image'])
+ FR.json = dict(meta=img_recv['image'])
r = self.client.create_image_metadata(img_ref, key, val)
IP.assert_called_once_with(
img_ref, 'metadata/%s' % key,
- json_data=dict(metadata={key: val}))
+ json_data=dict(meta={key: val}))
self.assert_dicts_are_equal(r, img_recv['image'])
@patch('%s.images_post' % compute_pkg, return_value=FR())
@patch('%s.servers_get' % cyclades_pkg, return_value=FR())
def test_list_server_nics(self, SG):
vm_id = vm_recv['server']['id']
- nics = dict(addresses=[dict(id='nic1'), dict(id='nic2')])
+ nics = dict(attachments=[dict(id='nic1'), dict(id='nic2')])
FR.json = nics
r = self.client.list_server_nics(vm_id)
SG.assert_called_once_with(vm_id, 'ips')
- expected = nics['addresses']
+ expected = nics['attachments']
for i in range(len(r)):
self.assert_dicts_are_equal(r[i], expected[i])
self.assertEqual(i + 1, len(r))
self.assertEqual(v, d2[k])
def assert_dicts_are_deeply_equal(self, d1, d2):
- (st1, st2) = (set(d1.keys()), set(d2.keys()))
+ st1 = set([unicode(k) for k in d1])
+ st2 = set([unicode(k) for k in d2])
diff1 = st1.difference(st2)
diff2 = st2.difference(st1)
self.assertTrue(
self._test_0020_get()
def _test_0020_get(self):
- for term in ('id', 'name', 'roles'):
+ for term in ('id', 'name'):
self.assertEqual(
self.client.term(term, self[self.cloud, 'token']),
- self['astakos', term] or ([] if term == 'roles' else ''))
+ self['astakos', term] or '')
def test_list_users(self):
self.client.authenticate()
def _create_server(self, servername, flavorid, imageid, personality=None):
server = self.client.create_server(
- servername,
- flavorid,
- imageid,
- personality)
+ servername, flavorid, imageid, personality)
print('CREATE VM %s (%s)' % (server['id'], server['name']))
self.servers[servername] = server
return server
def _test_0010_create_server(self):
self.assertEqual(self.server1["name"], self.servname1)
- self.assertEqual(self.server1["flavor"], self.flavorid)
- self.assertEqual(self.server1["image"], self.img)
+ self.assertEqual(self.server1["flavor"]["id"], self.flavorid)
+ self.assertEqual(self.server1["image"]["id"], self.img)
self.assertEqual(self.server1["status"], "BUILD")
def test_list_servers(self):
def _test_0040_get_server_details(self):
r = self.client.get_server_details(self.server1['id'])
self.assertEqual(r["name"], self.servname1)
- self.assertEqual(r["flavor"], self.flavorid)
- self.assertEqual(r["image"], self.img)
+ self.assertEqual(r["flavor"]["id"], self.flavorid)
+ self.assertEqual(r["image"]["id"], self.img)
self.assertEqual(r["status"], "ACTIVE")
def test_update_server_name(self):
'mymeta val')
self.assertTrue('mymeta' in r1)
r2 = self.client.get_server_metadata(self.server1['id'], 'mymeta')
- self.assert_dicts_are_deeply_equal(r1, r2)
+ self.assert_dicts_are_equal(r1, r2)
def test_get_server_metadata(self):
"""Test get server_metadata"""
def _test_0130_get_flavor_details(self):
r = self.client.get_flavor_details(self.flavorid)
- self.assert_dicts_are_deeply_equal(self._flavor_details, r)
+ self.assert_dicts_are_equal(self._flavor_details, r)
- def test_list_images(self):
- """Test list_images"""
- self._test_0140_list_images()
+ #def test_list_images(self):
+ # """Test list_images"""
+ # self._test_0140_list_images()
- def _test_0140_list_images(self):
- r = self.client.list_images()
- self.assertTrue(len(r) > 1)
- r = self.client.list_images(detail=True)
- for detailed_img in r:
- if detailed_img['id'] == self.img:
- break
- self.assert_dicts_are_deeply_equal(detailed_img, self.img_details)
+ #def _test_0140_list_images(self):
+ # r = self.client.list_images()
+ # self.assertTrue(len(r) > 1)
+ # r = self.client.list_images(detail=True)
+ # for detailed_img in r:
+ # if detailed_img['id'] == self.img:
+ # break
+ # self.assert_dicts_are_equal(detailed_img, self.img_details)
def test_get_image_details(self):
"""Test image_details"""
def _test_0150_get_image_details(self):
r = self.client.get_image_details(self.img)
- self.assert_dicts_are_deeply_equal(r, self.img_details)
+ self.assert_dicts_are_equal(r, self.img_details)
def test_get_image_metadata(self):
"""Test get_image_metadata"""
def _test_0160_get_image_metadata(self):
r = self.client.get_image_metadata(self.img)
- self.assert_dicts_are_deeply_equal(
+ self.assert_dicts_are_equal(
self.img_details['metadata'], r)
for key, val in self.img_details['metadata'].items():
r = self.client.get_image_metadata(self.img, key)
chosen.pop('updated')
net1 = dict(self.network1)
net1.pop('updated')
- self.assert_dicts_are_deeply_equal(chosen, net1)
+ self.assert_dicts_are_equal(chosen, net1)
full_args = dict(
cidr='192.168.1.0/24',
gateway='192.168.1.1',
r.pop('status')
r.pop('updated', None)
r.pop('attachments')
- self.assert_dicts_are_deeply_equal(net1, r)
+ self.assert_dicts_are_equal(net1, r)
def test_update_network_name(self):
self.network2 = self._create_network(self.netname2)
GAI.assert_called_once_with()
self.assertEqual(r[key], account_info[key])
- @patch('%s.get_account_info' % pithos_pkg, return_value=account_info)
- def test_get_account_versioning(self, GAI):
- key = 'x-account-policy-versioning'
- r = self.client.get_account_versioning()
- GAI.assert_called_once_with()
- self.assertEqual(r[key], account_info[key])
+ #@patch('%s.get_account_info' % pithos_pkg, return_value=account_info)
+ #def test_get_account_versioning(self, GAI):
+ # key = 'x-account-policy-versioning'
+ # r = self.client.get_account_versioning()
+ # GAI.assert_called_once_with()
+ # self.assertEqual(r[key], account_info[key])
def test_get_account_meta(self):
key = 'x-account-meta-'
self.client.set_account_meta(metas)
post.assert_called_once_with(update=True, metadata=metas)
- """
- @patch('%s.account_post' % pithos_pkg, return_value=FR())
- def test_set_account_quota(self, post):
- qu = 1024
- self.client.set_account_quota(qu)
- post.assert_called_once_with(update=True, quota=qu)
- """
+ #@patch('%s.account_post' % pithos_pkg, return_value=FR())
+ #def test_set_account_quota(self, post):
+ # qu = 1024
+ # self.client.set_account_quota(qu)
+ # post.assert_called_once_with(update=True, quota=qu)
- @patch('%s.account_post' % pithos_pkg, return_value=FR())
- def test_set_account_versioning(self, post):
- vrs = 'n3wV3r51on1ngTyp3'
- self.client.set_account_versioning(vrs)
- post.assert_called_once_with(update=True, versioning=vrs)
+ #@patch('%s.account_post' % pithos_pkg, return_value=FR())
+ #def test_set_account_versioning(self, post):
+ # vrs = 'n3wV3r51on1ngTyp3'
+ # self.client.set_account_versioning(vrs)
+ # post.assert_called_once_with(update=True, versioning=vrs)
@patch('%s.container_delete' % pithos_pkg, return_value=FR())
def test_del_container(self, delete):
def test_publish_object(self, post):
oinfo = dict(object_info)
val = 'pubL1c'
- oinfo['x-object-public'] = val
+ oinfo['x-object-public'] = 'https://www.example.com/' + val
with patch.object(
pithos.PithosClient, 'get_object_info',
return_value=oinfo) as GOF:
HEADERS = dict(k='v', k1='v1', k2='v2')
reason = 'some reason'
status = 42
+ status_code = 200
def read(self):
return self.READ
status_code = 200
-class FakeConnection(object):
- """A fake Connection class"""
-
- headers = dict()
- params = dict()
-
- def __init__(self):
- pass
-
- def set_header(self, name, value):
- pass
-
- def reset_headers(self):
- self.headers = {}
-
- def set_param(self, name, value):
- self.params = {}
-
- def reset_params(self):
- pass
-
- def perform_request(self, *args):
- return FR()
-
-
class Client(TestCase):
def assert_dicts_are_equal(self, d1, d2):
from kamaki.clients import ClientError as CE
self.base_url = 'http://example.com'
self.token = 's0m370k3n=='
- self.client = Client(self.base_url, self.token, FakeConnection())
+ self.client = Client(self.base_url, self.token)
self.CE = CE
def tearDown(self):
FR.text = None
FR.status = None
FR.status_code = 200
- FakeConnection.headers = dict()
self.client.token = self.token
def test___init__(self):
'%A, %d-%b-%y %H:%M:%S GMT',
'%a, %d %b %Y %H:%M:%S GMT']
self.assertEqual(self.client.DATE_FORMATS, DATE_FORMATS)
- self.assertTrue(isinstance(self.client.http_client, FakeConnection))
def test__init_thread_limit(self):
exp = 'Nothing set here'
self.assertEqual('%s' % ce, '%s %s\n' % (sts_code or '', msg))
self.assertEqual(ce.status, sts_code or 0)
- @patch('%s.FakeConnection.set_header' % __name__)
+ @patch('kamaki.clients.Client.set_header')
def test_set_header(self, SH):
- num_of_calls = 0
for name, value, condition in product(
('n4m3', '', None),
('v41u3', None, 42),
(True, False, None, 1, '')):
self.client.set_header(name, value, iff=condition)
- if value is not None and condition:
- self.assertEqual(SH.mock_calls[-1], call(name, value))
- num_of_calls += 1
- else:
- self.assertEqual(num_of_calls, len(SH.mock_calls))
+ self.assertEqual(
+ SH.mock_calls[-1], call(name, value, iff=condition))
- @patch('%s.FakeConnection.set_param' % __name__)
+ @patch('kamaki.clients.Client.set_param')
def test_set_param(self, SP):
- num_of_calls = 0
for name, value, condition in product(
('n4m3', '', None),
('v41u3', None, 42),
(True, False, None, 1, '')):
self.client.set_param(name, value, iff=condition)
- if condition:
- self.assertEqual(SP.mock_calls[-1], call(name, value))
- num_of_calls += 1
- else:
- self.assertEqual(num_of_calls, len(SP.mock_calls))
+ self.assertEqual(
+ SP.mock_calls[-1], call(name, value, iff=condition))
- @patch('%s.FakeConnection.perform_request' % __name__, return_value=FR())
- def test_request(self, PR):
+ @patch('kamaki.clients.RequestManager', return_value=FR)
+ @patch('kamaki.clients.ResponseManager', return_value=FakeResp())
+ @patch('kamaki.clients.ResponseManager.__init__')
+ def test_request(self, Requ, RespInit, Resp):
for args in product(
('get', '', dict(method='get')),
('/some/path', None, ['some', 'path']),
success=400,
json=dict(k2='v2', k1='v1')))):
method, path, kwargs = args[0], args[1], args[-1]
- args = args[:-1]
- if not (isinstance(method, str) and method and isinstance(
- path, str) and path):
+ FakeResp.status_code = kwargs.get('success', 200)
+ if not (method and (
+ isinstance(method, str) or isinstance(
+ method, unicode)) and (
+ isinstance(path, str) or isinstance(path, unicode))):
self.assertRaises(
- AssertionError,
- self.client.request,
- *args, **kwargs)
- else:
- atoken = 'a70k3n_%s' % randint(1, 30)
- self.client.token = atoken
- if 'success' in kwargs:
- self.assertRaises(
- self.CE,
- self.client.request,
- *args, **kwargs)
- FR.status_code = kwargs['success']
- else:
- FR.status_code = 200
- self.client.request(*args, **kwargs)
- data = kwargs.get(
- 'data',
- '{"k2": "v2", "k1": "v1"}' if 'json' in kwargs else None)
- self.assertEqual(self.client.http_client.url, self.base_url)
- self.assertEqual(self.client.http_client.path, path)
- self.assertEqual(
- PR.mock_calls[-1],
- call(method, data, *args[2:]))
- self.assertEqual(self.client.http_client.headers, dict())
- self.assertEqual(self.client.http_client.params, dict())
+ AssertionError, self.client.request, method, path,
+ **kwargs)
+ continue
+ self.client.request(method, path, **kwargs)
+ self.assertEqual(
+ RespInit.mock_calls[-1], call(FR, connection_retry_limit=0))
@patch('kamaki.clients.Client.request', return_value='lala')
def _test_foo(self, foo, request):
projects / kamaki / commitdiff