Revision 88d14415
b/lib/utils.py | ||
---|---|---|
749 | 749 |
|
750 | 750 |
""" |
751 | 751 |
return ' '.join([ShellQuote(i) for i in args]) |
752 |
|
|
753 |
|
|
754 |
def _ParseIpOutput(output): |
|
755 |
"""Parsing code for GetLocalIPAddresses(). |
|
756 |
|
|
757 |
This function is split out, so we can unit test it. |
|
758 |
|
|
759 |
""" |
|
760 |
re_ip = re.compile('^(\d+\.\d+\.\d+\.\d+)(?:/\d+)$') |
|
761 |
|
|
762 |
ips = [] |
|
763 |
for line in output.splitlines(False): |
|
764 |
fields = line.split() |
|
765 |
if len(line) < 4: |
|
766 |
continue |
|
767 |
m = re_ip.match(fields[3]) |
|
768 |
if m: |
|
769 |
ips.append(m.group(1)) |
|
770 |
|
|
771 |
return ips |
|
772 |
|
|
773 |
|
|
774 |
def GetLocalIPAddresses(): |
|
775 |
"""Gets a list of all local IP addresses. |
|
776 |
|
|
777 |
Should this break one day, a small Python module written in C could |
|
778 |
use the API call getifaddrs(). |
|
779 |
|
|
780 |
""" |
|
781 |
result = RunCmd(["ip", "-family", "inet", "-oneline", "addr", "show"]) |
|
782 |
if result.failed: |
|
783 |
raise errors.OpExecError("Command '%s' failed, error: %s," |
|
784 |
" output: %s" % (result.cmd, result.fail_reason, result.output)) |
|
785 |
|
|
786 |
return _ParseIpOutput(result.output) |
b/testing/ganeti.utils_unittest.py | ||
---|---|---|
32 | 32 |
from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \ |
33 | 33 |
RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \ |
34 | 34 |
ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \ |
35 |
ShellQuote, ShellQuoteArgs |
|
35 |
ShellQuote, ShellQuoteArgs, _ParseIpOutput
|
|
36 | 36 |
from ganeti.errors import LockError, UnitParseError |
37 | 37 |
|
38 | 38 |
class TestIsProcessAlive(unittest.TestCase): |
... | ... | |
424 | 424 |
self.assertEqual(ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c") |
425 | 425 |
|
426 | 426 |
|
427 |
class TestIpAdressList(unittest.TestCase): |
|
428 |
"""Test case for local IP addresses""" |
|
429 |
|
|
430 |
def _test(self, output, required): |
|
431 |
ips = _ParseIpOutput(output) |
|
432 |
|
|
433 |
# Sort the output, so our check below works in all cases |
|
434 |
ips.sort() |
|
435 |
required.sort() |
|
436 |
|
|
437 |
self.assertEqual(required, ips) |
|
438 |
|
|
439 |
def testSingleIpAddress(self): |
|
440 |
output = \ |
|
441 |
("3: lo inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n" |
|
442 |
"5: eth0 inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n") |
|
443 |
self._test(output, ['127.0.0.1', '10.0.0.1']) |
|
444 |
|
|
445 |
def testMultipleIpAddresses(self): |
|
446 |
output = \ |
|
447 |
("3: lo inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n" |
|
448 |
"5: eth0 inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n" |
|
449 |
"5: eth0 inet 1.2.3.4/8 brd 1.255.255.255 scope global eth0:test\n") |
|
450 |
self._test(output, ['127.0.0.1', '10.0.0.1', '1.2.3.4']) |
|
451 |
|
|
452 |
|
|
427 | 453 |
if __name__ == '__main__': |
428 | 454 |
unittest.main() |
Also available in: Unified diff