Commit 10a4e2a8 authored by Christos Stavrakakis's avatar Christos Stavrakakis

Refactor tests

parent 04deb668
# Copyright 2012 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
import json
from synnefo.api.tests import BaseAPITest
from synnefo.db.models import Flavor
from synnefo.db.models_factory import FlavorFactory
class FlavorAPITest(BaseAPITest):
def setUp(self):
self.flavor1 = FlavorFactory()
self.flavor2 = FlavorFactory(deleted=True)
self.flavor3 = FlavorFactory()
def test_flavor_list(self):
"""Test if the expected list of flavors is returned."""
response = self.get('/api/v1.1/flavors')
self.assertSuccess(response)
api_flavors = json.loads(response.content)['flavors']['values']
db_flavors = Flavor.objects.filter(deleted=False)
self.assertEqual(len(api_flavors), len(db_flavors))
for api_flavor in api_flavors:
db_flavor = Flavor.objects.get(id=api_flavor['id'])
self.assertEqual(api_flavor['id'], db_flavor.id)
self.assertEqual(api_flavor['name'], db_flavor.name)
def test_flavors_details(self):
"""Test if the flavors details are returned."""
response = self.get('/api/v1.1/flavors/detail')
self.assertSuccess(response)
db_flavors = Flavor.objects.filter(deleted=False)
api_flavors = json.loads(response.content)['flavors']['values']
self.assertEqual(len(db_flavors), len(api_flavors))
for i in range(0, len(db_flavors)):
api_flavor = api_flavors[i]
db_flavor = Flavor.objects.get(id=db_flavors[i].id)
self.assertEqual(api_flavor['cpu'], db_flavor.cpu)
self.assertEqual(api_flavor['id'], db_flavor.id)
self.assertEqual(api_flavor['disk'], db_flavor.disk)
self.assertEqual(api_flavor['name'], db_flavor.name)
self.assertEqual(api_flavor['ram'], db_flavor.ram)
self.assertEqual(api_flavor['SNF:disk_template'],
db_flavor.disk_template)
def test_flavor_details(self):
"""Test if the expected flavor is returned."""
flavor = self.flavor3
response = self.get('/api/v1.1/flavors/%d' % flavor.id)
self.assertSuccess(response)
api_flavor = json.loads(response.content)['flavor']
db_flavor = Flavor.objects.get(id=flavor.id)
self.assertEqual(api_flavor['cpu'], db_flavor.cpu)
self.assertEqual(api_flavor['id'], db_flavor.id)
self.assertEqual(api_flavor['disk'], db_flavor.disk)
self.assertEqual(api_flavor['name'], db_flavor.name)
self.assertEqual(api_flavor['ram'], db_flavor.ram)
self.assertEqual(api_flavor['SNF:disk_template'],
db_flavor.disk_template)
def test_deleted_flavor_details(self):
"""Test that API returns details for deleted flavors"""
flavor = self.flavor2
response = self.get('/api/v1.1/flavors/%d' % flavor.id)
self.assertSuccess(response)
api_flavor = json.loads(response.content)['flavor']
self.assertEquals(api_flavor['name'], flavor.name)
def test_deleted_flavors_list(self):
"""Test that deleted flavors do not appear to flavors list"""
response = self.get('/api/v1.1/flavors')
self.assertSuccess(response)
api_flavors = json.loads(response.content)['flavors']['values']
self.assertEqual(len(api_flavors), 2)
def test_deleted_flavors_details(self):
"""Test that deleted flavors do not appear to flavors detail list"""
FlavorFactory(deleted=True)
response = self.get('/api/v1.1/flavors/detail')
self.assertSuccess(response)
api_flavors = json.loads(response.content)['flavors']['values']
self.assertEqual(len(api_flavors), 2)
def test_wrong_flavor(self):
"""Test 404 result when requesting a flavor that does not exist."""
response = self.get('/api/v1.1/flavors/%d' % 22)
self.assertItemNotFound(response)
# Copyright 2012 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
import json
from synnefo.api.tests import BaseAPITest
from synnefo.api.faults import ItemNotFound
from mock import patch
from functools import wraps
def assert_backend_closed(func):
"""Decorator for ensuring that ImageBackend is returned to pool."""
@wraps(func)
def wrapper(self, backend):
result = func(self, backend)
if backend.called is True:
backend.return_value.close.assert_called_once_with()
return result
return wrapper
@patch('synnefo.api.images.ImageBackend')
class ImageAPITest(BaseAPITest):
@assert_backend_closed
def test_create_image(self, mimage):
"""Test that create image is not implemented"""
response = self.post('/api/v1.1/images/', 'user', json.dumps(''),
'json')
self.assertEqual(response.status_code, 503)
@assert_backend_closed
def test_list_images(self, mimage):
"""Test that expected list of images is returned"""
images = [{'id': 1, 'name': 'image-1'},
{'id': 2, 'name': 'image-2'},
{'id': 3, 'name': 'image-3'}]
mimage().list.return_value = images
response = self.get('/api/v1.1/images/', 'user')
self.assertSuccess(response)
api_images = json.loads(response.content)['images']['values']
self.assertEqual(images, api_images)
@assert_backend_closed
def test_list_images_detail(self, mimage):
images = [{'id': 1,
'name': 'image-1',
'status':'available',
'created_at': '2012-11-26 11:52:54',
'updated_at': '2012-12-26 11:52:54',
'deleted_at': '',
'properties': {'foo':'bar'}},
{'id': 2,
'name': 'image-2',
'status': 'deleted',
'created_at': '2012-11-26 11:52:54',
'updated_at': '2012-12-26 11:52:54',
'deleted_at': '2012-12-27 11:52:54',
'properties': ''},
{'id': 3,
'name': 'image-3',
'status': 'available',
'created_at': '2012-11-26 11:52:54',
'deleted_at': '',
'updated_at': '2012-12-26 11:52:54',
'properties': ''}]
result_images = [
{'id': 1,
'name': 'image-1',
'status':'ACTIVE',
'progress': 100,
'created': '2012-11-26T11:52:54+00:00',
'updated': '2012-12-26T11:52:54+00:00',
'metadata': {'values': {'foo':'bar'}}},
{'id': 2,
'name': 'image-2',
'status': 'DELETED',
'progress': 0,
'created': '2012-11-26T11:52:54+00:00',
'updated': '2012-12-26T11:52:54+00:00'},
{'id': 3,
'name': 'image-3',
'status': 'ACTIVE',
'progress': 100,
'created': '2012-11-26T11:52:54+00:00',
'updated': '2012-12-26T11:52:54+00:00'}]
mimage().list.return_value = images
response = self.get('/api/v1.1/images/detail', 'user')
self.assertSuccess(response)
api_images = json.loads(response.content)['images']['values']
self.assertEqual(len(result_images), len(api_images))
self.assertEqual(result_images, api_images)
@assert_backend_closed
def test_list_images_detail_since(self, mimage):
from datetime import datetime, timedelta
from time import sleep
old_time = datetime.now()
new_time = old_time + timedelta(seconds=0.1)
sleep(0.1)
images = [
{'id': 1,
'name': 'image-1',
'status':'available',
'progress': 100,
'created_at': old_time.isoformat(),
'deleted_at': '',
'updated_at': old_time.isoformat(),
'properties': ''},
{'id': 2,
'name': 'image-2',
'status': 'deleted',
'progress': 0,
'created_at': new_time.isoformat(),
'updated_at': new_time.isoformat(),
'deleted_at': new_time.isoformat(),
'properties': ''}]
mimage().iter.return_value = images
response =\
self.get('/api/v1.1/images/detail?changes-since=%sUTC' % new_time)
self.assertSuccess(response)
api_images = json.loads(response.content)['images']['values']
self.assertEqual(1, len(api_images))
@assert_backend_closed
def test_get_image_details(self, mimage):
image = {'id': 42,
'name': 'image-1',
'status': 'available',
'created_at': '2012-11-26 11:52:54',
'updated_at': '2012-12-26 11:52:54',
'deleted_at': '',
'properties': {'foo': 'bar'}}
result_image = \
{'id': 42,
'name': 'image-1',
'status': 'ACTIVE',
'progress': 100,
'created': '2012-11-26T11:52:54+00:00',
'updated': '2012-12-26T11:52:54+00:00',
'metadata': {'values': {'foo': 'bar'}}}
with patch('synnefo.api.util.get_image') as m:
m.return_value = image
response = self.get('/api/v1.1/images/42', 'user')
self.assertSuccess(response)
api_image = json.loads(response.content)['image']
self.assertEqual(api_image, result_image)
@assert_backend_closed
def test_invalid_image(self, mimage):
with patch('synnefo.api.util.get_image') as m:
m.side_effect = ItemNotFound('Image not found')
response = self.get('/api/v1.1/images/42', 'user')
self.assertItemNotFound(response)
def test_delete_image(self, mimage):
response = self.delete("/api/v1.1/images/42", "user")
self.assertEqual(response.status_code, 204)
mimage.return_value.delete.assert_called_once_with('42')
@patch('synnefo.api.util.ImageBackend')
class ImageMetadataAPITest(BaseAPITest):
def setUp(self):
self.image = {'id': 42,
'name': 'image-1',
'status': 'available',
'created_at': '2012-11-26 11:52:54',
'updated_at': '2012-12-26 11:52:54',
'deleted_at': '',
'properties': {'foo': 'bar', 'foo2': 'bar2'}}
self.result_image = \
{'id': 42,
'name': 'image-1',
'status': 'ACTIVE',
'progress': 100,
'created': '2012-11-26T11:52:54+00:00',
'updated': '2012-12-26T11:52:54+00:00',
'metadata': {'values': {'foo': 'bar'}}}
@assert_backend_closed
def test_list_metadata(self, backend):
backend.return_value.get_image.return_value = self.image
response = self.get('/api/v1.1/images/42/meta', 'user')
self.assertSuccess(response)
meta = json.loads(response.content)['metadata']['values']
self.assertEqual(meta, self.image['properties'])
@assert_backend_closed
def test_get_metadata(self, backend):
backend.return_value.get_image.return_value = self.image
response = self.get('/api/v1.1/images/42/meta/foo', 'user')
self.assertSuccess(response)
meta = json.loads(response.content)['meta']
self.assertEqual(meta['foo'], 'bar')
@assert_backend_closed
def test_get_invalid_metadata(self, backend):
backend.return_value.get_image.return_value = self.image
response = self.get('/api/v1.1/images/42/meta/not_found', 'user')
self.assertItemNotFound(response)
@assert_backend_closed
def test_delete_metadata_item(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend") as m:
response = self.delete('/api/v1.1/images/42/meta/foo', 'user')
self.assertEqual(response.status_code, 204)
m.return_value.update.assert_called_once_with('42',
{'properties': {'foo2': 'bar2'}})
@assert_backend_closed
def test_create_metadata_item(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend") as m:
request = {'meta': {'foo3': 'bar3'}}
response = self.put('/api/v1.1/images/42/meta/foo3', 'user',
json.dumps(request), 'json')
self.assertEqual(response.status_code, 201)
m.return_value.update.assert_called_once_with('42',
{'properties':
{'foo': 'bar', 'foo2': 'bar2', 'foo3': 'bar3'}})
@assert_backend_closed
def test_create_metadata_malformed_1(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend"):
request = {'met': {'foo3': 'bar3'}}
response = self.put('/api/v1.1/images/42/meta/foo3', 'user',
json.dumps(request), 'json')
self.assertBadRequest(response)
@assert_backend_closed
def test_create_metadata_malformed_2(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend"):
request = {'meta': [('foo3', 'bar3')]}
response = self.put('/api/v1.1/images/42/meta/foo3', 'user',
json.dumps(request), 'json')
self.assertBadRequest(response)
@assert_backend_closed
def test_create_metadata_malformed_3(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend"):
request = {'met': {'foo3': 'bar3', 'foo4': 'bar4'}}
response = self.put('/api/v1.1/images/42/meta/foo3', 'user',
json.dumps(request), 'json')
self.assertBadRequest(response)
@assert_backend_closed
def test_create_metadata_malformed_4(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend"):
request = {'met': {'foo3': 'bar3'}}
response = self.put('/api/v1.1/images/42/meta/foo4', 'user',
json.dumps(request), 'json')
self.assertBadRequest(response)
@assert_backend_closed
def test_update_metadata_item(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend") as m:
request = {'metadata': {'foo': 'bar_new', 'foo4': 'bar4'}}
response = self.post('/api/v1.1/images/42/meta', 'user',
json.dumps(request), 'json')
self.assertEqual(response.status_code, 201)
m.return_value.update.assert_called_once_with('42',
{'properties':
{'foo': 'bar_new', 'foo2': 'bar2', 'foo4': 'bar4'}
})
@assert_backend_closed
def test_update_metadata_malformed(self, backend):
backend.return_value.get_image.return_value = self.image
with patch("synnefo.api.images.ImageBackend"):
request = {'meta': {'foo': 'bar_new', 'foo4': 'bar4'}}
response = self.post('/api/v1.1/images/42/meta', 'user',
json.dumps(request), 'json')
self.assertBadRequest(response)
# Copyright 2012 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
import json
from mock import patch
from synnefo.api.tests import BaseAPITest
from synnefo.db.models import Network, NetworkInterface
from synnefo.db import models_factory as mfactory
@patch('synnefo.logic.rapi_pool.GanetiRapiClient')
class NetworkAPITest(BaseAPITest):
def setUp(self):
self.mac_prefixes = mfactory.MacPrefixPoolTableFactory()
self.bridges = mfactory.BridgePoolTableFactory()
self.user = 'dummy-user'
self.net1 = mfactory.NetworkFactory(userid=self.user)
self.vm1 = mfactory.VirtualMachineFactory(userid=self.user)
self.nic1 = mfactory.NetworkInterfaceFactory(network=self.net1,
machine=self.vm1)
self.nic2 = mfactory.NetworkInterfaceFactory(network=self.net1,
machine=self.vm1)
self.net2 = mfactory.NetworkFactory(userid=self.user)
self.nic3 = mfactory.NetworkInterfaceFactory(network=self.net2)
def assertNetworksEqual(self, db_net, api_net, detail=False):
self.assertEqual(str(db_net.id), api_net["id"])
self.assertEqual(db_net.name, api_net['name'])
if detail:
self.assertEqual(db_net.state, api_net['status'])
self.assertEqual(db_net.flavor, api_net['type'])
self.assertEqual(db_net.subnet, api_net['cidr'])
self.assertEqual(db_net.subnet6, api_net['cidr6'])
self.assertEqual(db_net.gateway, api_net['gateway'])
self.assertEqual(db_net.gateway6, api_net['gateway6'])
self.assertEqual(db_net.dhcp, api_net['dhcp'])
self.assertEqual(db_net.public, api_net['public'])
db_nics = ["nic-%d-%d" % (nic.machine.id, nic.index) for nic in
db_net.nics.filter(machine__userid=db_net.userid)]
self.assertEqual(db_nics, api_net['attachments']['values'])
def test_create_network_1(self, mrapi):
request = {
'network': {'name': 'foo'}
}
response = self.post('/api/v1.1/networks/', 'user1',
json.dumps(request), 'json')
self.assertEqual(response.status_code, 202)
db_networks = Network.objects.filter(userid='user1')
self.assertEqual(len(db_networks), 1)
db_net = db_networks[0]
api_net = json.loads(response.content)['network']
self.assertNetworksEqual(db_net, api_net)
mrapi.CreateNetwork.assert_called()
mrapi.ConnectNetwork.assert_called()
def test_invalid_data_1(self, mrapi):
"""Test invalid flavor"""
request = {
'network': {'name': 'foo', 'type': 'LoLo'}
}
response = self.post('/api/v1.1/networks/', 'user1',
json.dumps(request), 'json')
self.assertBadRequest(response)
self.assertEqual(len(Network.objects.filter(userid='user1')), 0)
def test_invalid_data_2(self, mrapi):
"""Test invalid subnet"""
request = {
'network': {'name': 'foo', 'cidr': '10.0.0.0/8'}
}
response = self.post('/api/v1.1/networks/', 'user1',
json.dumps(request), 'json')
self.assertFault(response, 413, "overLimit")
def test_invalid_data_3(self, mrapi):
"""Test unauthorized to create public network"""
request = {
'network': {'name': 'foo', 'public': True}
}
response = self.post('/api/v1.1/networks/', 'user1',
json.dumps(request), 'json')
self.assertFault(response, 403, "forbidden")
def test_invalid_data_4(self, mrapi):
"""Test unauthorized to create network not in settings"""
request = {
'network': {'name': 'foo', 'type': 'CUSTOM'}
}
response = self.post('/api/v1.1/networks/', 'user1',
json.dumps(request), 'json')
self.assertFault(response, 403, "forbidden")
def test_list_networks(self, mrapi):
"""Test that expected list of networks is returned."""
# Create a deleted network
mfactory.NetworkFactory(userid=self.user, deleted=True)
response = self.get('/api/v1.1/networks/', self.user)
self.assertSuccess(response)
db_nets = Network.objects.filter(userid=self.user, deleted=False)
api_nets = json.loads(response.content)["networks"]["values"]
self.assertEqual(len(db_nets), len(api_nets))
for api_net in api_nets:
net_id = api_net['id']
self.assertNetworksEqual(Network.objects.get(id=net_id), api_net)
def test_list_networks_detail(self, mrapi):
"""Test that expected networks details are returned."""
# Create a deleted network
mfactory.NetworkFactory(userid=self.user, deleted=True)
response = self.get('/api/v1.1/networks/detail', self.user)
self.assertSuccess(response)