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
  36 from kamaki.clients import ClientError
  37 from kamaki.clients.cyclades import CycladesClient
  38
  39 img_ref = "1m4g3-r3f3r3nc3"
  40 vm_name = "my new VM"
  41 fid = 42
  42 vm_recv = dict(server=dict(
  43     status="BUILD",
  44     updated="2013-03-01T10:04:00.637152+00:00",
  45     hostId="",
  46     name=vm_name,
  47     imageRef=img_ref,
  48     created="2013-03-01T10:04:00.087324+00:00",
  49     flavorRef=fid,
  50     adminPass="n0n3sh@11p@55",
  51     suspended=False,
  52     progress=0,
  53     id=31173,
  54     metadata=dict(values=dict(os="debian", users="root"))))
  55 vm_list = dict(servers=dict(values=[
  56     dict(name='n1', id=1),
  57     dict(name='n2', id=2)]))
  58 net_send = dict(network=dict(dhcp=False, name='someNet'))
  59 net_recv = dict(network=dict(
  60     status="PENDING",
  61     updated="2013-03-05T15:04:51.758780+00:00",
  62     name="someNet",
  63     created="2013-03-05T15:04:51.758728+00:00",
  64     cidr6=None,
  65     id="2130",
  66     gateway6=None,
  67     public=False,
  68     dhcp=False,
  69     cidr="192.168.1.0/24",
  70     type="MAC_FILTERED",
  71     gateway=None,
  72     attachments=dict(values=[dict(name='att1'), dict(name='att2')])))
  73 net_list = dict(networks=dict(values=[
  74     dict(id=1, name='n1'),
  75     dict(id=2, name='n2'),
  76     dict(id=3, name='n3')]))
  77 firewalls = dict(attachments=dict(values=[
  78     dict(firewallProfile='50m3_pr0f1L3', otherStuff='57uff')]))
  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 cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
  93
  94
  95 class Cyclades(TestCase):
  96
  97     def assert_dicts_are_equal(self, d1, d2):
  98         for k, v in d1.items():
  99             self.assertTrue(k in d2)
 100             if isinstance(v, dict):
 101                 self.assert_dicts_are_equal(v, d2[k])
 102             else:
 103                 self.assertEqual(unicode(v), unicode(d2[k]))
 104
 105     """Set up a Cyclades thorough test"""
 106     def setUp(self):
 107         self.url = 'http://cyclades.example.com'
 108         self.token = 'cyc14d3s70k3n'
 109         self.client = CycladesClient(self.url, self.token)
 110
 111     def tearDown(self):
 112         FR.status_code = 200
 113         FR.json = vm_recv
 114
 115     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 116     def test_list_servers(self, SG):
 117         FR.json = vm_list
 118         for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
 119             r = self.client.list_servers(detail=detail, changes_since=since)
 120             self.assertEqual(SG.mock_calls[-1], call(
 121                 command='detail' if detail else '',
 122                 changes_since=since))
 123             expected = vm_list['servers']['values']
 124             for i, vm in enumerate(r):
 125                 self.assert_dicts_are_equal(vm, expected[i])
 126             self.assertEqual(i + 1, len(expected))
 127
 128     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 129     def test_shutdown_server(self, SP):
 130         vm_id = vm_recv['server']['id']
 131         self.client.shutdown_server(vm_id)
 132         SP.assert_called_once_with(
 133             vm_id, 'action',
 134             json_data=dict(shutdown=dict()), success=202)
 135
 136     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 137     def test_start_server(self, SP):
 138         vm_id = vm_recv['server']['id']
 139         self.client.start_server(vm_id)
 140         SP.assert_called_once_with(
 141             vm_id, 'action',
 142             json_data=dict(start=dict()), success=202)
 143
 144     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 145     def test_get_server_console(self, SP):
 146         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 147         FR.json = cnsl
 148         vm_id = vm_recv['server']['id']
 149         r = self.client.get_server_console(vm_id)
 150         SP.assert_called_once_with(
 151             vm_id, 'action',
 152             json_data=dict(console=dict(type='vnc')), success=200)
 153         self.assert_dicts_are_equal(r, cnsl['console'])
 154
 155     def test_get_firewall_profile(self):
 156         vm_id = vm_recv['server']['id']
 157         v = firewalls['attachments']['values'][0]['firewallProfile']
 158         with patch.object(
 159                 CycladesClient, 'get_server_details',
 160                 return_value=firewalls) as GSD:
 161             r = self.client.get_firewall_profile(vm_id)
 162             GSD.assert_called_once_with(vm_id)
 163             self.assertEqual(r, v)
 164         with patch.object(
 165                 CycladesClient, 'get_server_details',
 166                 return_value=dict()):
 167             self.assertRaises(
 168                 ClientError,
 169                 self.client.get_firewall_profile,
 170                 vm_id)
 171
 172     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 173     def test_set_firewall_profile(self, SP):
 174         vm_id = vm_recv['server']['id']
 175         v = firewalls['attachments']['values'][0]['firewallProfile']
 176         self.client.set_firewall_profile(vm_id, v)
 177         SP.assert_called_once_with(
 178             vm_id, 'action',
 179             json_data=dict(firewallProfile=dict(profile=v)), success=202)
 180
 181     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 182     def test_get_server_stats(self, SG):
 183         vm_id = vm_recv['server']['id']
 184         stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
 185         FR.json = dict(stats=stats)
 186         r = self.client.get_server_stats(vm_id)
 187         SG.assert_called_once_with(vm_id, 'stats')
 188         self.assert_dicts_are_equal(stats, r)
 189
 190     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 191     def test_create_network(self, NP):
 192         net_name = net_send['network']['name']
 193         FR.json = net_recv
 194         full_args = dict(
 195                 cidr='192.168.0.0/24',
 196                 gateway='192.168.0.1',
 197                 type='MAC_FILTERED',
 198                 dhcp=True)
 199         test_args = dict(full_args)
 200         test_args.update(dict(empty=None, full=None))
 201         net_exp = dict(dhcp=False, name=net_name)
 202         for arg, val in test_args.items():
 203             kwargs = {} if arg == 'empty' else full_args if (
 204                 arg == 'full') else {arg: val}
 205             expected = dict(network=dict(net_exp))
 206             expected['network'].update(kwargs)
 207             r = self.client.create_network(net_name, **kwargs)
 208             self.assertEqual(
 209                 NP.mock_calls[-1],
 210                 call(json_data=expected, success=202))
 211             self.assert_dicts_are_equal(r, net_recv['network'])
 212
 213     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 214     def test_connect_server(self, NP):
 215         vm_id = vm_recv['server']['id']
 216         net_id = net_recv['network']['id']
 217         self.client.connect_server(vm_id, net_id)
 218         NP.assert_called_once_with(
 219             net_id, 'action',
 220             json_data=dict(add=dict(serverRef=vm_id)))
 221
 222     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 223     def test_disconnect_server(self, NP):
 224         net_id, vm_id = net_recv['network']['id'], vm_recv['server']['id']
 225         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 226         vm_nics = [
 227             dict(id=nic_id, network_id=net_id),
 228             dict(id='another-nic-id', network_id='another-net-id'),
 229             dict(id=nic_id * 2, network_id=net_id * 2)]
 230         with patch.object(
 231                 CycladesClient,
 232                 'list_server_nics',
 233                 return_value=vm_nics) as LSN:
 234             r = self.client.disconnect_server(vm_id, nic_id)
 235             LSN.assert_called_once_with(vm_id)
 236             NP.assert_called_once_with(
 237                 net_id, 'action',
 238                 json_data=dict(remove=dict(attachment=nic_id)))
 239             self.assertEqual(r, 1)
 240
 241     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 242     def test_list_server_nics(self, SG):
 243         vm_id = vm_recv['server']['id']
 244         nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
 245         FR.json = nics
 246         r = self.client.list_server_nics(vm_id)
 247         SG.assert_called_once_with(vm_id, 'ips')
 248         expected = nics['addresses']['values']
 249         for i in range(len(r)):
 250             self.assert_dicts_are_equal(r[i], expected[i])
 251         self.assertEqual(i + 1, len(r))
 252
 253     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 254     def test_list_networks(self, NG):
 255         FR.json = net_list
 256         expected = net_list['networks']['values']
 257         for detail in ('', 'detail'):
 258             r = self.client.list_networks(detail=True if detail else False)
 259             self.assertEqual(NG.mock_calls[-1], call(command=detail))
 260             for i, net in enumerate(expected):
 261                 self.assert_dicts_are_equal(r[i], net)
 262             self.assertEqual(i + 1, len(r))
 263
 264     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 265     def test_list_network_nics(self, NG):
 266         net_id = net_recv['network']['id']
 267         FR.json = net_recv
 268         r = self.client.list_network_nics(net_id)
 269         NG.assert_called_once_with(network_id=net_id)
 270         expected = net_recv['network']['attachments']['values']
 271         for i in range(len(r)):
 272             self.assert_dicts_are_equal(r[i], expected[i])
 273
 274     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 275     def test_disconnect_network_nics(self, NP):
 276         net_id = net_recv['network']['id']
 277         nics = ['nic1', 'nic2', 'nic3']
 278         with patch.object(
 279                 CycladesClient,
 280                 'list_network_nics',
 281                 return_value=nics) as LNN:
 282             self.client.disconnect_network_nics(net_id)
 283             LNN.assert_called_once_with(net_id)
 284             for i in range(len(nics)):
 285                 expected = call(net_id, 'action', json_data=dict(
 286                     remove=dict(attachment=nics[i])))
 287                 self.assertEqual(expected, NP.mock_calls[i])
 288
 289     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 290     def test_get_network_details(self, NG):
 291         FR.json = net_recv
 292         net_id = net_recv['network']['id']
 293         r = self.client.get_network_details(net_id)
 294         NG.assert_called_once_with(network_id=net_id)
 295         self.assert_dicts_are_equal(r, net_recv['network'])
 296
 297     @patch('%s.networks_put' % cyclades_pkg, return_value=FR())
 298     def test_update_network_name(self, NP):
 299         net_id = net_recv['network']['id']
 300         new_name = '%s_new' % net_id
 301         self.client.update_network_name(net_id, new_name)
 302         NP.assert_called_once_with(
 303             network_id=net_id,
 304             json_data=dict(network=dict(name=new_name)))
 305
 306     def test_delete_network(self):
 307         net_id = net_recv['network']['id']
 308         with patch.object(
 309                 CycladesClient, 'networks_delete',
 310                 return_value=FR()) as ND:
 311             self.client.delete_network(net_id)
 312             ND.assert_called_once_with(net_id)
 313         with patch.object(
 314                 CycladesClient, 'networks_delete',
 315                 side_effect=ClientError('A 421 Error', 421)):
 316             try:
 317                 self.client.delete_network(421)
 318             except ClientError as err:
 319                 self.assertEqual(err.status, 421)
 320                 self.assertEqual(err.details, [
 321                     'Network may be still connected to at least one server'])
 322
 323 if __name__ == '__main__':
 324     from sys import argv
 325     from kamaki.clients.test import runTestCase
 326     runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])