Commit 04deb668 authored by Christos Stavrakakis's avatar Christos Stavrakakis

Add more tests for db, logic and api

parent 2c37fb9d
......@@ -249,6 +249,9 @@ class ServerAPITest(BaseAPITest):
"""Test if a server details are returned."""
db_vm = self.vm2
user = self.vm2.userid
net = mfactory.NetworkFactory()
nic = mfactory.NetworkInterfaceFactory(machine=self.vm2, network=net)
db_vm_meta = mfactory.VirtualMachineMetadataFactory(vm=db_vm)
response = self.get('/api/v1.1/servers/%d' % db_vm.id, user)
......@@ -260,6 +263,13 @@ class ServerAPITest(BaseAPITest):
self.assertEqual(server['imageRef'], db_vm.imageid)
self.assertEqual(server['name'], db_vm.name)
self.assertEqual(server['status'], get_rsapi_state(db_vm))
api_nic = server['attachments']['values'][0]
self.assertEqual(api_nic['network_id'], str(net.id))
self.assertEqual(api_nic['mac_address'], nic.mac)
self.assertEqual(api_nic['firewallProfile'], nic.firewall_profile)
self.assertEqual(api_nic['ipv4'], nic.ipv4)
self.assertEqual(api_nic['ipv6'], nic.ipv6)
self.assertEqual(api_nic['id'], 'nic-%s-%s' % (db_vm.id, nic.index))
metadata = server['metadata']['values']
self.assertEqual(len(metadata), 1)
......@@ -285,6 +295,14 @@ class ServerAPITest(BaseAPITest):
response = self.post('/api/v1.1/servers', params={})
self.assertBadRequest(response)
def test_rename_server(self):
vm = self.vm2
request = {'server': {'name': 'new_name'}}
response = self.put('/api/v1.1/servers/%d' % vm.id, vm.userid,
json.dumps(request), 'json')
self.assertSuccess(response)
self.assertEqual(VirtualMachine.objects.get(id=vm.id).name, "new_name")
@patch('synnefo.api.util.get_image')
@patch('synnefo.logic.rapi_pool.GanetiRapiClient')
......@@ -359,7 +377,7 @@ class ServerDestroyAPITest(BaseAPITest):
vm = mfactory.VirtualMachineFactory()
response = self.delete('/api/v1.1/servers/%d' % 42, vm.userid)
self.assertItemNotFound(response)
mrapi().DeleteInstance.assert_not_called()
self.assertFalse(mrapi.mock_calls)
@patch('synnefo.api.util.get_image')
......@@ -383,6 +401,23 @@ class ServerActionAPITest(BaseAPITest):
self.assertEqual(VirtualMachine.objects.get(id=vm.id).action,
action.upper())
def test_action_in_building_vm(self, mrapi, mimage):
"""Test building in progress"""
vm = mfactory.VirtualMachineFactory()
request = {'start': '{}'}
response = self.post('/api/v1.1/servers/%d/action' % vm.id,
vm.userid, json.dumps(request), 'json')
self.assertEqual(response.status_code, 409)
self.assertFalse(mrapi.mock_calls)
def test_destroy_build_vm(self, mrapi, mimage):
"""Test building in progress"""
vm = mfactory.VirtualMachineFactory()
response = self.delete('/api/v1.1/servers/%d' % vm.id,
vm.userid)
self.assertSuccess(response)
mrapi().RemoveInstance.assert_called_once()
def test_firewall(self, mrapi, mimage):
vm = mfactory.VirtualMachineFactory()
vm.operstate = "STOPPED"
......@@ -393,6 +428,16 @@ class ServerActionAPITest(BaseAPITest):
self.assertEqual(response.status_code, 202)
mrapi().ModifyInstance.assert_called_once()
def test_unsupported_firewall(self, mrapi, mimage):
vm = mfactory.VirtualMachineFactory()
vm.operstate = "STOPPED"
vm.save()
request = {'firewallProfile': {'profile': 'FOO'}}
response = self.post('/api/v1.1/servers/%d/action' % vm.id,
vm.userid, json.dumps(request), 'json')
self.assertBadRequest(response)
self.assertFalse(mrapi.mock_calls)
class ServerMetadataAPITest(BaseAPITest):
def setUp(self):
......@@ -590,7 +635,16 @@ class NetworkAPITest(BaseAPITest):
response = self.delete('/api/v1.1/networks/%d' % net.id,
self.net2.userid)
self.assertFault(response, 403, 'forbidden')
mrapi.DeleteNetwork.assert_not_called()
self.assertFalse(mrapi.mock_calls)
def test_delete_network_in_use(self, mrapi):
net = mfactory.NetworkFactory(deleted=False)
vm = mfactory.VirtualMachineFactory()
mfactory.NetworkInterfaceFactory(machine=vm, network=net)
response = self.delete('/api/v1.1/networks/%d' % net.id,
net.userid)
self.assertFault(response, 421, 'networkInUse')
self.assertFalse(mrapi.mock_calls)
def test_add_nic(self, mrapi):
user = 'userr'
......@@ -601,6 +655,15 @@ class NetworkAPITest(BaseAPITest):
net.userid, json.dumps(request), 'json')
self.assertEqual(response.status_code, 202)
def test_add_nic_to_public_network(self, mrapi):
user = 'userr'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
net = mfactory.NetworkFactory(state='ACTIVE', userid=user, public=True)
request = {'add': {'serverRef': vm.id}}
response = self.post('/api/v1.1/networks/%d/action' % net.id,
net.userid, json.dumps(request), 'json')
self.assertFault(response, 403, 'forbidden')
def test_add_nic_malformed(self, mrapi):
user = 'userr'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
......@@ -610,6 +673,36 @@ class NetworkAPITest(BaseAPITest):
net.userid, json.dumps(request), 'json')
self.assertBadRequest(response)
def test_add_nic_not_active(self, mrapi):
"""Test connecting VM to non-active network"""
user = 'dummy'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
net = mfactory.NetworkFactory(state='PENDING', subnet='10.0.0.0/31',
userid=user)
request = {'add': {'serveRef': vm.id}}
response = self.post('/api/v1.1/networks/%d/action' % net.id,
net.userid, json.dumps(request), 'json')
# Test that returns BuildInProgress
self.assertEqual(response.status_code, 409)
def test_add_nic_full_network(self, mrapi):
"""Test connecting VM to a full network"""
user = 'userr'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
net = mfactory.NetworkFactory(state='ACTIVE', subnet='10.0.0.0/30',
userid=user, dhcp=True)
pool = net.get_pool()
while not pool.empty():
pool.get()
pool.save()
pool = net.get_pool()
self.assertTrue(pool.empty())
request = {'add': {'serverRef': vm.id}}
response = self.post('/api/v1.1/networks/%d/action' % net.id,
net.userid, json.dumps(request), 'json')
# Test that returns OverLimit
self.assertEqual(response.status_code, 413)
def test_remove_nic(self, mrapi):
user = 'userr'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
......@@ -637,6 +730,17 @@ class NetworkAPITest(BaseAPITest):
net.userid, json.dumps(request), 'json')
self.assertBadRequest(response)
def test_remove_nic_malformed_2(self, mrapi):
user = 'userr'
vm = mfactory.VirtualMachineFactory(name='yo', userid=user)
net = mfactory.NetworkFactory(state='ACTIVE', userid=user)
request = {'remove':
{'attachment': 'nic-%s' % vm.id}
}
response = self.post('/api/v1.1/networks/%d/action' % net.id,
net.userid, json.dumps(request), 'json')
self.assertBadRequest(response)
class ServerVNCConsole(BaseAPITest):
......@@ -665,6 +769,16 @@ class ServerVNCConsole(BaseAPITest):
self.assertEqual(set(console.keys()),
set(['type', 'host', 'port', 'password']))
def test_wrong_console_type(self):
"""Test console req for ACTIVE server"""
vm = mfactory.VirtualMachineFactory()
vm.operstate = 'STARTED'
vm.save()
data = json.dumps({'console': {'type': 'foo'}})
response = self.post('/api/v1.1/servers/%d/action' % vm.id,
vm.userid, data, 'json')
self.assertBadRequest(response)
def assert_backend_closed(func):
@wraps(func)
......@@ -845,6 +959,42 @@ class ImageMetadataAPITest(BaseAPITest):
{'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
......@@ -858,6 +1008,15 @@ class ImageMetadataAPITest(BaseAPITest):
{'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)
class APITest(TestCase):
def test_api_version(self):
......
......@@ -34,7 +34,8 @@
from django.test import TestCase
from synnefo.db.pools import (PoolManager, EmptyPool, BridgePool,
MacPrefixPool, IPPool)
MacPrefixPool, IPPool, find_padding,
bitarray_to_map)
from bitarray import bitarray
......@@ -136,6 +137,37 @@ class PoolManagerTestCase(TestCase):
self.assertEqual(pool.available, bitarray('1' * 39 + '0' * 1))
self.assertEqual(pool.reserved, bitarray('1' * 39 + '0' * 1))
def test_shrink_in_use(self):
obj = DummyObject(8)
pool = DummyPool(obj)
pool._reserve(6)
self.assertRaises(Exception, pool.shrink, 3)
def test_count(self):
obj = DummyObject(10)
pool = DummyPool(obj)
pool._reserve(1)
pool._reserve(3)
pool._reserve(4)
pool._reserve(2, external=True)
self.assertEqual(pool.count_available(), 6)
self.assertEqual(pool.count_unavailable(), 4)
self.assertEqual(pool.count_reserved(), 1)
self.assertEqual(pool.count_unreserved(), 9)
class HelpersTestCase(TestCase):
def test_find_padding(self):
self.assertEqual(find_padding(1), 7)
self.assertEqual(find_padding(8), 0)
self.assertEqual(find_padding(12), 4)
self.assertEqual(find_padding(16), 0)
def test_bitarray_to_map(self):
bt = bitarray('01001100101')
map_ = bitarray_to_map(bt)
self.assertEqual(map_, 'X.XX..XX.X.')
class BridgePoolTestCase(TestCase):
def test_bridge_conversion(self):
......@@ -159,6 +191,21 @@ class MacPrefixPoolTestCase(TestCase):
for i in range(0, 65536):
self.assertEqual(pool.is_available(i, index=True), False)
def test_mac_prefix_conversion(self):
obj = DummyObject(13)
obj.base = 'aa:00:0'
pool = MacPrefixPool(obj)
for i in range(1, 9):
self.assertEqual("aa:00:%s" % i, pool.get())
def test_value_to_index(self):
obj = DummyObject(13)
obj.base = 'aa:00:0'
pool = MacPrefixPool(obj)
index = pool.value_to_index('aa:bc:ee')
val = pool.index_to_value(index)
self.assertEqual(val, 'aa:bc:ee')
class IPPoolTestCase(TestCase):
def test_auto_reservations(self):
......
......@@ -192,6 +192,12 @@ class BackendNetworkTest(TestCase):
self.assertTrue(backend.index < 10)
self.assertEqual(bnet.mac_prefix, 'aa:bb:c%s' % backend.index)
def test_invalid_mac(self):
network = mfact.NetworkFactory(mac_prefix='zz:bb:c')
backend = mfact.BackendFactory()
self.assertRaises(utils.InvalidMacAddress,
mfact.BackendNetworkFactory, network=network, backend=backend)
class BridgePoolTest(TestCase):
def test_no_pool(self):
......
# Copyright 2011-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.
from django.test import TestCase
from synnefo.logic import rapi_pool
from mock import patch
@patch('synnefo.logic.rapi_pool.GanetiRapiClient', spec=True)
class GanetiRapiPoolTest(TestCase):
def test_new_client(self, rclient):
cl = rapi_pool.get_rapi_client(1, 'amxixa', 'cluster0', '5080', 'user',
'pass')
rclient.assert_called_once_with("cluster0", "5080", "user", "pass")
self.assertTrue('amxixa' in rapi_pool._pools)
self.assertTrue(cl._pool is rapi_pool._pools[rapi_pool._hashes[1]])
def test_invalid_get(self, rclient):
self.assertRaises(ValueError, rapi_pool.get_rapi_client, 1, 'amxixa',
None, '5080', 'user', 'pass')
self.assertRaises(ValueError, rapi_pool.get_rapi_client, 1, 'amxixa',
'Foo', None, 'user', 'pass')
def test_get_from_pool(self, rclient):
cl = rapi_pool.get_rapi_client(1, 'dummyhash', 'cluster1', '5080',
'user', 'pass')
rclient.assert_called_once_with("cluster1", "5080", "user", "pass")
rapi_pool.put_rapi_client(cl)
rclient.reset_mock()
cl2 = rapi_pool.get_rapi_client(1, 'dummyhash', 'cluster1', '5080',
'user', 'pass')
self.assertTrue(cl is cl2)
self.assertFalse(rclient.mock_calls)
def test_changed_credentials(self, rclient):
cl = rapi_pool.get_rapi_client(1, 'dummyhash2', 'cluster2', '5080',
'user', 'pass')
rclient.assert_called_once_with("cluster2", "5080", "user", "pass")
rapi_pool.put_rapi_client(cl)
rclient.reset_mock()
rapi_pool.get_rapi_client(1, 'dummyhash3', 'cluster2', '5080',
'user', 'new_pass')
rclient.assert_called_once_with("cluster2", "5080", "user", "new_pass")
self.assertFalse('dummyhash2' in rapi_pool._pools)
def test_no_pool(self, rclient):
cl = rapi_pool.get_rapi_client(1, 'dummyhash2', 'cluster2', '5080',
'user', 'pass')
cl._pool = None
rapi_pool.put_rapi_client(cl)
self.assertTrue(cl not in rapi_pool._pools.values())
# Copyright 2011-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.
from django.test import TestCase
from synnefo.logic import utils
from django.conf import settings
from synnefo.db.models import VirtualMachine, Network
from synnefo.db.models_factory import VirtualMachineFactory
class NameConversionTest(TestCase):
def setUp(self):
settings.BACKEND_PREFIX_ID = 'snf-'
def test_id_from_iname(self):
self.assertEqual(utils.id_from_instance_name('snf-42'), 42)
for name in [None, 'foo', 'snf42', 'snf-a', 'snf-snf-42', 1234]:
self.assertRaises(VirtualMachine.InvalidBackendIdError,
utils.id_from_instance_name, '')
def test_iname_from_id(self):
self.assertEqual(utils.id_to_instance_name(42), 'snf-42')
def test_id_from_net_name(self):
self.assertEqual(utils.id_from_network_name('snf-net-42'), 42)
for name in [None, 'foo', 'snf42', 'snf-net-a', 'snf-snf-42', 1234]:
self.assertRaises(Network.InvalidBackendIdError,
utils.id_from_network_name, '')
def test_net_name_from_id(self):
self.assertEqual(utils.id_to_network_name(42), 'snf-net-42')
class APIStateTest(TestCase):
def test_correct_state(self):
vm = VirtualMachineFactory()
vm.operstate = 'foo'
self.assertEqual(utils.get_rsapi_state(vm), "UNKNOWN")
vm.operstate = "STARTED"
vm.deleted = True
self.assertEqual(utils.get_rsapi_state(vm), "DELETED")
vm.deleted = False
vm.backendopcode = "OP_INSTANCE_REBOOT"
vm.backendjobstatus = "waiting"
self.assertEqual(utils.get_rsapi_state(vm), "REBOOT")
class HidePass(TestCase):
def test_no_osparams(self):
foo = {'foo': 'bar'}
self.assertTrue(foo is utils.hide_pass(foo))
foo = {'osparams': {}, 'bar': 'foo'}
self.assertTrue(foo is utils.hide_pass(foo))
def test_hidden_pass(self):
foo = {'osparams': {'img_passwd': 'pass'}, 'bar': 'foo'}
after = {'osparams': {'img_passwd': 'xxxxxxxx'}, 'bar': 'foo'}
self.assertEqual(after, utils.hide_pass(foo))
......@@ -84,6 +84,24 @@ class UpdateDBTest(TestCase):
update_db(client, msg)
client.basic_ack.assert_called_once()
def test_old_msg(self, client):
from time import sleep
from datetime import datetime
old_time = time()
sleep(0.01)
new_time = datetime.fromtimestamp(time())
vm = mfactory.VirtualMachineFactory(backendtime=new_time)
vm.operstate = 'STOPPED'
vm.save()
msg = self.create_msg(operation='OP_INSTANCE_STARTUP',
event_time=split_time(old_time),
instance=vm.backend_vm_id)
update_db(client, msg)
client.basic_ack.assert_called_once()
db_vm = VirtualMachine.objects.get(id=vm.id)
self.assertEquals(db_vm.operstate, "STOPPED")
self.assertEquals(db_vm.backendtime, new_time)
def test_start(self, client):
vm = mfactory.VirtualMachineFactory()
msg = self.create_msg(operation='OP_INSTANCE_STARTUP',
......@@ -541,3 +559,6 @@ class ReconciliationTest(TestCase):
set([(2, 'STARTED', False),
(3, 'BUILD', True), (4, 'STARTED', False),
(50, 'BUILD', True)]))
from synnefo.logic.test.rapi_pool_tests import *
from synnefo.logic.test.utils_tests import *
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment