Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / db / pools / tests.py @ b66e4833

History | View | Annotate | Download (8 kB)

1
# Copyright 2012 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

    
34

    
35
from django.test import TestCase
36
from synnefo.db.pools import (PoolManager, EmptyPool, BridgePool,
37
                              MacPrefixPool, IPPool, find_padding,
38
                              bitarray_to_map)
39
from bitarray import bitarray
40

    
41

    
42
class DummyObject():
43
    def __init__(self, size):
44
        self.size = size
45

    
46
        self.available_map = ''
47
        self.reserved_map = ''
48

    
49
    def save(self):
50
        pass
51

    
52

    
53
class DummyPool(PoolManager):
54
    def value_to_index(self, index):
55
        return index
56

    
57
    def index_to_value(self, value):
58
        return value
59

    
60

    
61
class PoolManagerTestCase(TestCase):
62
    def test_created_pool(self):
63
        obj = DummyObject(42)
64
        pool = DummyPool(obj)
65
        self.assertEqual(pool.to_01(), '1' * 42)
66
        self.assertEqual(pool.to_map(), '.' * 42)
67
        self.assertEqual(pool.available, bitarray('1' * 42 + '0' * 6))
68
        self.assertEqual(pool.reserved, bitarray('1' * 42 + '0' * 6))
69

    
70
    def test_save_pool(self):
71
        obj = DummyObject(42)
72
        pool = DummyPool(obj)
73
        pool.save()
74
        self.assertNotEqual(obj.available_map, '')
75
        available_map = obj.available_map
76
        b = DummyPool(obj)
77
        b.save()
78
        self.assertEqual(obj.available_map, available_map)
79

    
80
    def test_empty_pool(self):
81
        obj = DummyObject(42)
82
        pool = DummyPool(obj)
83
        self.assertEqual(pool.empty(), False)
84
        for i in range(0, 42):
85
            self.assertEqual(pool.get(), i)
86
        self.assertEqual(pool.empty(), True)
87
        self.assertRaises(EmptyPool, pool.get)
88

    
89
    def test_reserved_value(self):
90
        obj = DummyObject(42)
91
        pool = DummyPool(obj)
92
        available = pool.count_available()
93
        value = pool.get()
94
        self.assertEqual(pool.is_available(value), False)
95
        self.assertEqual(pool.count_available(), available - 1)
96
        pool.put(value)
97
        self.assertEqual(pool.is_available(value), True)
98
        self.assertEqual(pool.count_available(), available)
99

    
100
    def test_external_reserved(self):
101
        obj = DummyObject(42)
102
        pool = DummyPool(obj)
103
        for i in range(42, 48):
104
            self.assertEqual(pool.is_available(i), False)
105
        pool.reserve(32, external=True)
106
        values = []
107
        while True:
108
            try:
109
                values.append(pool.get())
110
            except EmptyPool:
111
                break
112
        self.assertEqual(32 not in values, True)
113

    
114
    def test_external_reserved_2(self):
115
        obj = DummyObject(42)
116
        pool = DummyPool(obj)
117
        self.assertEqual(pool.get(), 0)
118
        pool.reserve(0, external=True)
119
        pool.put(0)
120
        self.assertEqual(pool.get(), 1)
121

    
122
    def test_extend_pool(self):
123
        obj = DummyObject(42)
124
        pool = DummyPool(obj)
125
        pool.extend(7)
126
        self.assertEqual(pool.to_01(), '1' * 49)
127
        self.assertEqual(pool.to_map(), '.' * 49)
128
        self.assertEqual(pool.available, bitarray('1' * 49 + '0' * 7))
129
        self.assertEqual(pool.reserved, bitarray('1' * 49 + '0' * 7))
130

    
131
    def test_shrink_pool(self):
132
        obj = DummyObject(42)
133
        pool = DummyPool(obj)
134
        pool.shrink(3)
135
        self.assertEqual(pool.to_01(), '1' * 39)
136
        self.assertEqual(pool.to_map(), '.' * 39)
137
        self.assertEqual(pool.available, bitarray('1' * 39 + '0' * 1))
138
        self.assertEqual(pool.reserved, bitarray('1' * 39 + '0' * 1))
