Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (7.6 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, ValueNotAvailable,
39
                              InvalidValue)
40
from bitarray import bitarray
41

    
42

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

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

    
50
    def save(self):
51
        pass
52

    
53

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

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

    
61

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

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

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

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

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

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

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

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

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

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

    
159

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

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

    
172

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

    
186

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

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

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

    
210

    
211
class IPPoolTestCase(TestCase):
212
    def test_get_with_value(self):
213
        obj = DummyObject(16)
214
        subnet = DummyObject(0)
215
        obj.subnet = subnet
216
        subnet.cidr = "192.168.2.0/28"
217
        obj.base = "192.168.2.0/28"
218
        obj.offset = 0
219
        pool = IPPool(obj)
220
        # Test if reserved
221
        pool.reserve("192.168.2.2")
222
        self.assertRaises(ValueNotAvailable, pool.get, "192.168.2.2")
223
        # Test if externally reserved
224
        pool.reserve("192.168.2.3", external=True)
225
        self.assertRaises(ValueNotAvailable, pool.get, "192.168.2.3")
226
        self.assertRaises(InvalidValue, pool.get, "192.168.2.16")