Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / db / tests.py @ 1cfd5d4d

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