139

    
140
    def test_shrink_in_use(self):
141
        obj = DummyObject(8)
142
        pool = DummyPool(obj)
143
        pool._reserve(6)
144
        self.assertRaises(Exception, pool.shrink, 3)
145

    
146
    def test_count(self):
147
        obj = DummyObject(10)
148
        pool = DummyPool(obj)
149
        pool._reserve(1)
150
        pool._reserve(3)
151
        pool._reserve(4)
152
        pool._reserve(2, external=True)
153
        self.assertEqual(pool.count_available(), 6)
154
        self.assertEqual(pool.count_unavailable(), 4)
155
        self.assertEqual(pool.count_reserved(), 1)
156
        self.assertEqual(pool.count_unreserved(), 9)
157

    
158

    
159
class HelpersTestCase(TestCase):
160
    def test_find_padding(self):
161
        self.assertEqual(find_padding(1), 7)
162
        self.assertEqual(find_padding(8), 0)
163
        self.assertEqual(find_padding(12), 4)
164
        self.assertEqual(find_padding(16), 0)
165

    
166
    def test_bitarray_to_map(self):
167
        bt = bitarray('01001100101')
168
        map_ = bitarray_to_map(bt)
169
        self.assertEqual(map_, 'X.XX..XX.X.')
170

    
171

    
172
class BridgePoolTestCase(TestCase):
173
    def test_bridge_conversion(self):
174
        obj = DummyObject(13)
175
        obj.base = "bridge"
176
        pool = BridgePool(obj)
177
        for i in range(0, 13):
178
            self.assertEqual("bridge" + str(i + 1), pool.get())
179
        pool.put("bridge2")
180
        pool.put("bridge6")
181
        self.assertEqual("bridge2", pool.get())
182
        self.assertEqual("bridge6", pool.get())
183
        self.assertEqual(pool.empty(), True)
184

    
185

    
186
class MacPrefixPoolTestCase(TestCase):
187
    def test_invalid_mac_reservation(self):
188
        obj = DummyObject(65636)
189
        obj.base = 'ab:ff:ff'
190
        pool = MacPrefixPool(obj)
191
        for i in range(0, 65536):
192
            self.assertEqual(pool.is_available(i, index=True), False)
193

    
194
    def test_mac_prefix_conversion(self):
195
        obj = DummyObject(13)
196
        obj.base = 'aa:00:0'
197
        pool = MacPrefixPool(obj)
198
        for i in range(1, 9):
199
            self.assertEqual("aa:00:%s" % i, pool.get())
200

    
201
    def test_value_to_index(self):
202
        obj = DummyObject(13)
203
        obj.base = 'aa:00:0'
204
        pool = MacPrefixPool(obj)
205
        index = pool.value_to_index('aa:bc:ee')
206
        val = pool.index_to_value(index)
207
        self.assertEqual(val, 'aa:bc:ee')
208

    
209

    
210
class IPPoolTestCase(TestCase):
211
    def test_auto_reservations(self):
212
        obj = DummyObject(0)
213
        network = DummyObject(0)
214
        obj.network = network
215
        network.subnet = '192.168.2.0/24'
216
        network.gateway = '192.168.2.1'
217
        pool = IPPool(obj)
218
        self.assertEqual(pool.is_available('192.168.2.0'), False)
219
        self.assertEqual(pool.is_available('192.168.2.1'), False)
220
        self.assertEqual(pool.is_available('192.168.2.255'), False)
221
        self.assertEqual(pool.count_available(), 253)
222
        self.assertEqual(pool.get(), '192.168.2.2')
223

    
224
    def test_auto_reservations_2(self):
225
        obj = DummyObject(0)
226
        network = DummyObject(0)
227
        obj.network = network
228
        network.subnet = '192.168.2.0/31'
229
        network.gateway = '192.168.2.1'
230
        pool = IPPool(obj)
231
        self.assertEqual(pool.is_available('192.168.2.0'), False)
232
        self.assertEqual(pool.is_available('192.168.2.1'), False)
233
        self.assertEqual(pool.size(), 8)
234
        self.assertEqual(pool.empty(), True)