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, Mock, call
  34 from unittest import TestCase
  35 from json import loads
  36
  37 from kamaki.clients import Client, 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_send = dict(server=dict(
  45     flavorRef=fid,
  46     name=vm_name,
  47     imageRef=img_ref,
  48     metadata=dict(os="debian", users="root")))
  49 vm_recv = dict(server=dict(
  50     status="BUILD",
  51     updated="2013-03-01T10:04:00.637152+00:00",
  52     hostId="",
  53     name=vm_name,
  54     imageRef=img_ref,
  55     created="2013-03-01T10:04:00.087324+00:00",
  56     flavorRef=fid,
  57     adminPass="n0n3sh@11p@55",
  58     suspended=False,
  59     progress=0,
  60     id=31173,
  61     metadata=dict(values=dict(os="debian", users="root"))))
  62 img_recv = dict(image=dict(
  63     status="ACTIVE",
  64     updated="2013-02-26T11:10:14+00:00",
  65     name="Debian Base",
  66     created="2013-02-26T11:03:29+00:00",
  67     progress=100,
  68     id=img_ref,
  69     metadata=dict(values=dict(
  70         partition_table="msdos",
  71         kernel="2.6.32",
  72         osfamily="linux",
  73         users="root",
  74         gui="No GUI",
  75         sortorder="1",
  76         os="debian",
  77         root_partition="1",
  78         description="Debian 6.0.7 (Squeeze) Base System"))))
  79 vm_list = dict(servers=dict(values=[
  80     dict(name='n1', id=1),
  81     dict(name='n2', id=2)]))
  82 flavor_list = dict(flavors=dict(values=[
  83         dict(id=41, name="C1R1024D20"),
  84         dict(id=42, name="C1R1024D40"),
  85         dict(id=43, name="C1R1028D20")]))
  86 img_list = dict(images=dict(values=[
  87     dict(name="maelstrom", id="0fb03e45-7d5a-4515-bd4e-e6bbf6457f06"),
  88     dict(name="edx_saas", id="1357163d-5fd8-488e-a117-48734c526206"),
  89     dict(name="Debian_Wheezy_Base", id="1f8454f0-8e3e-4b6c-ab8e-5236b728dffe"),
  90     dict(name="CentOS", id="21894b48-c805-4568-ac8b-7d4bb8eb533d"),
  91     dict(name="Ubuntu Desktop", id="37bc522c-c479-4085-bfb9-464f9b9e2e31"),
  92     dict(name="Ubuntu 12.10", id="3a24fef9-1a8c-47d1-8f11-e07bd5e544fd"),
  93     dict(name="Debian Base", id="40ace203-6254-4e17-a5cb-518d55418a7d"),
  94     dict(name="ubuntu_bundled", id="5336e265-5c7c-4127-95cb-2bf832a79903")]))
  95 net_send = dict(network=dict(dhcp=False, name='someNet'))
  96 net_recv = dict(network=dict(
  97     status="PENDING",
  98     updated="2013-03-05T15:04:51.758780+00:00",
  99     name="someNet",
 100     created="2013-03-05T15:04:51.758728+00:00",
 101     cidr6=None,
 102     id="2130",
 103     gateway6=None,
 104     public=False,
 105     dhcp=False,
 106     cidr="192.168.1.0/24",
 107     type="MAC_FILTERED",
 108     gateway=None,
 109     attachments=dict(values=[dict(name='att1'), dict(name='att2')])))
 110 net_list = dict(networks=dict(values=[
 111     dict(id=1, name='n1'),
 112     dict(id=2, name='n2'),
 113     dict(id=3, name='n3')]))
 114
 115
 116 class FR(object):
 117     """FR stands for Fake Response"""
 118     json = vm_recv
 119     headers = {}
 120     content = json
 121     status = None
 122     status_code = 200
 123
 124     def release(self):
 125         pass
 126
 127 khttp = 'kamaki.clients.connection.kamakicon.KamakiHTTPConnection'
 128 cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
 129
 130
 131 class Cyclades(TestCase):
 132
 133     def assert_dicts_are_equal(self, d1, d2):
 134         for k, v in d1.items():
 135             self.assertTrue(k in d2)
 136             if isinstance(v, dict):
 137                 self.assert_dicts_are_equal(v, d2[k])
 138             else:
 139                 self.assertEqual(unicode(v), unicode(d2[k]))
 140
 141     """Set up a Cyclades thorough test"""
 142     def setUp(self):
 143         self.url = 'http://cyclades.example.com'
 144         self.token = 'cyc14d3s70k3n'
 145         self.client = CycladesClient(self.url, self.token)
 146         from kamaki.clients.connection.kamakicon import KamakiHTTPConnection
 147         self.C = KamakiHTTPConnection
 148
 149     def tearDown(self):
 150         FR.status_code = 200
 151         FR.json = vm_recv
 152
 153     def test_list_servers(self):
 154         FR.json = vm_list
 155         with patch.object(
 156                 self.C,
 157                 'perform_request',
 158                 return_value=FR()) as perform_req:
 159             r = self.client.list_servers()
 160             self.assertEqual(self.client.http_client.url, self.url)
 161             self.assertEqual(self.client.http_client.path, '/servers')
 162             (method, data, a_headers, a_params) = perform_req.call_args[0]
 163             self.assert_dicts_are_equal(dict(values=r), vm_list['servers'])
 164             r = self.client.list_servers(detail=True)
 165             self.assertEqual(self.client.http_client.url, self.url)
 166             self.assertEqual(self.client.http_client.path, '/servers/detail')
 167         with patch.object(
 168                 CycladesClientApi,
 169                 'servers_get',
 170                 return_value=FR()) as servers_get:
 171             self.client.list_servers(changes_since=True)
 172             self.assertTrue(servers_get.call_args[1]['changes_since'])
 173
 174     @patch('%s.perform_request' % khttp, return_value=FR())
 175     def test_shutdown_server(self, PR):
 176         vm_id = vm_recv['server']['id']
 177         FR.status_code = 202
 178         self.client.shutdown_server(vm_id)
 179         self.assertEqual(self.client.http_client.url, self.url)
 180         self.assertEqual(
 181             self.client.http_client.path,
 182             '/servers/%s/action' % vm_id)
 183         self.assertEqual(
 184             PR.call_args[0],
 185             ('post',  '{"shutdown": {}}', {}, {}))
 186
 187     @patch('%s.perform_request' % khttp, return_value=FR())
 188     def test_start_server(self, PR):
 189         vm_id = vm_recv['server']['id']
 190         FR.status_code = 202
 191         self.client.start_server(vm_id)
 192         self.assertEqual(self.client.http_client.url, self.url)
 193         self.assertEqual(
 194             self.client.http_client.path,
 195             '/servers/%s/action' % vm_id)
 196         self.assertEqual(PR.call_args[0], ('post',  '{"start": {}}', {}, {}))
 197
 198     @patch('%s.perform_request' % khttp, return_value=FR())
 199     def test_get_server_console(self, PR):
 200         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 201         FR.json = cnsl
 202         vm_id = vm_recv['server']['id']
 203         r = self.client.get_server_console(vm_id)
 204         self.assertEqual(self.client.http_client.url, self.url)
 205         self.assertEqual(
 206             self.client.http_client.path,
 207             '/servers/%s/action' % vm_id)
 208         self.assert_dicts_are_equal(cnsl['console'], r)
 209         self.assertEqual(
 210             PR.call_args[0],
 211             ('post',  '{"console": {"type": "vnc"}}', {}, {}))
 212
 213     def test_get_firewall_profile(self):
 214         vm_id = vm_recv['server']['id']
 215         v = 'Some profile'
 216         ret = {'attachments': {'values': [{'firewallProfile': v, 1:1}]}}
 217         with patch.object(
 218                 CycladesClient,
 219                 'get_server_details',
 220                 return_value=ret) as GSD:
 221             r = self.client.get_firewall_profile(vm_id)
 222             self.assertEqual(r, v)
 223             self.assertEqual(GSD.call_args[0], (vm_id,))
 224             ret['attachments']['values'][0].pop('firewallProfile')
 225             self.assertRaises(
 226                 ClientError,
 227                 self.client.get_firewall_profile,
 228                 vm_id)
 229
 230     @patch('%s.perform_request' % khttp, return_value=FR())
 231     def test_set_firewall_profile(self, PR):
 232         vm_id = vm_recv['server']['id']
 233         v = 'Some profile'
 234         FR.status_code = 202
 235         self.client.set_firewall_profile(vm_id, v)
 236         self.assertEqual(self.client.http_client.url, self.url)
 237         self.assertEqual(
 238             self.client.http_client.path,
 239             '/servers/%s/action' % vm_id)
 240         self.assertEqual(PR.call_args[0], (
 241             'post',
 242             '{"firewallProfile": {"profile": "%s"}}' % v,
 243             {},
 244             {}))
 245
 246     @patch('%s.perform_request' % khttp, return_value=FR())
 247     def test_get_server_stats(self, PR):
 248         vm_id = vm_recv['server']['id']
 249         stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
 250         FR.json = dict(stats=stats)
 251         r = self.client.get_server_stats(vm_id)
 252         self.assertEqual(self.client.http_client.url, self.url)
 253         self.assertEqual(
 254             self.client.http_client.path,
 255             '/servers/%s/stats' % vm_id)
 256         self.assert_dicts_are_equal(stats, r)
 257
 258     @patch('%s.perform_request' % khttp, return_value=FR())
 259     def test_create_network(self, PR):
 260         net_name = net_send['network']['name']
 261         FR.json = net_recv
 262         FR.status_code = 202
 263         full_args = dict(
 264                 cidr='192.168.0.0/24',
 265                 gateway='192.168.0.1',
 266                 type='MAC_FILTERED',
 267                 dhcp=True)
 268         test_args = dict(full_args)
 269         test_args.update(dict(empty=None, full=None))
 270         for arg, val in test_args.items():
 271             kwargs = {} if arg == 'empty' else full_args if (
 272                 arg == 'full') else {arg: val}
 273             r = self.client.create_network(net_name, **kwargs)
 274             self.assertEqual(self.client.http_client.url, self.url)
 275             self.assertEqual(
 276                 self.client.http_client.path,
 277                 '/networks')
 278             self.assert_dicts_are_equal(r, net_recv['network'])
 279             data = PR.call_args[0][1]
 280             expected = dict(network=dict(net_send['network']))
 281             expected['network'].update(kwargs)
 282             self.assert_dicts_are_equal(loads(data), expected)
 283
 284     @patch('%s.perform_request' % khttp, return_value=FR())
 285     def test_connect_server(self, PR):
 286         vm_id = vm_recv['server']['id']
 287         net_id = net_recv['network']['id']
 288         FR.status_code = 202
 289         self.client.connect_server(vm_id, net_id)
 290         self.assertEqual(self.client.http_client.url, self.url)
 291         self.assertEqual(
 292             self.client.http_client.path,
 293             '/networks/%s/action' % net_id)
 294         self.assertEqual(
 295             PR.call_args[0],
 296             ('post', '{"add": {"serverRef": %s}}' % vm_id, {}, {}))
 297
 298     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 299     def test_disconnect_server(self, NP):
 300         vm_id = vm_recv['server']['id']
 301         net_id = net_recv['network']['id']
 302         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 303         vm_nics = [
 304             dict(id=nic_id, network_id=net_id),
 305             dict(id='another-nic-id', network_id='another-net-id'),
 306             dict(id=nic_id * 2, network_id=net_id * 2)]
 307         with patch.object(
 308                 CycladesClient,
 309                 'list_server_nics',
 310                 return_value=vm_nics) as LSN:
 311             r = self.client.disconnect_server(vm_id, nic_id)
 312             self.assertEqual(r, 1)
 313             self.assertEqual(LSN.call_args[0], (vm_id,))
 314             self.assertEqual(NP.call_args[0], (net_id, 'action'))
 315             self.assertEqual(
 316                 NP.call_args[1],
 317                 dict(json_data=dict(remove=dict(attachment=nic_id))))
 318
 319     @patch('%s.perform_request' % khttp, return_value=FR())
 320     def test_list_server_nics(self, PR):
 321         vm_id = vm_recv['server']['id']
 322         nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
 323         FR.json = nics
 324         r = self.client.list_server_nics(vm_id)
 325         self.assertEqual(self.client.http_client.url, self.url)
 326         self.assertEqual(
 327             self.client.http_client.path,
 328             '/servers/%s/ips' % vm_id)
 329         expected = nics['addresses']['values']
 330         for i in range(len(r)):
 331             self.assert_dicts_are_equal(r[i], expected[i])
 332
 333     @patch('%s.perform_request' % khttp, return_value=FR())
 334     def test_list_networks(self, PR):
 335         FR.json = net_list
 336         r = self.client.list_networks()
 337         self.assertEqual(self.client.http_client.url, self.url)
 338         self.assertEqual(self.client.http_client.path, '/networks')
 339         expected = net_list['networks']['values']
 340         for i in range(len(r)):
 341             self.assert_dicts_are_equal(expected[i], r[i])
 342         self.client.list_networks(detail=True)
 343         self.assertEqual(self.client.http_client.url, self.url)
 344         self.assertEqual(self.client.http_client.path, '/networks/detail')
 345
 346     @patch('%s.perform_request' % khttp, return_value=FR())
 347     def test_list_network_nics(self, PR):
 348         net_id = net_recv['network']['id']
 349         FR.json = net_recv
 350         r = self.client.list_network_nics(net_id)
 351         self.assertEqual(self.client.http_client.url, self.url)
 352         self.assertEqual(
 353             self.client.http_client.path,
 354             '/networks/%s' % net_id)
 355         expected = net_recv['network']['attachments']['values']
 356         for i in range(len(r)):
 357             self.assert_dicts_are_equal(r[i], expected[i])
 358
 359     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 360     def test_disconnect_network_nics(self, NP):
 361         net_id = net_recv['network']['id']
 362         nics = ['nic1', 'nic2', 'nic3']
 363         with patch.object(
 364                 CycladesClient,
 365                 'list_network_nics',
 366                 return_value=nics) as lnn:
 367             self.client.disconnect_network_nics(net_id)
 368             lnn.assert_called_once_with(net_id)
 369             for i in range(len(nics)):
 370                 expected = call(net_id, 'action', json_data=dict(
 371                     remove=dict(attachment=nics[i])))
 372                 self.assertEqual(expected, NP.mock_calls[i])
 373
 374     @patch('%s.perform_request' % khttp, return_value=FR())
 375     def test_get_network_details(self, PR):
 376         FR.json = net_recv
 377         net_id = net_recv['network']['id']
 378         r = self.client.get_network_details(net_id)
 379         self.assertEqual(self.client.http_client.url, self.url)
 380         self.assertEqual(
 381             self.client.http_client.path,
 382             '/networks/%s' % net_id)
 383         self.assert_dicts_are_equal(r, net_recv['network'])
 384
 385     @patch('%s.perform_request' % khttp, return_value=FR())
 386     def test_update_network_name(self, PR):
 387         net_id = net_recv['network']['id']
 388         new_name = '%s_new' % net_id
 389         FR.status_code = 204
 390         self.client.update_network_name(net_id, new_name)
 391         self.assertEqual(self.client.http_client.url, self.url)
 392         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 393         (method, data, a_headers, a_params) = PR.call_args[0]
 394         self.assert_dicts_are_equal(
 395             dict(network=dict(name=new_name)),
 396             loads(data))
 397
 398     @patch('%s.perform_request' % khttp, return_value=FR())
 399     def test_delete_image(self, PR):
 400         FR.status_code = 204
 401         self.client.delete_image(img_ref)
 402         self.assertEqual(self.client.http_client.url, self.url)
 403         self.assertEqual(self.client.http_client.path, '/images/%s' % img_ref)
 404
 405     @patch('%s.perform_request' % khttp, return_value=FR())
 406     def test_delete_network(self, PR):
 407         net_id = net_recv['network']['id']
 408         FR.status_code = 204
 409         self.client.delete_network(net_id)
 410         self.assertEqual(self.client.http_client.url, self.url)
 411         self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id)
 412
 413     @patch('%s.perform_request' % khttp, return_value=FR())
 414     def test_create_image_metadata(self, PR):
 415         metadata = dict(m1='v1', m2='v2', m3='v3')
 416         FR.json = dict(meta=img_recv['image'])
 417         self.assertRaises(
 418             ClientError,
 419             self.client.create_image_metadata,
 420             img_ref, 'key', 'value')
 421         FR.status_code = 201
 422         for k, v in metadata.items():
 423             r = self.client.create_image_metadata(img_ref, k, v)
 424             self.assertEqual(self.client.http_client.url, self.url)
 425             self.assertEqual(
 426                 self.client.http_client.path,
 427                 '/images/%s/meta/%s' % (img_ref, k))
 428             (method, data, a_headers, a_params) = PR.call_args[0]
 429             self.assertEqual(dict(meta={k: v}), loads(data))
 430             self.assert_dicts_are_equal(r, img_recv['image'])
 431
 432     @patch('%s.images_post' % cyclades_pkg, return_value=FR())
 433     def test_update_image_metadata(self, images_post):
 434         metadata = dict(m1='v1', m2='v2', m3='v3')
 435         FR.json = dict(metadata=metadata)
 436         r = self.client.update_image_metadata(img_ref, **metadata)
 437         self.assert_dicts_are_equal(r, metadata)
 438         (called_id, cmd) = images_post.call_args[0]
 439         self.assertEqual(called_id, img_ref)
 440         self.assertEqual(cmd, 'meta')
 441         data = images_post.call_args[1]['json_data']
 442         self.assert_dicts_are_equal(data, dict(metadata=metadata))
 443
 444     @patch('%s.images_delete' % cyclades_pkg, return_value=FR())
 445     def test_delete_image_metadata(self, images_delete):
 446         key = 'metakey'
 447         self.client.delete_image_metadata(img_ref, key)
 448         self.assertEqual(
 449             (img_ref, '/meta/' + key),
 450             images_delete.call_args[0])
 451
 452 if __name__ == '__main__':
 453     from sys import argv
 454     from kamaki.clients.test import runTestCase
 455     runTestCase(Cyclades, 'Cyclades (multi) Client', argv[1:])