| commit | author | age | ||
| 659253 | 1 | import ipaddress |
| JK | 2 | import unittest |
| 3 | ||
| 4 | from lib.Ipam import * | |
| 5 | from docker_plugin_api.Plugin import InputValidationException | |
| 6 | ||
| 7 | ||
| 8 | class PoolInvalidCreateTest(unittest.TestCase): | |
| 9 | def test_noargs(self): | |
| 10 | with self.assertRaises(InputValidationException): | |
| 11 | Pool() | |
| 12 | ||
| 13 | def test_subpool_only(self): | |
| 14 | with self.assertRaises(InputValidationException): | |
| 15 | Pool(subPool='127.0.0.0/24') | |
| 16 | ||
| 17 | def test_subpool_invalid(self): | |
| 18 | with self.assertRaises(InputValidationException): | |
| 19 | Pool(pool='127.0.0.0/24', subPool='127.1.2.3/24') | |
| 20 | ||
| 21 | ||
| 22 | class PoolCreateTest(unittest.TestCase): | |
| 23 | def test_auto_ipv4(self): | |
| 24 | pool = Pool(v6=False) | |
| 25 | ip = ipaddress.ip_network(str(pool)) | |
| 26 | self.assertTrue(isinstance(ip, ipaddress.IPv4Network)) | |
| 27 | ||
| 28 | def test_auto_ipv6(self): | |
| 29 | pool = Pool(v6=True) | |
| 30 | ip = ipaddress.ip_network(str(pool)) | |
| 31 | self.assertTrue(isinstance(ip, ipaddress.IPv6Network)) | |
| 32 | ||
| 33 | def test_pool_ipv4(self): | |
| 34 | pool = Pool(pool='127.0.0.1/24') | |
| 35 | ip = ipaddress.ip_network(str(pool)) | |
| 36 | self.assertTrue(isinstance(ip, ipaddress.IPv4Network)) | |
| 37 | self.assertEqual(ip, ipaddress.ip_network('127.0.0.0/24')) | |
| 38 | ||
| 39 | def test_pool_ipv6(self): | |
| 40 | pool = Pool(pool='fd00:a:b:c:1:2:3:4/64') | |
| 41 | ip = ipaddress.ip_network(str(pool)) | |
| 42 | self.assertTrue(isinstance(ip, ipaddress.IPv6Network)) | |
| 43 | self.assertEqual(ip, ipaddress.ip_network('fd00:a:b:c::/64')) | |
| 44 | ||
| 45 | def test_subpool_ipv4(self): | |
| 46 | pool = Pool(pool='127.0.0.1/8', subPool='127.0.0.2/24') | |
| 47 | ip = ipaddress.ip_network(str(pool)) | |
| 48 | self.assertTrue(isinstance(ip, ipaddress.IPv4Network)) | |
| 49 | self.assertEqual(ip, ipaddress.ip_network('127.0.0.0/8')) | |
| 50 | ip = ipaddress.ip_network(str(pool.subpool)) | |
| 51 | self.assertTrue(isinstance(ip, ipaddress.IPv4Network)) | |
| 52 | self.assertEqual(ip, ipaddress.ip_network('127.0.0.0/24')) | |
| 53 | ||
| 54 | def test_subpool_ipv6(self): | |
| 55 | pool = Pool(pool='fd00:a:b:c:1:2:3:4/48', subPool='fd00:a:b:c:1:2:3:4/64') | |
| 56 | ip = ipaddress.ip_network(str(pool)) | |
| 57 | self.assertTrue(isinstance(ip, ipaddress.IPv6Network)) | |
| 58 | self.assertEqual(ip, ipaddress.ip_network('fd00:a:b::/48')) | |
| 59 | ip = ipaddress.ip_network(str(pool.subpool)) | |
| 60 | self.assertTrue(isinstance(ip, ipaddress.IPv6Network)) | |
| 61 | self.assertEqual(ip, ipaddress.ip_network('fd00:a:b:c::/64')) | |
| 62 | ||
| 63 | ||
| 64 | class PoolComparisonTest(unittest.TestCase): | |
| 65 | def test_pool_same_ipv4(self): | |
| 66 | pool1 = Pool(pool='127.0.0.1/8') | |
| 67 | pool2 = Pool(pool='127.0.0.2/8') | |
| 68 | self.assertEqual(pool1, pool2) | |
| 69 | self.assertEqual(pool2, pool1) | |
| 70 | ||
| 71 | def test_pool_same_ipv6(self): | |
| 72 | pool1 = Pool(pool='fd00:a:b::/64') | |
| 73 | pool2 = Pool(pool='fd00:a:b:0:1:2:3:4/64') | |
| 74 | self.assertEqual(pool1, pool2) | |
| 75 | self.assertEqual(pool2, pool1) | |
| 76 | ||
| 77 | ||
| 78 | class PoolOverlapTest(unittest.TestCase): | |
| 79 | def test_pool_overlap_ipv4(self): | |
| 80 | pool1 = Pool(pool='127.0.0.1/24') | |
| 81 | pool2 = Pool(pool='127.0.2.3/16') | |
| 82 | self.assertTrue(pool1.overlaps(pool1)) | |
| 83 | self.assertTrue(pool2.overlaps(pool2)) | |
| 84 | self.assertTrue(pool1.overlaps(pool2)) | |
| 85 | self.assertTrue(pool2.overlaps(pool1)) | |
| 86 | ||
| 87 | def test_pool_overlap_ipv6(self): | |
| 88 | pool1 = Pool(pool='fe80::/64') | |
| 89 | pool2 = Pool(pool='fe80::1:2:3:4/72') | |
| 90 | self.assertTrue(pool1.overlaps(pool1)) | |
| 91 | self.assertTrue(pool2.overlaps(pool2)) | |
| 92 | self.assertTrue(pool1.overlaps(pool2)) | |
| 93 | self.assertTrue(pool2.overlaps(pool1)) | |
| 94 | ||
| 95 | ||
| 96 | class PoolAllocateInvalidTest(unittest.TestCase): | |
| 97 | def test_pool_allocate_invalid_ipv4(self): | |
| 98 | pool = Pool(pool='127.0.0.0/30') | |
| 99 | self.assertEqual(pool.allocate(), '127.0.0.1/30') | |
| 100 | self.assertEqual(pool.allocate(), '127.0.0.2/30') | |
| 101 | with self.assertRaises(InputValidationException): | |
| 102 | pool.allocate('126.255.255.255') | |
| 103 | with self.assertRaises(InputValidationException): | |
| 104 | pool.allocate('127.0.0.0') | |
| 105 | with self.assertRaises(InputValidationException): | |
| 106 | pool.allocate('127.0.0.1') | |
| 107 | with self.assertRaises(InputValidationException): | |
| 108 | pool.allocate('127.0.0.2') | |
| 109 | with self.assertRaises(InputValidationException): | |
| 110 | pool.allocate('127.0.0.3') | |
| 111 | with self.assertRaises(InputValidationException): | |
| 112 | pool.allocate('127.0.0.4') | |
| 113 | ||
| 114 | def test_pool_allocate_invalid_ipv6(self): | |
| 115 | pool = Pool(pool='fd00::/126') | |
| 116 | self.assertEqual(pool.allocate(), 'fd00::1/126') | |
| 117 | self.assertEqual(pool.allocate(), 'fd00::2/126') | |
| 118 | self.assertEqual(pool.allocate(), 'fd00::3/126') | |
| 119 | with self.assertRaises(InputValidationException): | |
| 120 | pool.allocate('fcff:ffff:ffff:ffff:ffff:ffff:ffff:ffff') | |
| 121 | with self.assertRaises(InputValidationException): | |
| 122 | pool.allocate('fd00::') | |
| 123 | with self.assertRaises(InputValidationException): | |
| 124 | pool.allocate('fd00::1') | |
| 125 | with self.assertRaises(InputValidationException): | |
| 126 | pool.allocate('fd00::2') | |
| 127 | with self.assertRaises(InputValidationException): | |
| 128 | pool.allocate('fd00::3') | |
| 129 | with self.assertRaises(InputValidationException): | |
| 130 | pool.allocate('fd00::4') | |
| 131 | ||
| 132 | ||
| 133 | class PoolAllocateOrderTest(unittest.TestCase): | |
| 134 | def test_pool_allocate_ipv4(self): | |
| 135 | pool = Pool(pool='127.0.0.0/30') | |
| 136 | self.assertEqual(pool.allocate(), '127.0.0.1/30') | |
| 137 | self.assertEqual(pool.allocate(), '127.0.0.2/30') | |
| 138 | with self.assertRaises(InputValidationException): | |
| 139 | pool.allocate() | |
| 140 | with self.assertRaises(InputValidationException): | |
| 141 | pool.allocate() | |
| 142 | with self.assertRaises(InputValidationException): | |
| 143 | pool.allocate() | |
| 144 | ||
| 145 | def test_pool_allocate_ipv6(self): | |
| 146 | pool = Pool(pool='fd00::/126') | |
| 147 | self.assertEqual(pool.allocate(), 'fd00::1/126') | |
| 148 | self.assertEqual(pool.allocate(), 'fd00::2/126') | |
| 149 | self.assertEqual(pool.allocate(), 'fd00::3/126') | |
| 150 | with self.assertRaises(InputValidationException): | |
| 151 | pool.allocate() | |
| 152 | with self.assertRaises(InputValidationException): | |
| 153 | pool.allocate() | |
| 154 | with self.assertRaises(InputValidationException): | |
| 155 | pool.allocate() | |
| 156 | ||
| 157 | def test_pool_allocate_large_ipv4(self): | |
| 158 | pool = Pool(pool='127.0.0.0/8') | |
| 159 | self.assertEqual(pool.allocate(), '127.0.0.1/8') | |
| 160 | self.assertEqual(pool.allocate(), '127.0.0.2/8') | |
| 161 | ||
| 162 | def test_pool_allocate_large_ipv6(self): | |
| 163 | pool = Pool(pool='fd00::/56') | |
| 164 | self.assertEqual(pool.allocate(), 'fd00::1/56') | |
| 165 | self.assertEqual(pool.allocate(), 'fd00::2/56') | |
| 166 | ||
| 167 | ||
| 168 | class PoolAllocateManualTest(unittest.TestCase): | |
| 169 | def test_pool_allocate_ipv4(self): | |
| 170 | pool = Pool(pool='127.0.0.0/28') | |
| 171 | self.assertEqual(pool.allocate('127.0.0.3'), '127.0.0.3/28') | |
| 172 | self.assertEqual(pool.allocate('127.0.0.5'), '127.0.0.5/28') | |
| 173 | self.assertEqual(pool.allocate('127.0.0.11'), '127.0.0.11/28') | |
| 174 | self.assertEqual(pool.allocate('127.0.0.14'), '127.0.0.14/28') | |
| 175 | self.assertEqual(pool.allocate(), '127.0.0.1/28') | |
| 176 | self.assertEqual(pool.allocate(), '127.0.0.2/28') | |
| 177 | self.assertEqual(pool.allocate(), '127.0.0.4/28') | |
| 178 | self.assertEqual(pool.allocate(), '127.0.0.6/28') | |
| 179 | self.assertEqual(pool.allocate(), '127.0.0.7/28') | |
| 180 | self.assertEqual(pool.allocate(), '127.0.0.8/28') | |
| 181 | self.assertEqual(pool.allocate(), '127.0.0.9/28') | |
| 182 | self.assertEqual(pool.allocate(), '127.0.0.10/28') | |
| 183 | self.assertEqual(pool.allocate(), '127.0.0.12/28') | |
| 184 | self.assertEqual(pool.allocate(), '127.0.0.13/28') | |
| 185 | with self.assertRaises(InputValidationException): | |
| 186 | pool.allocate() | |
| 187 | ||
| 188 | def test_pool_allocate_ipv6(self): | |
| 189 | pool = Pool(pool='fd00::/124') | |
| 190 | self.assertEqual(pool.allocate('fd00::3'), 'fd00::3/124') | |
| 191 | self.assertEqual(pool.allocate('fd00::5'), 'fd00::5/124') | |
| 192 | self.assertEqual(pool.allocate('fd00::b'), 'fd00::b/124') | |
| 193 | self.assertEqual(pool.allocate('fd00::e'), 'fd00::e/124') | |
| 194 | self.assertEqual(pool.allocate(), 'fd00::1/124') | |
| 195 | self.assertEqual(pool.allocate(), 'fd00::2/124') | |
| 196 | self.assertEqual(pool.allocate(), 'fd00::4/124') | |
| 197 | self.assertEqual(pool.allocate(), 'fd00::6/124') | |
| 198 | self.assertEqual(pool.allocate(), 'fd00::7/124') | |
| 199 | self.assertEqual(pool.allocate(), 'fd00::8/124') | |
| 200 | self.assertEqual(pool.allocate(), 'fd00::9/124') | |
| 201 | self.assertEqual(pool.allocate(), 'fd00::a/124') | |
| 202 | self.assertEqual(pool.allocate(), 'fd00::c/124') | |
| 203 | self.assertEqual(pool.allocate(), 'fd00::d/124') | |
| 204 | self.assertEqual(pool.allocate(), 'fd00::f/124') | |
| 205 | with self.assertRaises(InputValidationException): | |
| 206 | pool.allocate() | |
| 207 | ||
| 208 | def test_subpool_allocate_ipv4(self): | |
| 209 | pool = Pool(pool='127.0.0.0/8', subPool='127.0.0.0/30') | |
| 210 | self.assertEqual(pool.allocate('127.0.0.2'), '127.0.0.2/8') | |
| 211 | self.assertEqual(pool.allocate('127.1.2.3'), '127.1.2.3/8') | |
| 212 | self.assertEqual(pool.allocate('127.255.0.5'), '127.255.0.5/8') | |
| 213 | self.assertEqual(pool.allocate('127.248.255.11'), '127.248.255.11/8') | |
| 214 | self.assertEqual(pool.allocate('127.71.12.14'), '127.71.12.14/8') | |
| 215 | self.assertEqual(pool.allocate(), '127.0.0.1/8') | |
| 216 | # This is not-fully-correct as we can theoretically assign 127.0.0.3 - feel free to fix it | |
| 217 | with self.assertRaises(InputValidationException): | |
| 218 | pool.allocate() | |
| 219 | self.assertEqual(pool.allocate('127.0.0.3'), '127.0.0.3/8') | |
| 220 | ||
| 221 | def test_subpool_middle_allocate_ipv4(self): | |
| 222 | pool = Pool(pool='127.0.0.0/8', subPool='127.248.255.0/30') | |
| 223 | self.assertEqual(pool.allocate('127.0.0.2'), '127.0.0.2/8') | |
| 224 | self.assertEqual(pool.allocate('127.1.2.3'), '127.1.2.3/8') | |
| 225 | self.assertEqual(pool.allocate('127.255.0.5'), '127.255.0.5/8') | |
| 226 | self.assertEqual(pool.allocate('127.248.255.2'), '127.248.255.2/8') | |
| 227 | self.assertEqual(pool.allocate('127.71.12.14'), '127.71.12.14/8') | |
| 228 | self.assertEqual(pool.allocate(), '127.248.255.1/8') | |
| 229 | # This is not-fully-correct as we can theoretically assign 127.248.255.0 and 127.248.255.3 - feel free to fix it | |
| 230 | with self.assertRaises(InputValidationException): | |
| 231 | pool.allocate() | |
| 232 | self.assertEqual(pool.allocate('127.248.255.0'), '127.248.255.0/8') | |
| 233 | self.assertEqual(pool.allocate('127.248.255.3'), '127.248.255.3/8') | |
| 234 | ||
| 235 | def test_subpool_allocate_ipv6(self): | |
| 236 | pool = Pool(pool='fe80::/64', subPool='fe80::/126') | |
| 237 | self.assertEqual(pool.allocate('fe80::2'), 'fe80::2/64') | |
| 238 | self.assertEqual(pool.allocate('fe80::1:2:3:4'), 'fe80::1:2:3:4/64') | |
| 239 | self.assertEqual(pool.allocate('fe80::ffff:fefe:fdfd:fcfc'), 'fe80::ffff:fefe:fdfd:fcfc/64') | |
| 240 | self.assertEqual(pool.allocate(), 'fe80::1/64') | |
| 241 | self.assertEqual(pool.allocate(), 'fe80::3/64') | |
| 242 | with self.assertRaises(InputValidationException): | |
| 243 | pool.allocate() | |
| 244 | ||
| 245 | def test_subpool_middle_allocate_ipv6(self): | |
| 246 | pool = Pool(pool='fe80::/64', subPool='fe80::ffff:0:0:0/126') | |
| 247 | self.assertEqual(pool.allocate('fe80::2'), 'fe80::2/64') | |
| 248 | self.assertEqual(pool.allocate('fe80::1:2:3:4'), 'fe80::1:2:3:4/64') | |
| 249 | self.assertEqual(pool.allocate('fe80::ffff:0:0:2'), 'fe80::ffff:0:0:2/64') | |
| 250 | self.assertEqual(pool.allocate(), 'fe80::ffff:0:0:1/64') | |
| 251 | self.assertEqual(pool.allocate(), 'fe80::ffff:0:0:3/64') | |
| 252 | # This is not-fully-correct as we can theoretically assign fe80::ffff:0:0:0 - feel free to fix it | |
| 253 | with self.assertRaises(InputValidationException): | |
| 254 | pool.allocate() | |
| 255 | self.assertEqual(pool.allocate('fe80::ffff:0:0:0'), 'fe80::ffff:0:0:0/64') | |
| 256 | ||
| 257 | ||
| 258 | class TestPoolAllocateRelease(unittest.TestCase): | |
| 259 | def test_pool_allocate_release_ipv4(self): | |
| 260 | pool = Pool(pool='127.0.0.0/29') | |
| 261 | self.assertEqual(pool.allocate('127.0.0.3'), '127.0.0.3/29') | |
| 262 | self.assertEqual(pool.allocate('127.0.0.5'), '127.0.0.5/29') | |
| 263 | self.assertEqual(pool.allocate(), '127.0.0.1/29') | |
| 264 | self.assertEqual(pool.allocate(), '127.0.0.2/29') | |
| 265 | pool.deallocate('127.0.0.3') | |
| 266 | self.assertEqual(pool.allocate(), '127.0.0.3/29') | |
| 267 | self.assertEqual(pool.allocate(), '127.0.0.4/29') | |
| 268 | self.assertEqual(pool.allocate(), '127.0.0.6/29') | |
| 269 | with self.assertRaises(InputValidationException): | |
| 270 | pool.allocate() | |
| 271 | pool.deallocate('127.0.0.5') | |
| 272 | self.assertEqual(pool.allocate(), '127.0.0.5/29') | |
| 273 | pool.deallocate('127.0.0.1') | |
| 274 | pool.deallocate('127.0.0.2') | |
| 275 | self.assertEqual(pool.allocate('127.0.0.2'), '127.0.0.2/29') | |
| 276 | self.assertEqual(pool.allocate(), '127.0.0.1/29') | |
| 277 | ||
| 278 | def test_pool_allocate_release_ipv6(self): | |
| 279 | pool = Pool(pool='fe80::/125') | |
| 280 | self.assertEqual(pool.allocate('fe80::3'), 'fe80::3/125') | |
| 281 | self.assertEqual(pool.allocate('fe80::5'), 'fe80::5/125') | |
| 282 | self.assertEqual(pool.allocate(), 'fe80::1/125') | |
| 283 | self.assertEqual(pool.allocate(), 'fe80::2/125') | |
| 284 | pool.deallocate('fe80::3') | |
| 285 | self.assertEqual(pool.allocate(), 'fe80::3/125') | |
| 286 | self.assertEqual(pool.allocate(), 'fe80::4/125') | |
| 287 | self.assertEqual(pool.allocate(), 'fe80::6/125') | |
| 288 | self.assertEqual(pool.allocate(), 'fe80::7/125') | |
| 289 | with self.assertRaises(InputValidationException): | |
| 290 | pool.allocate() | |
| 291 | pool.deallocate('fe80::5') | |
| 292 | self.assertEqual(pool.allocate(), 'fe80::5/125') | |
| 293 | pool.deallocate('fe80::1') | |
| 294 | pool.deallocate('fe80::2') | |
| 295 | self.assertEqual(pool.allocate('fe80::2'), 'fe80::2/125') | |
| 296 | self.assertEqual(pool.allocate(), 'fe80::1/125') | |
