root / snf-cyclades-app / synnefo / db / pools / tests.py @ f8ce6d44
History | View | Annotate | Download (7.7 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 |
subnet.gateway = None
|
218 |
obj.base = "192.168.2.0/28"
|
219 |
obj.offset = 0
|
220 |
pool = IPPool(obj) |
221 |
# Test if reserved
|
222 |
pool.reserve("192.168.2.2")
|
223 |
self.assertRaises(ValueNotAvailable, pool.get, "192.168.2.2") |
224 |
# Test if externally reserved
|
225 |
pool.reserve("192.168.2.3", external=True) |
226 |
self.assertRaises(ValueNotAvailable, pool.get, "192.168.2.3") |
227 |
self.assertRaises(InvalidValue, pool.get, "192.168.2.16") |