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.perform_request' % khttp, return_value=FR())
 132     def test_shutdown_server(self, PR):
 133         vm_id = vm_recv['server']['id']
 134         FR.status_code = 202
 135         self.client.shutdown_server(vm_id)
 136         self.assertEqual(self.client.http_client.url, self.url)
 137         self.assertEqual(
 138             self.client.http_client.path,
 139             '/servers/%s/action' % vm_id)
 140         self.assertEqual(
 141             PR.call_args[0],
 142             ('post',  '{"shutdown": {}}', {}, {}))
 143
 144     @patch('%s.perform_request' % khttp, return_value=FR())
 145     def test_start_server(self, PR):
 146         vm_id = vm_recv['server']['id']
 147         FR.status_code = 202
 148         self.client.start_server(vm_id)
 149         self.assertEqual(self.client.http_client.url, self.url)
 150         self.assertEqual(
 151             self.client.http_client.path,
 152             '/servers/%s/action' % vm_id)
 153         self.assertEqual(PR.call_args[0], ('post',  '{"start": {}}', {}, {}))
 154
 155     @patch('%s.perform_request' % khttp, return_value=FR())
 156     def test_get_server_console(self, PR):
 157         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 158         FR.json = cnsl
 159         vm_id = vm_recv['server']['id']
 160         r = self.client.get_server_console(vm_id)
 161         self.assertEqual(self.client.http_client.url, self.url)
 162         self.assertEqual(
 163             self.client.http_client.path,
 164             '/servers/%s/action' % vm_id)
 165         self.assert_dicts_are_equal(cnsl['console'], r)
 166         self.assertEqual(
 167             PR.call_args[0],
 168             ('post',  '{"console": {"type": "vnc"}}', {}, {}))
 169
 170     def test_get_firewall_profile(self):
 171         vm_id = vm_recv['server']['id']
 172         v = 'Some profile'
 173         ret = {'attachments': {'values': [{'firewallProfile': v, 1:1}]}}
 174         with patch.object(
 175                 CycladesClient,
 176                 'get_server_details',
 177                 return_value=ret) as GSD:
 178             r = self.client.get_firewall_profile(vm_id)
 179             self.assertEqual(r, v)
 180             self.assertEqual(GSD.call_args[0], (vm_id,))
 181             ret['attachments']['values'][0].pop('firewallProfile')
 182             self.assertRaises(
 183                 ClientError,
 184                 self.client.get_firewall_profile,
 185                 vm_id)
 186
 187     @patch('%s.perform_request' % khttp, return_value=FR())
 188     def test_set_firewall_profile(self, PR):
 189         vm_id = vm_recv['server']['id']
 190         v = 'Some profile'
 191         FR.status_code = 202
 192         self.client.set_firewall_profile(vm_id, v)
 193         self.assertEqual(self.client.http_client.url, self.url)
 194         self.assertEqual(
 195             self.client.http_client.path,
 196             '/servers/%s/action' % vm_id)
 197         self.assertEqual(PR.call_args[0], (
 198             'post',
 199             '{"firewallProfile": {"profile": "%s"}}' % v,
 200             {},
 201             {}))
 202
 203     @patch('%s.perform_request' % khttp, return_value=FR())
 204     def test_get_server_stats(self, PR):
 205         vm_id = vm_recv['server']['id']
 206         stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
 207         FR.json = dict(stats=stats)
 208         r = self.client.get_server_stats(vm_id)
 209         self.assertEqual(self.client.http_client.url, self.url)
 210         self.assertEqual(
 211             self.client.http_client.path,
 212             '/servers/%s/stats' % vm_id)
 213         self.assert_dicts_are_equal(stats, r)
 214
 215     @patch('%s.perform_request' % khttp, return_value=FR())
 216     def test_create_network(self, PR):
 217         net_name = net_send['network']['name']
 218         FR.json = net_recv
 219         FR.status_code = 202
 220         full_args = dict(
 221                 cidr='192.168.0.0/24',
 222                 gateway='192.168.0.1',
 223                 type='MAC_FILTERED',
 224                 dhcp=True)
 225         test_args = dict(full_args)
 226         test_args.update(dict(empty=None, full=None))
 227         for arg, val in test_args.items():
 228             kwargs = {} if arg == 'empty' else full_args if (
 229                 arg == 'full') else {arg: val}
 230             r = self.client.create_network(net_name, **kwargs)
 231             self.assertEqual(self.client.http_client.url, self.url)
 232             self.assertEqual(
 233                 self.client.http_client.path,
 234                 '/networks')
 235             self.assert_dicts_are_equal(r, net_recv['network'])
 236             data = PR.call_args[0][1]
 237             expected = dict(network=dict(net_send['network']))
 238             expected['network'].update(kwargs)
 239             self.assert_dicts_are_equal(loads(data), expected)
 240
 241     @patch('%s.perform_request' % khttp, return_value=FR())
 242     def test_connect_server(self, PR):
 243         vm_id = vm_recv['server']['id']
 244         net_id = net_recv['network']['id']
 245         FR.status_code = 202
 246         self.client.connect_server(vm_id, net_id)
 247         self.assertEqual(self.client.http_client.url, self.url)
 248         self.assertEqual(
 249             self.client.http_client.path,
 250             '/networks/%s/action' % net_id)
 251         self.assertEqual(
 252             PR.call_args[0],
 253             ('post', '{"add": {"serverRef": %s}}' % vm_id, {}, {}))
 254
 255     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 256     def test_disconnect_server(self, NP):
 257         vm_id = vm_recv['server']['id']
 258         net_id = net_recv['network']['id']
 259         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 260         vm_nics = [
 261             dict(id=nic_id, network_id=net_id),
 262             dict(id='another-nic-id', network_id='another-net-id'),
 263             dict(id=nic_id * 2, network_id=net_id * 2)]
 264         with patch.object(
 265                 CycladesClient,
 266                 'list_server_nics',
 267                 return_value=vm_nics) as LSN:
 268             r = self.client.disconnect_server(vm_id, nic_id)
 269             self.assertEqual(r, 1)
 270             self.assertEqual(LSN.call_args[0], (vm_id,))
 271             self.assertEqual(NP.call_args[0], (net_id, 'action'))
 272             self.assertEqual(
 273                 NP.call_args[1],
 274                 dict(json_data=dict(remove=dict(attachment=nic_id))))
 275
 276     @patch('%s.perform_request' % khttp, return_value=FR())
 277     def test_list_server_nics(self, PR):
 278         vm_id = vm_recv['server']['id']
 279         nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
 280         FR.json = nics
 281         r = self.client.list_server_nics(vm_id)
 282         self.assertEqual(self.client.http_client.url, self.url)
 283         self.assertEqual(
 284             self.client.http_client.path,
 285             '/servers/%s/ips' % vm_id)
 286         expected = nics['addresses']['values']
 287         for i in range(len(r)):
 288             self.assert_dicts_are_equal(r[i], expected[i])
 289
 290     @patch('%s.perform_request' % khttp, return_value=FR())
 291     def test_list_networks(self, PR):
 292         FR.json = net_list
 293         r = self.client.list_networks()
 294         self.assertEqual(self.client.http_client.url, self.url)
 295         self.assertEqual(self.client.http_client.path, '/networks')
 296         expected = net_list['networks']['values']
 297         for i in range(len(r)):
 298             self.assert_dicts_are_equal(expected[i], r[i])
 299         self.client.list_networks(detail=True)
 300         self.assertEqual(self.client.http_client.url, self.url)
 301         self.assertEqual(self.client.http_client.path, '/networks/detail')
 302
 303     @patch('%s.perform_request' % khttp, return_value=FR())
 304     def test_list_network_nics(self, PR):
 305         net_id = net_recv['network']['id']
 306         FR.json = net_recv
 307         r = self.client.list_network_nics(net_id)
 308         self.assertEqual(self.client.http_client.url, self.url)
 309         self.assertEqual(
 310             self.client.http_client.path,
 311             '/networks/%s' % net_id)
 312         expected = net_recv['network']['attachments']['values']
 313         for i in range(len(r)):
 314             self.assert_dicts_are_equal(r[i], expected[i])
 315
 316     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 317     def test_disconnect_network_nics(self, NP):
 318         net_id = net_recv['network']['id']
 319         nics = ['nic1', 'nic2', 'nic3']
 320         with patch.object(
 321                 CycladesClient,
 322                 'list_network_nics',
 323                 return_value=nics) as lnn:
 324             self.client.disconnect_network_nics(net_id)
 325             lnn.assert_called_once_with(net_id)
 326             for i in range(len(nics)):
 327                 expected = call(net_id, 'action', json_data=dict(
 328                     remove=dict(attachment=nics[i])))
 329                 self.assertEqual(expected, NP.mock_calls[i])
 330
 331     @patch('%s.perform_request' % khttp, return_value=FR())
 332     def test_get_network_details(self, PR):
 333         FR.json = net_recv
 334         net_id = net_recv['network']['id']
 335         r = self.client.get_network_details(net_id)
 336         self.assertEqual(self.client.http_client.url, self.url)
 337         self.assertEqual(
 338             self.client.http_client.path,
 339             '/networks/%s' % net_id)
 340         self.assert_dicts_are_equal(r, net_recv['network'])
 341
 342     @patch('%s.perform_request' % khttp, return_value=FR())
 343     def test_update_network_name(self, PR):
 344         net_id = net_recv['network']['id']
 345         new_name = '%s_new' % net_id
 346         FR.status_code = 204
 347         self.client.update_network_name(net_id, new_name)
 348         self.assertEqual(self.client.http_client.url, self.url)
 349         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 350         (method, data, a_headers, a_params) = PR.call_args[0]
 351         self.assert_dicts_are_equal(
 352             dict(network=dict(name=new_name)),
 353             loads(data))
 354
 355     @patch('%s.perform_request' % khttp, return_value=FR())
 356     def test_delete_network(self, PR):
 357         net_id = net_recv['network']['id']
 358         FR.status_code = 204
 359         self.client.delete_network(net_id)
 360         self.assertEqual(self.client.http_client.url, self.url)
 361         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 362
 363 if __name__ == '__main__':
 364     from sys import argv
 365     from kamaki.clients.test import runTestCase
 366     runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])