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) |