root / snf-cyclades-app / synnefo / db / tests.py @ a633e247
History | View | Annotate | Download (11.1 kB)
1 |
# Copyright 2012 GRNET S.A. All rights reserved.
|
---|---|
2 |
#
|
3 |
# Redistribution and use in source and binary forms, with or without
|
4 |
# modification, are permitted provided that the following conditions
|
5 |
# are met:
|
6 |
#
|
7 |
# 1. Redistributions of source code must retain the above copyright
|
8 |
# notice, this list of conditions and the following disclaimer.
|
9 |
#
|
10 |
# 2. Redistributions in binary form must reproduce the above copyright
|
11 |
# notice, this list of conditions and the following disclaimer in the
|
12 |
# documentation and/or other materials provided with the distribution.
|
13 |
#
|
14 |
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
|
15 |
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
16 |
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
17 |
# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
|
18 |
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
|
19 |
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
|
20 |
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
|
21 |
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
22 |
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
|
23 |
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
|
24 |
# SUCH DAMAGE.
|
25 |
#
|
26 |
# The views and conclusions contained in the software and documentation are
|
27 |
# those of the authors and should not be interpreted as representing official
|
28 |
# policies, either expressed or implied, of GRNET S.A.
|
29 |
|
30 |
# Unit Tests for db
|
31 |
#
|
32 |
# Provides automated tests for db module
|
33 |
|
34 |
from django.test import TestCase |
35 |
|
36 |
from synnefo import settings |
37 |
# Import pool tests
|
38 |
from synnefo.db.pools.tests import * |
39 |
|
40 |
from synnefo.db.models import * |
41 |
from synnefo.db import models_factory as mfact |
42 |
from synnefo.db.pools import IPPool, EmptyPool |
43 |
|
44 |
from django.db import IntegrityError |
45 |
from django.core.exceptions import MultipleObjectsReturned |
46 |
from snf_django.utils.testing import override_settings |
47 |
from mock import patch |
48 |
|
49 |
|
50 |
class FlavorTest(TestCase): |
51 |
def test_flavor_name(self): |
52 |
"""Test a flavor object name method."""
|
53 |
flavor = mfact.FlavorFactory(cpu=1, ram=1024, disk=40, |
54 |
disk_template="temp")
|
55 |
self.assertEqual(flavor.name, "C1R1024D40temp", "flavor.name is not" |
56 |
" generated correctly. Name is %s instead of C1R1024D40temp" %
|
57 |
flavor.name) |
58 |
|
59 |
|
60 |
class BackendTest(TestCase): |
61 |
def setUp(self): |
62 |
self.backend = mfact.BackendFactory()
|
63 |
|
64 |
@patch("synnefo.db.models.get_rapi_client") |
65 |
def test_get_client(self, client): |
66 |
id_ = self.backend.id
|
67 |
hash_ = self.backend.hash
|
68 |
name = self.backend.clustername
|
69 |
passwd = self.backend.password
|
70 |
user = self.backend.username
|
71 |
port = self.backend.port
|
72 |
self.backend.get_client()
|
73 |
client.assert_called_once_with(id_, hash_, name, port, user, passwd) |
74 |
|
75 |
def test_save_hash(self): |
76 |
"""Test that backend hash is generated on credential change"""
|
77 |
old_hash = self.backend.hash
|
78 |
for field in ['clustername', 'username', 'password', 'port']: |
79 |
value = 5181 if field == 'port' else 'foo' |
80 |
self.backend.__setattr__(field, value)
|
81 |
self.backend.save()
|
82 |
self.assertNotEqual(old_hash, self.backend.hash) |
83 |
old_hash = self.backend.hash
|
84 |
|
85 |
def test_unique_index(self): |
86 |
"""Test that each backend gets a unique index"""
|
87 |
backends = [self.backend]
|
88 |
for i in xrange(0, 14): |
89 |
backends.append(mfact.BackendFactory()) |
90 |
indexes = map(lambda x: x.index, backends) |
91 |
self.assertEqual(len(indexes), len(set(indexes))) |
92 |
|
93 |
def test_backend_number(self): |
94 |
"""Test that no more than 16 backends are created"""
|
95 |
for i in xrange(0, 14): |
96 |
mfact.BackendFactory() |
97 |
self.assertRaises(Exception, mfact.BackendFactory, ()) |
98 |
|
99 |
def test_delete_backend(self): |
100 |
vm = mfact.VirtualMachineFactory(backend=self.backend, deleted=True) |
101 |
bnet = mfact.BackendNetworkFactory(backend=self.backend)
|
102 |
self.backend.delete()
|
103 |
self.assertRaises(Backend.DoesNotExist, Backend.objects.get,
|
104 |
id=self.backend.id)
|
105 |
# Test that VM is not deleted
|
106 |
vm2 = VirtualMachine.objects.get(id=vm.id) |
107 |
self.assertEqual(vm2.backend, None) |
108 |
# Test tha backend networks are deleted, but not the network
|
109 |
self.assertRaises(BackendNetwork.DoesNotExist,
|
110 |
BackendNetwork.objects.get, id=bnet.id) |
111 |
Network.objects.get(id=bnet.network.id) |
112 |
|
113 |
def test_delete_active_backend(self): |
114 |
"""Test that a backend with non-deleted VMS is not deleted"""
|
115 |
mfact.VirtualMachineFactory(backend=self.backend)
|
116 |
self.assertRaises(IntegrityError, self.backend.delete, ()) |
117 |
|
118 |
def test_password_encryption(self): |
119 |
password_hash = self.backend.password
|
120 |
self.backend.password = '123' |
121 |
self.assertNotEqual(self.backend.password_hash, '123') |
122 |
self.assertNotEqual(self.backend.password_hash, password_hash) |
123 |
self.assertEqual(self.backend.password, '123') |
124 |
|
125 |
def test_hypervisor(self): |
126 |
from synnefo.db.models import snf_settings |
127 |
kvm_backend = mfact.BackendFactory(hypervisor="kvm")
|
128 |
xen_pvm_backend = mfact.BackendFactory(hypervisor="xen-pvm")
|
129 |
xen_hvm_backend = mfact.BackendFactory(hypervisor="xen-hvm")
|
130 |
with override_settings(snf_settings, GANETI_USE_HOTPLUG=True): |
131 |
self.assertTrue(kvm_backend.use_hotplug())
|
132 |
self.assertFalse(xen_pvm_backend.use_hotplug())
|
133 |
self.assertFalse(xen_hvm_backend.use_hotplug())
|
134 |
with override_settings(snf_settings, GANETI_USE_HOTPLUG=False): |
135 |
self.assertFalse(kvm_backend.use_hotplug())
|
136 |
self.assertFalse(xen_pvm_backend.use_hotplug())
|
137 |
self.assertFalse(xen_hvm_backend.use_hotplug())
|
138 |
kwargs = {"os": "snf-image+default", |
139 |
"hvparams": {"kvm": {"foo1": "mpaz1"}, |
140 |
"xen-pvm": {"foo2": "mpaz2"}, |
141 |
"xen-hvm": {"foo3": "mpaz3"}}} |
142 |
with override_settings(snf_settings,
|
143 |
GANETI_CREATEINSTANCE_KWARGS=kwargs): |
144 |
self.assertEqual(kvm_backend.get_create_params(),
|
145 |
{"os": "snf-image+default", |
146 |
"hvparams": {"foo1": "mpaz1"}}) |
147 |
self.assertEqual(xen_pvm_backend.get_create_params(),
|
148 |
{"os": "snf-image+default", |
149 |
"hvparams": {"foo2": "mpaz2"}}) |
150 |
self.assertEqual(xen_hvm_backend.get_create_params(),
|
151 |
{"os": "snf-image+default", |
152 |
"hvparams": {"foo3": "mpaz3"}}) |
153 |
with override_settings(snf_settings, GANETI_CREATEINSTANCE_KWARGS={}):
|
154 |
self.assertEqual(kvm_backend.get_create_params(), {"hvparams": {}}) |
155 |
|
156 |
|
157 |
class VirtualMachineTest(TestCase): |
158 |
def setUp(self): |
159 |
self.vm = mfact.VirtualMachineFactory()
|
160 |
|
161 |
@patch("synnefo.db.models.get_rapi_client") |
162 |
def test_get_client(self, client): |
163 |
backend = self.vm.backend
|
164 |
id_ = backend.id |
165 |
hash_ = backend.hash |
166 |
name = backend.clustername |
167 |
passwd = backend.password |
168 |
user = backend.username |
169 |
port = backend.port |
170 |
self.vm.get_client()
|
171 |
client.assert_called_once_with(id_, hash_, name, port, user, passwd) |
172 |
|
173 |
def test_create(self): |
174 |
vm = VirtualMachine() |
175 |
self.assertEqual(vm.action, None) |
176 |
self.assertEqual(vm.backendjobid, None) |
177 |
self.assertEqual(vm.backendjobstatus, None) |
178 |
self.assertEqual(vm.backendopcode, None) |
179 |
self.assertEqual(vm.backendlogmsg, None) |
180 |
self.assertEqual(vm.operstate, 'BUILD') |
181 |
|
182 |
|
183 |
class NetworkTest(TestCase): |
184 |
def setUp(self): |
185 |
self.net = mfact.NetworkFactory()
|
186 |
|
187 |
def test_tags(self): |
188 |
net1 = mfact.NetworkFactory(flavor='IP_LESS_ROUTED')
|
189 |
self.assertEqual(net1.backend_tag, ['ip-less-routed']) |
190 |
net1 = mfact.NetworkFactory(flavor='CUSTOM')
|
191 |
self.assertEqual(net1.backend_tag, [])
|
192 |
|
193 |
def test_create_backend_network(self): |
194 |
len_backends = len(Backend.objects.all())
|
195 |
back = mfact.BackendFactory() |
196 |
self.net.create_backend_network(backend=back)
|
197 |
BackendNetwork.objects.get(network=self.net, backend=back)
|
198 |
back1 = mfact.BackendFactory() |
199 |
back2 = mfact.BackendFactory() |
200 |
self.net.create_backend_network()
|
201 |
BackendNetwork.objects.get(network=self.net, backend=back1)
|
202 |
BackendNetwork.objects.get(network=self.net, backend=back2)
|
203 |
self.assertEqual(len(BackendNetwork.objects.filter(network=self.net)), |
204 |
len_backends + 3)
|
205 |
|
206 |
def test_pool(self): |
207 |
pool = self.net.get_pool()
|
208 |
pool.network = self.net
|
209 |
self.assertTrue(isinstance(pool, IPPool)) |
210 |
|
211 |
def test_reserve_ip(self): |
212 |
net1 = mfact.NetworkFactory(subnet='192.168.2.0/24')
|
213 |
net1.reserve_address('192.168.2.12')
|
214 |
pool = net1.get_pool() |
215 |
self.assertFalse(pool.is_available('192.168.2.12')) |
216 |
net1.release_address('192.168.2.12')
|
217 |
pool = net1.get_pool() |
218 |
self.assertTrue(pool.is_available('192.168.2.12')) |
219 |
|
220 |
|
221 |
class BackendNetworkTest(TestCase): |
222 |
def test_mac_prefix(self): |
223 |
network = mfact.NetworkFactory(mac_prefix='aa:bb:c')
|
224 |
backend = mfact.BackendFactory() |
225 |
bnet = mfact.BackendNetworkFactory(network=network, backend=backend) |
226 |
self.assertTrue(backend.index < 10) |
227 |
self.assertEqual(bnet.mac_prefix, 'aa:bb:c%s' % backend.index) |
228 |
|
229 |
def test_invalid_mac(self): |
230 |
network = mfact.NetworkFactory(mac_prefix='zz:bb:c')
|
231 |
backend = mfact.BackendFactory() |
232 |
self.assertRaises(utils.InvalidMacAddress,
|
233 |
mfact.BackendNetworkFactory, network=network, backend=backend) |
234 |
|
235 |
|
236 |
class BridgePoolTest(TestCase): |
237 |
def test_no_pool(self): |
238 |
self.assertRaises(EmptyPool,
|
239 |
BridgePoolTable.get_pool) |
240 |
|
241 |
def test_two_pools(self): |
242 |
mfact.BridgePoolTableFactory() |
243 |
mfact.BridgePoolTableFactory() |
244 |
self.assertRaises(MultipleObjectsReturned, BridgePoolTable.get_pool)
|
245 |
|
246 |
|
247 |
class AESTest(TestCase): |
248 |
def test_encrypt_decrtypt(self): |
249 |
from synnefo.db import aes_encrypt as aes |
250 |
old = 'bar'
|
251 |
new = aes.decrypt_db_charfield(aes.encrypt_db_charfield(old)) |
252 |
self.assertEqual(old, new)
|
253 |
|
254 |
def test_password_change(self): |
255 |
from synnefo.db import aes_encrypt as aes |
256 |
old_pass = aes.SECRET_ENCRYPTION_KEY |
257 |
old = 'bar'
|
258 |
encrypted = aes.encrypt_db_charfield(old) |
259 |
aes.SECRET_ENCRYPTION_KEY = 'foo2'
|
260 |
self.assertRaises(aes.CorruptedPassword, aes.decrypt_db_charfield,
|
261 |
encrypted) |
262 |
aes.SECRET_ENCRYPTION_KEY = old_pass |
263 |
new = aes.decrypt_db_charfield(encrypted) |
264 |
self.assertEqual(old, new)
|
265 |
|
266 |
def test_big_secret(self): |
267 |
from synnefo.db import aes_encrypt as aes |
268 |
old = aes.SECRET_ENCRYPTION_KEY |
269 |
aes.SECRET_ENCRYPTION_KEY = \ |
270 |
'91490231234814234812348913289481294812398421893489'
|
271 |
self.assertRaises(ValueError, aes.encrypt_db_charfield, 'la') |
272 |
aes.SECRET_ENCRYPTION_KEY = old |