Revision b43dcc5a test/ganeti.rpc_unittest.py
| b/test/ganeti.rpc_unittest.py | ||
|---|---|---|
| 59 | 59 |
def GetFakeSimpleStoreClass(fn): |
| 60 | 60 |
class FakeSimpleStore: |
| 61 | 61 |
GetNodePrimaryIPList = fn |
| 62 |
GetPrimaryIPFamily = lambda _: None |
|
| 62 | 63 |
|
| 63 | 64 |
return FakeSimpleStore |
| 64 | 65 |
|
| ... | ... | |
| 239 | 240 |
addr_list = ["192.0.2.%d" % n for n in range(0, 255, 13)] |
| 240 | 241 |
node_list = ["node%d.example.com" % n for n in range(0, 255, 13)] |
| 241 | 242 |
node_addr_list = [ " ".join(t) for t in zip(node_list, addr_list)] |
| 242 |
ssc = GetFakeSimpleStoreClass(lambda s: node_addr_list)
|
|
| 243 |
ssc = GetFakeSimpleStoreClass(lambda _: node_addr_list)
|
|
| 243 | 244 |
result = rpc._AddressLookup(node_list, ssc=ssc) |
| 244 | 245 |
self.assertEqual(result, addr_list) |
| 245 | 246 |
|
| 246 | 247 |
def testAddressLookupNSLookup(self): |
| 247 | 248 |
addr_list = ["192.0.2.%d" % n for n in range(0, 255, 13)] |
| 248 | 249 |
node_list = ["node%d.example.com" % n for n in range(0, 255, 13)] |
| 249 |
ssc = GetFakeSimpleStoreClass(lambda s: [])
|
|
| 250 |
ssc = GetFakeSimpleStoreClass(lambda _: [])
|
|
| 250 | 251 |
node_addr_map = dict(zip(node_list, addr_list)) |
| 251 |
nslookup_fn = lambda name: node_addr_map.get(name) |
|
| 252 |
nslookup_fn = lambda name, family=None: node_addr_map.get(name)
|
|
| 252 | 253 |
result = rpc._AddressLookup(node_list, ssc=ssc, nslookup_fn=nslookup_fn) |
| 253 | 254 |
self.assertEqual(result, addr_list) |
| 254 | 255 |
|
| ... | ... | |
| 257 | 258 |
node_list = ["node%d.example.com" % n for n in range(0, 255, 13)] |
| 258 | 259 |
n = len(addr_list) / 2 |
| 259 | 260 |
node_addr_list = [ " ".join(t) for t in zip(node_list[n:], addr_list[n:])] |
| 260 |
ssc = GetFakeSimpleStoreClass(lambda s: node_addr_list)
|
|
| 261 |
ssc = GetFakeSimpleStoreClass(lambda _: node_addr_list)
|
|
| 261 | 262 |
node_addr_map = dict(zip(node_list[:n], addr_list[:n])) |
| 262 |
nslookup_fn = lambda name: node_addr_map.get(name) |
|
| 263 |
nslookup_fn = lambda name, family=None: node_addr_map.get(name)
|
|
| 263 | 264 |
result = rpc._AddressLookup(node_list, ssc=ssc, nslookup_fn=nslookup_fn) |
| 264 | 265 |
self.assertEqual(result, addr_list) |
| 265 | 266 |
|
| 267 |
def testAddressLookupIPv6(self): |
|
| 268 |
addr_list = ["2001:db8::%d" % n for n in range(0, 255, 13)] |
|
| 269 |
node_list = ["node%d.example.com" % n for n in range(0, 255, 13)] |
|
| 270 |
node_addr_list = [ " ".join(t) for t in zip(node_list, addr_list)] |
|
| 271 |
ssc = GetFakeSimpleStoreClass(lambda _: node_addr_list) |
|
| 272 |
result = rpc._AddressLookup(node_list, ssc=ssc) |
|
| 273 |
self.assertEqual(result, addr_list) |
|
| 274 |
|
|
| 266 | 275 |
|
| 267 | 276 |
if __name__ == "__main__": |
| 268 | 277 |
testutils.GanetiTestProgram() |
Also available in: Unified diff