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