# or implied, of GRNET S.A.
from kamaki.clients import ClientError
-from kamaki.clients.compute.rest_api import ComputeClientApi
+from kamaki.clients.compute.rest_api import ComputeRestClient
from kamaki.clients.utils import path4url
-class ComputeClient(ComputeClientApi):
+class ComputeClient(ComputeRestClient):
"""OpenStack Compute API 1.1 client"""
def list_servers(self, detail=False):
import json
-class ComputeClientApi(Client):
+class ComputeRestClient(Client):
def servers_get(self, server_id='', command='', success=200, **kwargs):
"""GET base_url/servers[/server_id][/command] request
from itertools import product
from json import dumps
-from kamaki.clients.compute import ComputeClient, ComputeClientApi
+from kamaki.clients.compute import ComputeClient, ComputeRestClient
from kamaki.clients import ClientError
-rest_pkg = 'kamaki.clients.compute.rest_api.ComputeClientApi'
+rest_pkg = 'kamaki.clients.compute.rest_api.ComputeRestClient'
compute_pkg = 'kamaki.clients.compute.ComputeClient'
img_ref = "1m4g3-r3f3r3nc3"
pass
-class ComputeRestApi(TestCase):
+class ComputeRest(TestCase):
- """Set up a ComputesRestApi thorough test"""
+ """Set up a ComputesRest thorough test"""
def setUp(self):
self.url = 'http://cyclades.example.com'
self.token = 'cyc14d3s70k3n'
- self.client = ComputeClientApi(self.url, self.token)
+ self.client = ComputeRestClient(self.url, self.token)
def tearDown(self):
FR.json = vm_recv
if not argv[1:] or argv[1] == 'Compute':
not_found = False
runTestCase(Compute, 'Compute Client', argv[2:])
- if not argv[1:] or argv[1] == 'ComputeRestApi':
+ if not argv[1:] or argv[1] == 'ComputeRest':
not_found = False
- runTestCase(ComputeRestApi, 'ComputeRestApi Client', argv[2:])
+ runTestCase(ComputeRest, 'ComputeRest Client', argv[2:])
if not_found:
print('TestCase %s not found' % argv[1])
from sys import stdout
from time import sleep
-from kamaki.clients.cyclades.rest_api import CycladesClientApi
+from kamaki.clients.cyclades.rest_api import CycladesRestClient
from kamaki.clients import ClientError
-class CycladesClient(CycladesClientApi):
+class CycladesClient(CycladesRestClient):
"""GRNet Cyclades API client"""
def start_server(self, server_id):
--- /dev/null
+# Copyright 2012 GRNET S.A. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or
+# without modification, are permitted provided that the following
+# conditions are met:
+#
+# 1. Redistributions of source code must retain the above
+# copyright notice, this list of conditions and the following
+# disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following
+# disclaimer in the documentation and/or other materials
+# provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
+# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# The views and conclusions contained in the software and
+# documentation are those of the authors and should not be
+# interpreted as representing official policies, either expressed
+# or implied, of GRNET S.A.
+
+from kamaki.clients.compute import ComputeClient
+from kamaki.clients.utils import path4url
+import json
+
+
+class CycladesRestClient(ComputeClient):
+ """GRNet Cyclades REST API Client"""
+
+ def servers_get(
+ self,
+ server_id='',
+ command='',
+ success=200,
+ changes_since=None,
+ **kwargs):
+ """GET base_url/servers[/server_id][/command] request
+
+ :param server_id: integer (as int or str)
+
+ :param command: 'ips', 'stats', or ''
+
+ :param success: success code or list or tupple of accepted success
+ codes. if server response code is not in this list, a ClientError
+ raises
+
+ :param changes_since: (date)
+
+ :returns: request response
+ """
+ path = path4url('servers', server_id, command)
+ self.set_param('changes-since', changes_since, changes_since)
+ return self.get(path, success=success, **kwargs)
+
+ def networks_get(
+ self,
+ network_id='',
+ command='',
+ success=(200, 203),
+ **kwargs):
+ """GET base_url/networks[/network_id][/command] request
+
+ :param network_id: integer (str or int)
+
+ :param command: (str) 'detail' or ''
+
+ :param success: success code or list or tuple of accepted success
+ codes. if server response code is not in this list, a ClientError
+ raises
+
+ :returns: request response
+ """
+ path = path4url('networks', network_id, command)
+ return self.get(path, success=success, **kwargs)
+
+ def networks_delete(
+ self,
+ network_id='',
+ command='',
+ success=204,
+ **kwargs):
+ """DEL ETE base_url/networks[/network_id][/command] request
+
+ :param network_id: integer (str or int)
+
+ :param command: (str) 'detail' or ''
+
+ :param success: success code or list or tuple of accepted success
+ codes. if server response code is not in this list, a ClientError
+ raises
+
+ :returns: request response
+ """
+ path = path4url('networks', network_id, command)
+ return self.delete(path, success=success, **kwargs)
+
+ def networks_post(
+ self,
+ network_id='',
+ command='',
+ json_data=None,
+ success=202,
+ **kwargs):
+ """POST base_url/servers[/server_id]/[command] request
+
+ :param network_id: integer (str or int)
+
+ :param command: (str) 'detail' or ''
+
+ :param json_data: (dict) will be send as data
+
+ :param success: success code or list or tuple of accepted success
+ codes. if server response code is not in this list, a ClientError
+ raises
+
+ :returns: request response
+ """
+ data = json_data
+ if json_data is not None:
+ data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(data))
+
+ path = path4url('networks', network_id, command)
+ return self.post(path, data=data, success=success, **kwargs)
+
+ def networks_put(
+ self,
+ network_id='',
+ command='',
+ json_data=None,
+ success=204,
+ **kwargs):
+ """PUT base_url/servers[/server_id]/[command] request
+
+ :param network_id: integer (str or int)
+
+ :param command: (str) 'detail' or ''
+
+ :param json_data: (dict) will be send as data
+
+ :param success: success code or list or tuple of accepted success
+ codes. if server response code is not in this list, a ClientError
+ raises
+
+ :returns: request response
+ """
+ data = json_data
+ if json_data is not None:
+ data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(data))
+
+ path = path4url('networks', network_id, command)
+ return self.put(path, data=data, success=success, **kwargs)
from itertools import product
from kamaki.clients import ClientError
-from kamaki.clients.cyclades import CycladesClient, CycladesClientApi
+from kamaki.clients.cyclades import CycladesClient, CycladesRestClient
img_ref = "1m4g3-r3f3r3nc3"
vm_name = "my new VM"
def release(self):
pass
-rest_pkg = 'kamaki.clients.cyclades.CycladesClientApi'
+rest_pkg = 'kamaki.clients.cyclades.CycladesRestClient'
cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
-class CycladesRestApi(TestCase):
+class CycladesRest(TestCase):
"""Set up a Cyclades thorough test"""
def setUp(self):
self.url = 'http://cyclades.example.com'
self.token = 'cyc14d3s70k3n'
- self.client = CycladesClientApi(self.url, self.token)
+ self.client = CycladesRestClient(self.url, self.token)
def tearDown(self):
FR.json = vm_recv
if not argv[1:] or argv[1] == 'Cyclades':
not_found = False
runTestCase(Cyclades, 'Cyclades Client', argv[2:])
- if not argv[1:] or argv[1] == 'CycladesRestApi':
+ if not argv[1:] or argv[1] == 'CycladesRest':
not_found = False
- runTestCase(CycladesRestApi, 'CycladesRestApi Client', argv[2:])
+ runTestCase(CycladesRest, 'CycladesRest Client', argv[2:])
if not_found:
print('TestCase %s not found' % argv[1])
from binascii import hexlify
from kamaki.clients import SilentEvent, sendlog
-from kamaki.clients.pithos_rest_api import PithosRestAPI
+from kamaki.clients.pithos.rest_api import PithosRestClient
from kamaki.clients.storage import ClientError
from kamaki.clients.utils import path4url, filter_in
from StringIO import StringIO
return (start, end)
-class PithosClient(PithosRestAPI):
+class PithosClient(PithosRestClient):
"""GRNet Pithos API client"""
def __init__(self, base_url, token, account=None, container=None):
from kamaki.clients.utils import path4url
-class PithosRestAPI(StorageClient):
+class PithosRestClient(StorageClient):
def account_head(
self,
from os import urandom
from kamaki.clients import ClientError
-from kamaki.clients.pithos import PithosClient as PC
+from kamaki.clients.pithos import PithosClient, PithosRestClient
pithos_pkg = 'kamaki.clients.pithos.PithosClient'
pass
+class PithosRest(TestCase):
+
+ def setUp(self):
+ self.url = 'https://www.example.com/pithos'
+ self.token = 'p17h0570k3n'
+ self.client = PithosClient(self.url, self.token)
+ self.client.account = user_id
+ self.client.container = 'c0nt@1n3r_i'
+
+ def tearDown(self):
+ FR.headers = dict()
+ FR.status_code = 200
+ FR.json = dict()
+ FR.content = FR.json
+ for f in self.files:
+ f.close()
+
+
class Pithos(TestCase):
files = []
def setUp(self):
self.url = 'https://www.example.com/pithos'
self.token = 'p17h0570k3n'
- self.client = PC(self.url, self.token)
+ self.client = PithosClient(self.url, self.token)
self.client.account = user_id
self.client.container = 'c0nt@1n3r_i'
def test_get_object_info(self):
FR.headers = object_info
version = 'v3r510n'
- with patch.object(PC, 'object_head', return_value=FR()) as head:
+ with patch.object(
+ PithosClient, 'object_head',
+ return_value=FR()) as head:
r = self.client.get_object_info(obj)
self.assertEqual(r, object_info)
r = self.client.get_object_info(obj, version=version)
call(obj, version=None),
call(obj, version=version)])
with patch.object(
- PC, 'object_head',
+ PithosClient, 'object_head',
side_effect=ClientError('Obj not found', 404)):
self.assertRaises(
ClientError,
FR.json = object_hashmap
for empty in (304, 412):
with patch.object(
- PC, 'object_get',
+ PithosClient, 'object_get',
side_effect=ClientError('Empty', status=empty)):
r = self.client.get_object_hashmap(obj)
self.assertEqual(r, {})
if_modified_since='some date here',
if_unmodified_since='some date here',
data_range='10-20')
- with patch.object(PC, 'object_get', return_value=FR()) as get:
+ with patch.object(
+ PithosClient, 'object_get',
+ return_value=FR()) as get:
r = self.client.get_object_hashmap(obj)
self.assertEqual(r, object_hashmap)
self.assertEqual(get.mock_calls[-1], call(obj, **exp_args))
def test_get_account_meta(self):
key = 'x-account-meta-'
- with patch.object(PC, 'get_account_info', return_value=account_info):
+ with patch.object(
+ PithosClient, 'get_account_info',
+ return_value=account_info):
r = self.client.get_account_meta()
keys = [k for k in r if k.startswith(key)]
self.assertFalse(keys)
acc_info['%sk1' % key] = 'v1'
acc_info['%sk2' % key] = 'v2'
acc_info['%sk3' % key] = 'v3'
- with patch.object(PC, 'get_account_info', return_value=acc_info):
+ with patch.object(
+ PithosClient, 'get_account_info',
+ return_value=acc_info):
r = self.client.get_account_meta()
for k in [k for k in acc_info if k.startswith(key)]:
self.assertEqual(r[k], acc_info[k])
def test_get_account_group(self):
key = 'x-account-group-'
- with patch.object(PC, 'get_account_info', return_value=account_info):
+ with patch.object(
+ PithosClient, 'get_account_info',
+ return_value=account_info):
r = self.client.get_account_group()
keys = [k for k in r if k.startswith(key)]
self.assertFalse(keys)
acc_info['%sk1' % key] = 'g1'
acc_info['%sk2' % key] = 'g2'
acc_info['%sk3' % key] = 'g3'
- with patch.object(PC, 'get_account_info', return_value=acc_info):
+ with patch.object(
+ PithosClient, 'get_account_info',
+ return_value=acc_info):
r = self.client.get_account_group()
for k in [k for k in acc_info if k.startswith(key)]:
self.assertEqual(r[k], acc_info[k])
container_plus[key] = metaval
for ret in ((container_info, {}), (container_plus, {key: metaval})):
with patch.object(
- PC,
+ PithosClient,
'get_container_info',
return_value=ret[0]) as GCI:
for until in (None, somedate):
(container_info, {key: ''}),
(container_plus, {key: metaval})):
with patch.object(
- PC,
+ PithosClient,
'get_container_info',
return_value=ret[0]) as GCI:
for until in (None, somedate):
oinfo = dict(object_info)
val = 'pubL1c'
oinfo['x-object-public'] = val
- with patch.object(PC, 'get_object_info', return_value=oinfo) as GOF:
+ with patch.object(
+ PithosClient, 'get_object_info',
+ return_value=oinfo) as GOF:
r = self.client.publish_object(obj)
self.assertEqual(
post.mock_calls[-1],
expected = dict(read='u1,g1,u2', write='u1')
info['x-object-sharing'] = '; '.join(
['%s=%s' % (k, v) for k, v in expected.items()])
- with patch.object(PC, 'get_object_info', return_value=info) as GOF:
+ with patch.object(
+ PithosClient, 'get_object_info',
+ return_value=info) as GOF:
r = self.client.get_object_sharing(obj)
self.assertEqual(GOF.mock_calls[-1], call(obj))
self.assert_dicts_are_equal(r, expected)
info = dict(object_info)
info['content-length'] = file_size
block_size = container_info['x-container-block-size']
- with patch.object(PC, 'get_object_info', return_value=info) as GOI:
+ with patch.object(
+ PithosClient, 'get_object_info',
+ return_value=info) as GOI:
for start, end in (
(0, file_size + 1),
(file_size + 1, file_size + 2)):
if __name__ == '__main__':
from sys import argv
from kamaki.clients.test import runTestCase
- runTestCase(Pithos, 'Pithos+ Client', argv[1:])
+ not_found = True
+ if not argv[1:] or argv[1] == 'Pithos':
+ not_found = False
+ runTestCase(Pithos, 'Pithos Client', argv[2:])
+ if not argv[1:] or argv[1] == 'PithosRest':
+ not_found = False
+ runTestCase(PithosRest, 'PithosRest Client', argv[2:])
+ if not_found:
+ print('TestCase %s not found' % argv[1])
from inspect import getmembers, isclass
from kamaki.clients.astakos.test import Astakos
-from kamaki.clients.compute.test import Compute, ComputeRestApi
-from kamaki.clients.cyclades.test import Cyclades, CycladesRestApi
+from kamaki.clients.compute.test import Compute, ComputeRest
+from kamaki.clients.cyclades.test import Cyclades, CycladesRest
from kamaki.clients.image.test import Image
from kamaki.clients.storage.test import Storage
-from kamaki.clients.pithos.test import Pithos
+from kamaki.clients.pithos.test import Pithos, PithosRest
class SilentEvent(TestCase):
projects / kamaki / commitdiff