Revision 2dbe7563
b/snf-cyclades-app/synnefo/db/pools/tests.py | ||
---|---|---|
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() |
Also available in: Unified diff