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 firewalls = dict(attachments=dict(values=[
  80     dict(firewallProfile='50m3_pr0f1L3', otherStuff='57uff')]))
  81
  82
  83 class FR(object):
  84     """FR stands for Fake Response"""
  85     json = vm_recv
  86     headers = {}
  87     content = json
  88     status = None
  89     status_code = 200
  90
  91     def release(self):
  92         pass
  93
  94 khttp = 'kamaki.clients.connection.kamakicon.KamakiHTTPConnection'
  95 cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
  96
  97
  98 class Cyclades(TestCase):
  99
 100     def assert_dicts_are_equal(self, d1, d2):
 101         for k, v in d1.items():
 102             self.assertTrue(k in d2)
 103             if isinstance(v, dict):
 104                 self.assert_dicts_are_equal(v, d2[k])
 105             else:
 106                 self.assertEqual(unicode(v), unicode(d2[k]))
 107
 108     """Set up a Cyclades thorough test"""
 109     def setUp(self):
 110         self.url = 'http://cyclades.example.com'
 111         self.token = 'cyc14d3s70k3n'
 112         self.client = CycladesClient(self.url, self.token)
 113         from kamaki.clients.connection.kamakicon import KamakiHTTPConnection
 114         self.C = KamakiHTTPConnection
 115
 116     def tearDown(self):
 117         FR.status_code = 200
 118         FR.json = vm_recv
 119
 120     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 121     def test_list_servers(self, SG):
 122         FR.json = vm_list
 123         for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
 124             r = self.client.list_servers(detail=detail, changes_since=since)
 125             self.assertEqual(SG.mock_calls[-1], call(
 126                 command='detail' if detail else '',
 127                 changes_since=since))
 128             expected = vm_list['servers']['values']
 129             for i, vm in enumerate(r):
 130                 self.assert_dicts_are_equal(vm, expected[i])
 131             self.assertEqual(i + 1, len(expected))
 132
 133     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 134     def test_shutdown_server(self, SP):
 135         vm_id = vm_recv['server']['id']
 136         self.client.shutdown_server(vm_id)
 137         self.assertEqual(SP.mock_calls[-1], call(
 138             vm_id, 'action',
 139             json_data=dict(shutdown=dict()), success=202))
 140
 141     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 142     def test_start_server(self, SP):
 143         vm_id = vm_recv['server']['id']
 144         self.client.start_server(vm_id)
 145         self.assertEqual(SP.mock_calls[-1], call(
 146             vm_id, 'action',
 147             json_data=dict(start=dict()), success=202))
 148
 149     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 150     def test_get_server_console(self, SP):
 151         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 152         FR.json = cnsl
 153         vm_id = vm_recv['server']['id']
 154         r = self.client.get_server_console(vm_id)
 155         self.assertEqual(SP.mock_calls[-1], call(
 156             vm_id, 'action',
 157             json_data=dict(console=dict(type='vnc')), success=200))
 158         self.assert_dicts_are_equal(r, cnsl['console'])
 159
 160     def test_get_firewall_profile(self):
 161         vm_id = vm_recv['server']['id']
 162         v = firewalls['attachments']['values'][0]['firewallProfile']
 163         with patch.object(
 164                 CycladesClient, 'get_server_details',
 165                 return_value=firewalls) as GSD:
 166             r = self.client.get_firewall_profile(vm_id)
 167             self.assertEqual(r, v)
 168             self.assertEqual(GSD.mock_calls[-1], call(vm_id))
 169         with patch.object(
 170                 CycladesClient, 'get_server_details',
 171                 return_value=dict()):
 172             self.assertRaises(
 173                 ClientError,
 174                 self.client.get_firewall_profile,
 175                 vm_id)
 176
 177     @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
 178     def test_set_firewall_profile(self, SP):
 179         vm_id = vm_recv['server']['id']
 180         v = firewalls['attachments']['values'][0]['firewallProfile']
 181         self.client.set_firewall_profile(vm_id, v)
 182         self.assertEqual(SP.mock_calls[-1], call(
 183             vm_id, 'action',
 184             json_data=dict(firewallProfile=dict(profile=v)), success=202))
 185
 186     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 187     def test_get_server_stats(self, SG):
 188         vm_id = vm_recv['server']['id']
 189         stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
 190         FR.json = dict(stats=stats)
 191         r = self.client.get_server_stats(vm_id)
 192         self.assertEqual(SG.mock_calls[-1], call(vm_id, 'stats'))
 193         self.assert_dicts_are_equal(stats, r)
 194
 195     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 196     def test_create_network(self, NP):
 197         net_name = net_send['network']['name']
 198         FR.json = net_recv
 199         full_args = dict(
 200                 cidr='192.168.0.0/24',
 201                 gateway='192.168.0.1',
 202                 type='MAC_FILTERED',
 203                 dhcp=True)
 204         test_args = dict(full_args)
 205         test_args.update(dict(empty=None, full=None))
 206         net_exp = dict(dhcp=False, name=net_name)
 207         for arg, val in test_args.items():
 208             kwargs = {} if arg == 'empty' else full_args if (
 209                 arg == 'full') else {arg: val}
 210             expected = dict(network=dict(net_exp))
 211             expected['network'].update(kwargs)
 212             r = self.client.create_network(net_name, **kwargs)
 213             self.assertEqual(
 214                 NP.mock_calls[-1],
 215                 call(json_data=expected, success=202))
 216             self.assert_dicts_are_equal(r, net_recv['network'])
 217
 218     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 219     def test_connect_server(self, NP):
 220         vm_id = vm_recv['server']['id']
 221         net_id = net_recv['network']['id']
 222         self.client.connect_server(vm_id, net_id)
 223         self.assertEqual(NP.mock_calls[-1], call(
 224             net_id, 'action',
 225             json_data=dict(add=dict(serverRef=vm_id))))
 226
 227     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 228     def test_disconnect_server(self, NP):
 229         net_id, vm_id = net_recv['network']['id'], vm_recv['server']['id']
 230         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 231         vm_nics = [
 232             dict(id=nic_id, network_id=net_id),
 233             dict(id='another-nic-id', network_id='another-net-id'),
 234             dict(id=nic_id * 2, network_id=net_id * 2)]
 235         with patch.object(
 236                 CycladesClient,
 237                 'list_server_nics',
 238                 return_value=vm_nics) as LSN:
 239             r = self.client.disconnect_server(vm_id, nic_id)
 240             self.assertEqual(r, 1)
 241             self.assertEqual(LSN.mock_calls[-1], call(vm_id,))
 242             self.assertEqual(NP.mock_calls[-1], call(
 243                 net_id, 'action',
 244                 json_data=dict(remove=dict(attachment=nic_id))))
 245
 246     @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
 247     def test_list_server_nics(self, SG):
 248         vm_id = vm_recv['server']['id']
 249         nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
 250         FR.json = nics
 251         r = self.client.list_server_nics(vm_id)
 252         self.assertEqual(SG.mock_calls[-1], call(vm_id, 'ips'))
 253         expected = nics['addresses']['values']
 254         for i in range(len(r)):
 255             self.assert_dicts_are_equal(r[i], expected[i])
 256         self.assertEqual(i + 1, len(r))
 257
 258     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 259     def test_list_networks(self, NG):
 260         FR.json = net_list
 261         expected = net_list['networks']['values']
 262         for detail in ('', 'detail'):
 263             r = self.client.list_networks(detail=True if detail else False)
 264             self.assertEqual(NG.mock_calls[-1], call(command=detail))
 265             for i, net in enumerate(expected):
 266                 self.assert_dicts_are_equal(r[i], net)
 267             self.assertEqual(i + 1, len(r))
 268
 269     @patch('%s.perform_request' % khttp, return_value=FR())
 270     def test_list_network_nics(self, PR):
 271         net_id = net_recv['network']['id']
 272         FR.json = net_recv
 273         r = self.client.list_network_nics(net_id)
 274         self.assertEqual(self.client.http_client.url, self.url)
 275         self.assertEqual(
 276             self.client.http_client.path,
 277             '/networks/%s' % net_id)
 278         expected = net_recv['network']['attachments']['values']
 279         for i in range(len(r)):
 280             self.assert_dicts_are_equal(r[i], expected[i])
 281
 282     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 283     def test_disconnect_network_nics(self, NP):
 284         net_id = net_recv['network']['id']
 285         nics = ['nic1', 'nic2', 'nic3']
 286         with patch.object(
 287                 CycladesClient,
 288                 'list_network_nics',
 289                 return_value=nics) as lnn:
 290             self.client.disconnect_network_nics(net_id)
 291             lnn.assert_called_once_with(net_id)
 292             for i in range(len(nics)):
 293                 expected = call(net_id, 'action', json_data=dict(
 294                     remove=dict(attachment=nics[i])))
 295                 self.assertEqual(expected, NP.mock_calls[i])
 296
 297     @patch('%s.perform_request' % khttp, return_value=FR())
 298     def test_get_network_details(self, PR):
 299         FR.json = net_recv
 300         net_id = net_recv['network']['id']
 301         r = self.client.get_network_details(net_id)
 302         self.assertEqual(self.client.http_client.url, self.url)
 303         self.assertEqual(
 304             self.client.http_client.path,
 305             '/networks/%s' % net_id)
 306         self.assert_dicts_are_equal(r, net_recv['network'])
 307
 308     @patch('%s.perform_request' % khttp, return_value=FR())
 309     def test_update_network_name(self, PR):
 310         net_id = net_recv['network']['id']
 311         new_name = '%s_new' % net_id
 312         FR.status_code = 204
 313         self.client.update_network_name(net_id, new_name)
 314         self.assertEqual(self.client.http_client.url, self.url)
 315         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 316         (method, data, a_headers, a_params) = PR.call_args[0]
 317         self.assert_dicts_are_equal(
 318             dict(network=dict(name=new_name)),
 319             loads(data))
 320
 321     @patch('%s.perform_request' % khttp, return_value=FR())
 322     def test_delete_network(self, PR):
 323         net_id = net_recv['network']['id']
 324         FR.status_code = 204
 325         self.client.delete_network(net_id)
 326         self.assertEqual(self.client.http_client.url, self.url)
 327         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 328
 329 if __name__ == '__main__':
 330     from sys import argv
 331     from kamaki.clients.test import runTestCase
 332     runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])