class Astakos(livetest.Generic):
def setUp(self):
+ self.remote = 'remote.%s' % self['testremote']
self.client = AstakosClient(
- self['remote.default', 'url'],
- self['remote.default', 'token'])
+ self[self.remote, 'url'], self[self.remote, 'token'])
with open(self['astakos', 'details']) as f:
self._astakos_details = eval(f.read())
self._test_0020_get_services()
def _test_0020_get_services(self):
- for args in (tuple(), (self['remote.default', 'token'],)):
+ for args in (tuple(), (self[self.remote, 'token'],)):
r = self.client.get_services(*args)
services = self._astakos_details['access']['serviceCatalog']
self.assertEqual(len(services), len(r))
parsed_services = dict()
for args in product(
self._astakos_details['access']['serviceCatalog'],
- ([tuple(), (self['remote.default', 'token'],)])):
+ ([tuple(), (self[self.remote, 'token'],)])):
service = args[0]
if service['type'] in parsed_services:
continue
parsed_services = dict()
for args in product(
self._astakos_details['access']['serviceCatalog'],
- ([], [self['remote.default', 'token']])):
+ ([], [self[self.remote, 'token']])):
service = args[0]
if service['type'] in parsed_services:
continue
def _test_0020_get(self):
for term in ('id', 'name', 'roles'):
self.assertEqual(
- self.client.term(term, self['remote.default', 'token']),
+ self.client.term(term, self[self.remote, 'token']),
self['astakos', term] or ([] if term == 'roles' else ''))
def test_list_users(self):
from kamaki.clients import livetest, ClientError
from kamaki.clients.cyclades import CycladesClient
+from kamaki.clients.astakos import AstakosClient
class Cyclades(livetest.Generic):
self.netname1 = 'net' + unicode(self.now)
self.netname2 = 'net' + unicode(self.now) + '_v2'
- self.client = CycladesClient(self['compute', 'url'], self['token'])
+ self.remote = 'remote.%s' % self['testremote']
+ aurl, self.token = self[self.remote, 'url'], self[self.remote, 'token']
+ self.auth_base = AstakosClient(aurl, self.token)
+ curl = self.auth_base.get_service_endpoints('compute')['publicURL']
+ self.client = CycladesClient(curl, self.token)
def tearDown(self):
"""Destoy servers used in testing"""
import time
from kamaki.clients import livetest
+from kamaki.clients.astakos import AstakosClient as AstakosCachedClient
from kamaki.clients.cyclades import CycladesClient
from kamaki.clients.image import ImageClient
from kamaki.clients import ClientError
class Image(livetest.Generic):
def setUp(self):
self.now = time.mktime(time.gmtime())
-
+ self.remote = 'remote.%s' % self['testremote']
+ aurl, self.token = self[self.remote, 'url'], self[self.remote, 'token']
+ self.auth_base = AstakosCachedClient(aurl, self.token)
self.imgname = 'img_%s' % self.now
- url = self['image', 'url']
- self.client = ImageClient(url, self['token'])
- cyclades_url = self['compute', 'url']
- self.cyclades = CycladesClient(cyclades_url, self['token'])
+ url = self.auth_base.get_service_endpoints('image')['publicURL']
+ self.token = self.auth_base.token
+ self.client = ImageClient(url, self.token)
+ cyclades_url = self.auth_base.get_service_endpoints(
+ 'compute')['publicURL']
+ self.cyclades = CycladesClient(cyclades_url, self.token)
self._imglist = {}
self._imgdetails = {}
def _prepare_img(self):
f = open(self['image', 'local_path'], 'rb')
- (token, uuid) = (self['token'], self['file', 'account'])
- if not uuid:
- from kamaki.clients.astakos import AstakosClient
- uuid = AstakosClient(self['user', 'url'], token).term('uuid')
+ (token, uuid) = (self.token, self.auth_base.user_term('id'))
+ purl = self.auth_base.get_service_endpoints(
+ 'object-store')['publicURL']
from kamaki.clients.pithos import PithosClient
- self.pithcli = PithosClient(self['file', 'url'], token, uuid)
+ self.pithcli = PithosClient(purl, token, uuid)
cont = 'cont_%s' % self.now
self.pithcli.container = cont
self.obj = 'obj_%s' % self.now
return img
return None
- def assert_dicts_are_deeply_equal(self, d1, d2):
- for k, v in d1.items():
- self.assertTrue(k in d2)
- if isinstance(v, dict):
- self.assert_dicts_are_deeply_equal(v, d2[k])
- else:
- self.assertEqual(unicode(v), unicode(d2[k]))
-
def test_list_public(self):
"""Test list_public"""
self._test_list_public()
self.assertTrue(r, r0)
r0.reverse()
for i, img in enumerate(r):
- self.assert_dicts_are_deeply_equal(img, r0[i])
+ self.assert_dicts_are_equal(img, r0[i])
r1 = self.client.list_public(detail=True)
for img in r1:
for term in (
files = []
def setUp(self):
- self.client = PithosClient(
- self['file', 'url'],
- self['file', 'token'],
- AstakosClient(
- self['user', 'url'],
- self['file', 'token']
- ).term('uuid'))
+ self.remote = 'remote.%s' % self['testremote']
+ aurl, self.token = self[self.remote, 'url'], self[self.remote, 'token']
+ self.auth_base = AstakosClient(aurl, self.token)
+ purl = self.auth_base.get_service_endpoints(
+ 'object-store')['publicURL']
+ self.uuid = self.auth_base.user_term('id')
+ self.client = PithosClient(purl, self.token, self.uuid)
self.now = time.mktime(time.gmtime())
self.now_unformated = datetime.datetime.utcnow()