root / snf-cyclades-app / synnefo / db / pools / tests.py @ 3e7c63f8
History | View | Annotate | Download (5.3 kB)
1 |
import sys |
---|---|
2 |
from synnefo.db.pools import (PoolManager, EmptyPool, BridgePool, |
3 |
MacPrefixPool, IPPool) |
4 |
from bitarray import bitarray |
5 |
|
6 |
# Use backported unittest functionality if Python < 2.7
|
7 |
try:
|
8 |
import unittest2 as unittest |
9 |
except ImportError: |
10 |
if sys.version_info < (2, 7): |
11 |
raise Exception("The unittest2 package is required for Python < 2.7") |
12 |
import unittest |
13 |
|
14 |
|
15 |
class DummyObject(): |
16 |
def __init__(self, size): |
17 |
self.size = size
|
18 |
|
19 |
self.available_map = '' |
20 |
self.reserved_map = '' |
21 |
|
22 |
def save(self): |
23 |
pass
|
24 |
|
25 |
|
26 |
class DummyPool(PoolManager): |
27 |
def value_to_index(self, index): |
28 |
return index
|
29 |
|
30 |
def index_to_value(self, value): |
31 |
return value
|
32 |
|
33 |
|
34 |
class PoolManagerTestCase(unittest.TestCase): |
35 |
def test_created_pool(self): |
36 |
obj = DummyObject(42)
|
37 |
pool = DummyPool(obj) |
38 |
self.assertEqual(pool.to_01(), '1' * 42) |
39 |
self.assertEqual(pool.to_map(), '.' * 42) |
40 |
self.assertEqual(pool.available, bitarray('1' * 42 + '0' * 6)) |
41 |
self.assertEqual(pool.reserved, bitarray('1' * 42 + '0' * 6)) |
42 |
|
43 |
def test_save_pool(self): |
44 |
obj = DummyObject(42)
|
45 |
pool = DummyPool(obj) |
46 |
pool.save() |
47 |
self.assertNotEqual(obj.available_map, '') |
48 |
available_map = obj.available_map |
49 |
b = DummyPool(obj) |
50 |
b.save() |
51 |
self.assertEqual(obj.available_map, available_map)
|
52 |
|
53 |
def test_empty_pool(self): |
54 |
obj = DummyObject(42)
|
55 |
pool = DummyPool(obj) |
56 |
self.assertEqual(pool.empty(), False) |
57 |
for i in range(0, 42): |
58 |
self.assertEqual(pool.get(), i)
|
59 |
self.assertEqual(pool.empty(), True) |
60 |
self.assertRaises(EmptyPool, pool.get)
|
61 |
|
62 |
def test_reserved_value(self): |
63 |
obj = DummyObject(42)
|
64 |
pool = DummyPool(obj) |
65 |
available = pool.count_available() |
66 |
value = pool.get() |
67 |
self.assertEqual(pool.is_available(value), False) |
68 |
self.assertEqual(pool.count_available(), available - 1) |
69 |
pool.put(value) |
70 |
self.assertEqual(pool.is_available(value), True) |
71 |
self.assertEqual(pool.count_available(), available)
|
72 |
|
73 |
def test_external_reserved(self): |
74 |
obj = DummyObject(42)
|
75 |
pool = DummyPool(obj) |
76 |
for i in range(42, 48): |
77 |
self.assertEqual(pool.is_available(i), False) |
78 |
pool.reserve(32, external=True) |
79 |
values = [] |
80 |
while True: |
81 |
try:
|
82 |
values.append(pool.get()) |
83 |
except EmptyPool:
|
84 |
break
|
85 |
self.assertEqual(32 not in values, True) |
86 |
|
87 |
def test_external_reserved_2(self): |
88 |
obj = DummyObject(42)
|
89 |
pool = DummyPool(obj) |
90 |
self.assertEqual(pool.get(), 0) |
91 |
pool.reserve(0, external=True) |
92 |
pool.put(0)
|
93 |
self.assertEqual(pool.get(), 1) |
94 |
|
95 |
def test_extend_pool(self): |
96 |
obj = DummyObject(42)
|
97 |
pool = DummyPool(obj) |
98 |
pool.extend(7)
|
99 |
self.assertEqual(pool.to_01(), '1' * 49) |
100 |
self.assertEqual(pool.to_map(), '.' * 49) |
101 |
self.assertEqual(pool.available, bitarray('1' * 49 + '0' * 7)) |
102 |
self.assertEqual(pool.reserved, bitarray('1' * 49 + '0' * 7)) |
103 |
|
104 |
def test_shrink_pool(self): |
105 |
obj = DummyObject(42)
|
106 |
pool = DummyPool(obj) |
107 |
pool.shrink(3)
|
108 |
self.assertEqual(pool.to_01(), '1' * 39) |
109 |
self.assertEqual(pool.to_map(), '.' * 39) |
110 |
self.assertEqual(pool.available, bitarray('1' * 39 + '0' * 1)) |
111 |
self.assertEqual(pool.reserved, bitarray('1' * 39 + '0' * 1)) |
112 |
|
113 |
|
114 |
class BridgePoolTestCase(unittest.TestCase): |
115 |
def test_bridge_conversion(self): |
116 |
obj = DummyObject(13)
|
117 |
obj.base = "bridge"
|
118 |
pool = BridgePool(obj) |
119 |
for i in range(0, 13): |
120 |
self.assertEqual("bridge" + str(i + 1), pool.get()) |
121 |
pool.put("bridge2")
|
122 |
pool.put("bridge6")
|
123 |
self.assertEqual("bridge2", pool.get()) |
124 |
self.assertEqual("bridge6", pool.get()) |
125 |
self.assertEqual(pool.empty(), True) |
126 |
|
127 |
|
128 |
class MacPrefixPoolTestCase(unittest.TestCase): |
129 |
def test_invalid_mac_reservation(self): |
130 |
obj = DummyObject(65636)
|
131 |
obj.base = 'ab:ff:ff'
|
132 |
pool = MacPrefixPool(obj) |
133 |
for i in range(0, 65536): |
134 |
self.assertEqual(pool.is_available(i, index=True), False) |
135 |
|
136 |
class IPPoolTestCase(unittest.TestCase): |
137 |
def test_auto_reservations(self): |
138 |
obj = DummyObject(0)
|
139 |
network = DummyObject(0)
|
140 |
obj.network = network |
141 |
network.subnet = '192.168.2.0/24'
|
142 |
network.gateway = '192.168.2.1'
|
143 |
pool = IPPool(obj) |
144 |
self.assertEqual(pool.is_available('192.168.2.0'), False) |
145 |
self.assertEqual(pool.is_available('192.168.2.1'), False) |
146 |
self.assertEqual(pool.is_available('192.168.2.255'), False) |
147 |
self.assertEqual(pool.count_available(), 253) |
148 |
self.assertEqual(pool.get(), '192.168.2.2') |
149 |
|
150 |
def test_auto_reservations_2(self): |
151 |
obj = DummyObject(0)
|
152 |
network = DummyObject(0)
|
153 |
obj.network = network |
154 |
network.subnet = '192.168.2.0/31'
|
155 |
network.gateway = '192.168.2.1'
|
156 |
pool = IPPool(obj) |
157 |
self.assertEqual(pool.is_available('192.168.2.0'), False) |
158 |
self.assertEqual(pool.is_available('192.168.2.1'), False) |
159 |
self.assertEqual(pool.size(), 8) |
160 |
self.assertEqual(pool.empty(), True) |
161 |
|
162 |
|
163 |
if __name__ == '__main__': |
164 |
unittest.main() |