code.grnet.gr Git - kamaki/blob - kamaki/clients/cyclades/test.py

   1 # Copyright 2013 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, this list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, this list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33 from mock import patch, call
  34 from unittest import TestCase
  35 from json import loads
  36
  37 from kamaki.clients import ClientError
  38 from kamaki.clients.cyclades import CycladesClient
  39 from kamaki.clients.cyclades_rest_api import CycladesClientApi
  40
  41 img_ref = "1m4g3-r3f3r3nc3"
  42 vm_name = "my new VM"
  43 fid = 42
  44 vm_recv = dict(server=dict(
  45     status="BUILD",
  46     updated="2013-03-01T10:04:00.637152+00:00",
  47     hostId="",
  48     name=vm_name,
  49     imageRef=img_ref,
  50     created="2013-03-01T10:04:00.087324+00:00",
  51     flavorRef=fid,
  52     adminPass="n0n3sh@11p@55",
  53     suspended=False,
  54     progress=0,
  55     id=31173,
  56     metadata=dict(values=dict(os="debian", users="root"))))
  57 vm_list = dict(servers=dict(values=[
  58     dict(name='n1', id=1),
  59     dict(name='n2', id=2)]))
  60 net_send = dict(network=dict(dhcp=False, name='someNet'))
  61 net_recv = dict(network=dict(
  62     status="PENDING",
  63     updated="2013-03-05T15:04:51.758780+00:00",
  64     name="someNet",
  65     created="2013-03-05T15:04:51.758728+00:00",
  66     cidr6=None,
  67     id="2130",
  68     gateway6=None,
  69     public=False,
  70     dhcp=False,
  71     cidr="192.168.1.0/24",
  72     type="MAC_FILTERED",
  73     gateway=None,
  74     attachments=dict(values=[dict(name='att1'), dict(name='att2')])))
  75 net_list = dict(networks=dict(values=[
  76     dict(id=1, name='n1'),
  77     dict(id=2, name='n2'),
  78     dict(id=3, name='n3')]))
  79
  80
  81 class FR(object):
  82     """FR stands for Fake Response"""
  83     json = vm_recv
  84     headers = {}
  85     content = json
  86     status = None
  87     status_code = 200
  88
  89     def release(self):
  90         pass
  91
  92 khttp = 'kamaki.clients.connection.kamakicon.KamakiHTTPConnection'
  93 cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
  94
  95
  96 class Cyclades(TestCase):
  97
  98     def assert_dicts_are_equal(self, d1, d2):
  99         for k, v in d1.items():
 100             self.assertTrue(k in d2)
 101             if isinstance(v, dict):
 102                 self.assert_dicts_are_equal(v, d2[k])
 103             else:
 104                 self.assertEqual(unicode(v), unicode(d2[k]))
 105
 106     """Set up a Cyclades thorough test"""
 107     def setUp(self):
 108         self.url = 'http://cyclades.example.com'
 109         self.token = 'cyc14d3s70k3n'
 110         self.client = CycladesClient(self.url, self.token)
 111         from kamaki.clients.connection.kamakicon import KamakiHTTPConnection
 112         self.C = KamakiHTTPConnection
 113
 114     def tearDown(self):
 115         FR.status_code = 200
 116         FR.json = vm_recv
 117
 118     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 119     def test_list_servers(self, SG):
 120         FR.json = vm_list
 121         for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
 122             r = self.client.list_servers(detail=detail, changes_since=since)
 123             self.assertEqual(SG.mock_calls[-1], call(
 124                 command='detail' if detail else '',
 125                 changes_since=since))
 126             expected = vm_list['servers']['values']
 127             for i, vm in enumerate(r):
 128                 self.assert_dicts_are_equal(vm, expected[i])
 129             self.assertEqual(i + 1, len(expected))
 130
 131     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 132     def test_shutdown_server(self, SP):
 133         vm_id = vm_recv['server']['id']
 134         self.client.shutdown_server(vm_id)
 135         self.assertEqual(SP.mock_calls[-1], call(
 136             vm_id, 'action',
 137             json_data=dict(shutdown=dict()), success=202))
 138
 139     @patch('%s.perform_request' % khttp, return_value=FR())
 140     def test_start_server(self, PR):
 141         vm_id = vm_recv['server']['id']
 142         FR.status_code = 202
 143         self.client.start_server(vm_id)
 144         self.assertEqual(self.client.http_client.url, self.url)
 145         self.assertEqual(
 146             self.client.http_client.path,
 147             '/servers/%s/action' % vm_id)
 148         self.assertEqual(PR.call_args[0], ('post',  '{"start": {}}', {}, {}))
 149
 150     @patch('%s.perform_request' % khttp, return_value=FR())
 151     def test_get_server_console(self, PR):
 152         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 153         FR.json = cnsl
 154         vm_id = vm_recv['server']['id']
 155         r = self.client.get_server_console(vm_id)
 156         self.assertEqual(self.client.http_client.url, self.url)
 157         self.assertEqual(
 158             self.client.http_client.path,
 159             '/servers/%s/action' % vm_id)
 160         self.assert_dicts_are_equal(cnsl['console'], r)
 161         self.assertEqual(
 162             PR.call_args[0],
 163             ('post',  '{"console": {"type": "vnc"}}', {}, {}))
 164
 165     def test_get_firewall_profile(self):
 166         vm_id = vm_recv['server']['id']
 167         v = 'Some profile'
 168         ret = {'attachments': {'values': [{'firewallProfile': v, 1:1}]}}
 169         with patch.object(
 170                 CycladesClient,
 171                 'get_server_details',
 172                 return_value=ret) as GSD:
 173             r = self.client.get_firewall_profile(vm_id)
 174             self.assertEqual(r, v)
 175             self.assertEqual(GSD.call_args[0], (vm_id,))
 176             ret['attachments']['values'][0].pop('firewallProfile')
 177             self.assertRaises(
 178                 ClientError,
 179                 self.client.get_firewall_profile,
 180                 vm_id)
 181
 182     @patch('%s.perform_request' % khttp, return_value=FR())
 183     def test_set_firewall_profile(self, PR):
 184         vm_id = vm_recv['server']['id']
 185         v = 'Some profile'
 186         FR.status_code = 202
 187         self.client.set_firewall_profile(vm_id, v)
 188         self.assertEqual(self.client.http_client.url, self.url)
 189         self.assertEqual(
 190             self.client.http_client.path,
 191             '/servers/%s/action' % vm_id)
 192         self.assertEqual(PR.call_args[0], (
 193             'post',
 194             '{"firewallProfile": {"profile": "%s"}}' % v,
 195             {},
 196             {}))
 197
 198     @patch('%s.perform_request' % khttp, return_value=FR())
 199     def test_get_server_stats(self, PR):
 200         vm_id = vm_recv['server']['id']
 201         stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
 202         FR.json = dict(stats=stats)
 203         r = self.client.get_server_stats(vm_id)
 204         self.assertEqual(self.client.http_client.url, self.url)
 205         self.assertEqual(
 206             self.client.http_client.path,
 207             '/servers/%s/stats' % vm_id)
 208         self.assert_dicts_are_equal(stats, r)
 209
 210     @patch('%s.perform_request' % khttp, return_value=FR())
 211     def test_create_network(self, PR):
 212         net_name = net_send['network']['name']
 213         FR.json = net_recv
 214         FR.status_code = 202
 215         full_args = dict(
 216                 cidr='192.168.0.0/24',
 217                 gateway='192.168.0.1',
 218                 type='MAC_FILTERED',
 219                 dhcp=True)
 220         test_args = dict(full_args)
 221         test_args.update(dict(empty=None, full=None))
 222         for arg, val in test_args.items():
 223             kwargs = {} if arg == 'empty' else full_args if (
 224                 arg == 'full') else {arg: val}
 225             r = self.client.create_network(net_name, **kwargs)
 226             self.assertEqual(self.client.http_client.url, self.url)
 227             self.assertEqual(
 228                 self.client.http_client.path,
 229                 '/networks')
 230             self.assert_dicts_are_equal(r, net_recv['network'])
 231             data = PR.call_args[0][1]
 232             expected = dict(network=dict(net_send['network']))
 233             expected['network'].update(kwargs)
 234             self.assert_dicts_are_equal(loads(data), expected)
 235
 236     @patch('%s.perform_request' % khttp, return_value=FR())
 237     def test_connect_server(self, PR):
 238         vm_id = vm_recv['server']['id']
 239         net_id = net_recv['network']['id']
 240         FR.status_code = 202
 241         self.client.connect_server(vm_id, net_id)
 242         self.assertEqual(self.client.http_client.url, self.url)
 243         self.assertEqual(
 244             self.client.http_client.path,
 245             '/networks/%s/action' % net_id)
 246         self.assertEqual(
 247             PR.call_args[0],
 248             ('post', '{"add": {"serverRef": %s}}' % vm_id, {}, {}))
 249
 250     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 251     def test_disconnect_server(self, NP):
 252         vm_id = vm_recv['server']['id']
 253         net_id = net_recv['network']['id']
 254         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 255         vm_nics = [
 256             dict(id=nic_id, network_id=net_id),
 257             dict(id='another-nic-id', network_id='another-net-id'),
 258             dict(id=nic_id * 2, network_id=net_id * 2)]
 259         with patch.object(
 260                 CycladesClient,
 261                 'list_server_nics',
 262                 return_value=vm_nics) as LSN:
 263             r = self.client.disconnect_server(vm_id, nic_id)
 264             self.assertEqual(r, 1)
 265             self.assertEqual(LSN.call_args[0], (vm_id,))
 266             self.assertEqual(NP.call_args[0], (net_id, 'action'))
 267             self.assertEqual(
 268                 NP.call_args[1],
 269                 dict(json_data=dict(remove=dict(attachment=nic_id))))
 270
 271     @patch('%s.perform_request' % khttp, return_value=FR())
 272     def test_list_server_nics(self, PR):
 273         vm_id = vm_recv['server']['id']
 274         nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
 275         FR.json = nics
 276         r = self.client.list_server_nics(vm_id)
 277         self.assertEqual(self.client.http_client.url, self.url)
 278         self.assertEqual(
 279             self.client.http_client.path,
 280             '/servers/%s/ips' % vm_id)
 281         expected = nics['addresses']['values']
 282         for i in range(len(r)):
 283             self.assert_dicts_are_equal(r[i], expected[i])
 284
 285     @patch('%s.perform_request' % khttp, return_value=FR())
 286     def test_list_networks(self, PR):
 287         FR.json = net_list
 288         r = self.client.list_networks()
 289         self.assertEqual(self.client.http_client.url, self.url)
 290         self.assertEqual(self.client.http_client.path, '/networks')
 291         expected = net_list['networks']['values']
 292         for i in range(len(r)):
 293             self.assert_dicts_are_equal(expected[i], r[i])
 294         self.client.list_networks(detail=True)
 295         self.assertEqual(self.client.http_client.url, self.url)
 296         self.assertEqual(self.client.http_client.path, '/networks/detail')
 297
 298     @patch('%s.perform_request' % khttp, return_value=FR())
 299     def test_list_network_nics(self, PR):
 300         net_id = net_recv['network']['id']
 301         FR.json = net_recv
 302         r = self.client.list_network_nics(net_id)
 303         self.assertEqual(self.client.http_client.url, self.url)
 304         self.assertEqual(
 305             self.client.http_client.path,
 306             '/networks/%s' % net_id)
 307         expected = net_recv['network']['attachments']['values']
 308         for i in range(len(r)):
 309             self.assert_dicts_are_equal(r[i], expected[i])
 310
 311     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 312     def test_disconnect_network_nics(self, NP):
 313         net_id = net_recv['network']['id']
 314         nics = ['nic1', 'nic2', 'nic3']
 315         with patch.object(
 316                 CycladesClient,
 317                 'list_network_nics',
 318                 return_value=nics) as lnn:
 319             self.client.disconnect_network_nics(net_id)
 320             lnn.assert_called_once_with(net_id)
 321             for i in range(len(nics)):
 322                 expected = call(net_id, 'action', json_data=dict(
 323                     remove=dict(attachment=nics[i])))
 324                 self.assertEqual(expected, NP.mock_calls[i])
 325
 326     @patch('%s.perform_request' % khttp, return_value=FR())
 327     def test_get_network_details(self, PR):
 328         FR.json = net_recv
 329         net_id = net_recv['network']['id']
 330         r = self.client.get_network_details(net_id)
 331         self.assertEqual(self.client.http_client.url, self.url)
 332         self.assertEqual(
 333             self.client.http_client.path,
 334             '/networks/%s' % net_id)
 335         self.assert_dicts_are_equal(r, net_recv['network'])
 336
 337     @patch('%s.perform_request' % khttp, return_value=FR())
 338     def test_update_network_name(self, PR):
 339         net_id = net_recv['network']['id']
 340         new_name = '%s_new' % net_id
 341         FR.status_code = 204
 342         self.client.update_network_name(net_id, new_name)
 343         self.assertEqual(self.client.http_client.url, self.url)
 344         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 345         (method, data, a_headers, a_params) = PR.call_args[0]
 346         self.assert_dicts_are_equal(
 347             dict(network=dict(name=new_name)),
 348             loads(data))
 349
 350     @patch('%s.perform_request' % khttp, return_value=FR())
 351     def test_delete_network(self, PR):
 352         net_id = net_recv['network']['id']
 353         FR.status_code = 204
 354         self.client.delete_network(net_id)
 355         self.assertEqual(self.client.http_client.url, self.url)
 356         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 357
 358 if __name__ == '__main__':
 359     from sys import argv
 360     from kamaki.clients.test import runTestCase
 361     runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])