# or implied, of GRNET S.A.
from mock import patch, call
from unittest import TestCase
+from itertools import product
from kamaki.clients import ClientError
-from kamaki.clients.cyclades import CycladesClient
+from kamaki.clients.cyclades import CycladesClient, CycladesClientApi
img_ref = "1m4g3-r3f3r3nc3"
vm_name = "my new VM"
def release(self):
pass
+rest_pkg = 'kamaki.clients.cyclades.CycladesClientApi'
cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
+class CycladesRestApi(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)
+
+ def tearDown(self):
+ FR.json = vm_recv
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.get' % rest_pkg, return_value=FR())
+ def _test_get(self, service, get, SP):
+ for args in product(
+ ('', '%s_id' % service),
+ ('', 'cmd'),
+ (200, 204),
+ (None, '50m3-d473'),
+ ({}, {'k': 'v'})):
+ (srv_id, command, success, changes_since, kwargs) = args
+ method = getattr(self.client, '%s_get' % service)
+ method(*args[:4], **kwargs)
+ srv_str = '/%s' % srv_id if srv_id else ''
+ cmd_str = '/%s' % command if command else ''
+ self.assertEqual(get.mock_calls[-1], call(
+ '/%s%s%s' % (service, srv_str, cmd_str),
+ success=success,
+ **kwargs))
+ if changes_since:
+ self.assertEqual(
+ SP.mock_calls[-1],
+ call('changes-since', changes_since, changes_since))
+
+ def test_servers_get(self):
+ self._test_get('servers')
+
+
class Cyclades(TestCase):
def assert_dicts_are_equal(self, d1, d2):
self.assertEqual(err.details, [
'Network may be still connected to at least one server'])
+
if __name__ == '__main__':
from sys import argv
from kamaki.clients.test import runTestCase
- runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])
+ not_found = True
+ 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':
+ not_found = False
+ runTestCase(CycladesRestApi, 'CycladesRestApi Client', argv[2:])
+ if not_found:
+ print('TestCase %s not found' % argv[1])
+++ /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 CycladesClientApi(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)