Rename spindle_usage to spindle_use on Ganeti side
[ganeti-local] / test / ganeti.netutils_unittest.py
index bedbdca..dcb7c4e 100755 (executable)
@@ -22,6 +22,7 @@
 """Script for unittesting the netutils module"""
 
 import os
+import re
 import shutil
 import socket
 import tempfile
@@ -102,21 +103,21 @@ class TestGetSocketCredentials(unittest.TestCase):
     self.assertEqual(gid, os.getgid())
 
 
-class TestHostInfo(unittest.TestCase):
-  """Testing case for HostInfo"""
+class TestHostname(unittest.TestCase):
+  """Testing case for Hostname"""
 
   def testUppercase(self):
     data = "AbC.example.com"
-    self.failUnlessEqual(netutils.HostInfo.NormalizeName(data), data.lower())
+    self.assertEqual(netutils.Hostname.GetNormalizedName(data), data.lower())
 
   def testTooLongName(self):
     data = "a.b." + "c" * 255
-    self.failUnlessRaises(errors.OpPrereqError,
-                          netutils.HostInfo.NormalizeName, data)
+    self.assertRaises(errors.OpPrereqError,
+                      netutils.Hostname.GetNormalizedName, data)
 
   def testTrailingDot(self):
     data = "a.b.c"
-    self.failUnlessEqual(netutils.HostInfo.NormalizeName(data + "."), data)
+    self.assertEqual(netutils.Hostname.GetNormalizedName(data + "."), data)
 
   def testInvalidName(self):
     data = [
@@ -126,8 +127,8 @@ class TestHostInfo(unittest.TestCase):
       "a..b",
       ]
     for value in data:
-      self.failUnlessRaises(errors.OpPrereqError,
-                            netutils.HostInfo.NormalizeName, value)
+      self.assertRaises(errors.OpPrereqError,
+                        netutils.Hostname.GetNormalizedName, value)
 
   def testValidName(self):
     data = [
@@ -137,55 +138,164 @@ class TestHostInfo(unittest.TestCase):
       "a.b.c",
       ]
     for value in data:
-      netutils.HostInfo.NormalizeName(value)
+      self.assertEqual(netutils.Hostname.GetNormalizedName(value), value)
+
+
+class TestIPAddress(unittest.TestCase):
+  def testIsValid(self):
+    self.assert_(netutils.IPAddress.IsValid("0.0.0.0"))
+    self.assert_(netutils.IPAddress.IsValid("127.0.0.1"))
+    self.assert_(netutils.IPAddress.IsValid("::"))
+    self.assert_(netutils.IPAddress.IsValid("::1"))
+
+  def testNotIsValid(self):
+    self.assertFalse(netutils.IPAddress.IsValid("0"))
+    self.assertFalse(netutils.IPAddress.IsValid("1.1.1.256"))
+    self.assertFalse(netutils.IPAddress.IsValid("a:g::1"))
+
+  def testGetAddressFamily(self):
+    fn = netutils.IPAddress.GetAddressFamily
+    self.assertEqual(fn("127.0.0.1"), socket.AF_INET)
+    self.assertEqual(fn("10.2.0.127"), socket.AF_INET)
+    self.assertEqual(fn("::1"), socket.AF_INET6)
+    self.assertEqual(fn("2001:db8::1"), socket.AF_INET6)
+    self.assertRaises(errors.IPAddressError, fn, "0")
+
+  def testValidateNetmask(self):
+    for netmask in [0, 33]:
+      self.assertFalse(netutils.IP4Address.ValidateNetmask(netmask))
+
+    for netmask in [1, 32]:
+      self.assertTrue(netutils.IP4Address.ValidateNetmask(netmask))
+
+    for netmask in [0, 129]:
+      self.assertFalse(netutils.IP6Address.ValidateNetmask(netmask))
+
+    for netmask in [1, 128]:
+      self.assertTrue(netutils.IP6Address.ValidateNetmask(netmask))
+
+  def testGetClassFromX(self):
+    self.assert_(
+        netutils.IPAddress.GetClassFromIpVersion(constants.IP4_VERSION) ==
+        netutils.IP4Address)
+    self.assert_(
+        netutils.IPAddress.GetClassFromIpVersion(constants.IP6_VERSION) ==
+        netutils.IP6Address)
+    self.assert_(
+        netutils.IPAddress.GetClassFromIpFamily(socket.AF_INET) ==
+        netutils.IP4Address)
+    self.assert_(
+        netutils.IPAddress.GetClassFromIpFamily(socket.AF_INET6) ==
+        netutils.IP6Address)
 
-
-class TestIsValidIP4(unittest.TestCase):
-  def test(self):
-    self.assert_(netutils.IsValidIP4("127.0.0.1"))
-    self.assert_(netutils.IsValidIP4("0.0.0.0"))
-    self.assert_(netutils.IsValidIP4("255.255.255.255"))
-    self.assertFalse(netutils.IsValidIP4("0"))
-    self.assertFalse(netutils.IsValidIP4("1"))
-    self.assertFalse(netutils.IsValidIP4("1.1.1"))
-    self.assertFalse(netutils.IsValidIP4("255.255.255.256"))
-    self.assertFalse(netutils.IsValidIP4("::1"))
-
-
-class TestIsValidIP6(unittest.TestCase):
-  def test(self):
-    self.assert_(netutils.IsValidIP6("::"))
-    self.assert_(netutils.IsValidIP6("::1"))
-    self.assert_(netutils.IsValidIP6("1" + (":1" * 7)))
-    self.assert_(netutils.IsValidIP6("ffff" + (":ffff" * 7)))
-    self.assertFalse(netutils.IsValidIP6("0"))
-    self.assertFalse(netutils.IsValidIP6(":1"))
-    self.assertFalse(netutils.IsValidIP6("f" + (":f" * 6)))
-    self.assertFalse(netutils.IsValidIP6("fffg" + (":ffff" * 7)))
-    self.assertFalse(netutils.IsValidIP6("fffff" + (":ffff" * 7)))
-    self.assertFalse(netutils.IsValidIP6("1" + (":1" * 8)))
-    self.assertFalse(netutils.IsValidIP6("127.0.0.1"))
-
-
-class TestIsValidIP(unittest.TestCase):
-  def test(self):
-    self.assert_(netutils.IsValidIP("0.0.0.0"))
-    self.assert_(netutils.IsValidIP("127.0.0.1"))
-    self.assert_(netutils.IsValidIP("::"))
-    self.assert_(netutils.IsValidIP("::1"))
-    self.assertFalse(netutils.IsValidIP("0"))
-    self.assertFalse(netutils.IsValidIP("1.1.1.256"))
-    self.assertFalse(netutils.IsValidIP("a:g::1"))
-
-
-class TestGetAddressFamily(unittest.TestCase):
-  def test(self):
-    self.assertEqual(netutils.GetAddressFamily("127.0.0.1"), socket.AF_INET)
-    self.assertEqual(netutils.GetAddressFamily("10.2.0.127"), socket.AF_INET)
-    self.assertEqual(netutils.GetAddressFamily("::1"), socket.AF_INET6)
-    self.assertEqual(netutils.GetAddressFamily("fe80::a00:27ff:fe08:5048"),
-                     socket.AF_INET6)
-    self.assertRaises(errors.GenericError, netutils.GetAddressFamily, "0")
+  def testOwnLoopback(self):
+    # FIXME: In a pure IPv6 environment this is no longer true
+    self.assert_(netutils.IPAddress.Own("127.0.0.1"),
+                 "Should own 127.0.0.1 address")
+
+  def testNotOwnAddress(self):
+    self.assertFalse(netutils.IPAddress.Own("2001:db8::1"),
+                     "Should not own IP address 2001:db8::1")
+    self.assertFalse(netutils.IPAddress.Own("192.0.2.1"),
+                     "Should not own IP address 192.0.2.1")
+
+  def testFamilyVersionConversions(self):
+    # IPAddress.GetAddressFamilyFromVersion
+    self.assertEqual(
+        netutils.IPAddress.GetAddressFamilyFromVersion(constants.IP4_VERSION),
+        socket.AF_INET)
+    self.assertEqual(
+        netutils.IPAddress.GetAddressFamilyFromVersion(constants.IP6_VERSION),
+        socket.AF_INET6)
+    self.assertRaises(errors.ProgrammerError,
+        netutils.IPAddress.GetAddressFamilyFromVersion, 3)
+
+    # IPAddress.GetVersionFromAddressFamily
+    self.assertEqual(
+        netutils.IPAddress.GetVersionFromAddressFamily(socket.AF_INET),
+        constants.IP4_VERSION)
+    self.assertEqual(
+        netutils.IPAddress.GetVersionFromAddressFamily(socket.AF_INET6),
+        constants.IP6_VERSION)
+    self.assertRaises(errors.ProgrammerError,
+        netutils.IPAddress.GetVersionFromAddressFamily, socket.AF_UNIX)
+
+
+class TestIP4Address(unittest.TestCase):
+  def testGetIPIntFromString(self):
+    fn = netutils.IP4Address._GetIPIntFromString
+    self.assertEqual(fn("0.0.0.0"), 0)
+    self.assertEqual(fn("0.0.0.1"), 1)
+    self.assertEqual(fn("127.0.0.1"), 2130706433)
+    self.assertEqual(fn("192.0.2.129"), 3221226113)
+    self.assertEqual(fn("255.255.255.255"), 2**32 - 1)
+    self.assertNotEqual(fn("0.0.0.0"), 1)
+    self.assertNotEqual(fn("0.0.0.0"), 1)
+
+  def testIsValid(self):
+    self.assert_(netutils.IP4Address.IsValid("0.0.0.0"))
+    self.assert_(netutils.IP4Address.IsValid("127.0.0.1"))
+    self.assert_(netutils.IP4Address.IsValid("192.0.2.199"))
+    self.assert_(netutils.IP4Address.IsValid("255.255.255.255"))
+
+  def testNotIsValid(self):
+    self.assertFalse(netutils.IP4Address.IsValid("0"))
+    self.assertFalse(netutils.IP4Address.IsValid("1"))
+    self.assertFalse(netutils.IP4Address.IsValid("1.1.1"))
+    self.assertFalse(netutils.IP4Address.IsValid("255.255.255.256"))
+    self.assertFalse(netutils.IP4Address.IsValid("::1"))
+
+  def testInNetwork(self):
+    self.assert_(netutils.IP4Address.InNetwork("127.0.0.0/8", "127.0.0.1"))
+
+  def testNotInNetwork(self):
+    self.assertFalse(netutils.IP4Address.InNetwork("192.0.2.0/24",
+                                                   "127.0.0.1"))
+
+  def testIsLoopback(self):
+    self.assert_(netutils.IP4Address.IsLoopback("127.0.0.1"))
+
+  def testNotIsLoopback(self):
+    self.assertFalse(netutils.IP4Address.IsLoopback("192.0.2.1"))
+
+
+class TestIP6Address(unittest.TestCase):
+  def testGetIPIntFromString(self):
+    fn = netutils.IP6Address._GetIPIntFromString
+    self.assertEqual(fn("::"), 0)
+    self.assertEqual(fn("::1"), 1)
+    self.assertEqual(fn("2001:db8::1"),
+                     42540766411282592856903984951653826561L)
+    self.assertEqual(fn("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"), 2**128-1)
+    self.assertNotEqual(netutils.IP6Address._GetIPIntFromString("::2"), 1)
+
+  def testIsValid(self):
+    self.assert_(netutils.IP6Address.IsValid("::"))
+    self.assert_(netutils.IP6Address.IsValid("::1"))
+    self.assert_(netutils.IP6Address.IsValid("1" + (":1" * 7)))
+    self.assert_(netutils.IP6Address.IsValid("ffff" + (":ffff" * 7)))
+    self.assert_(netutils.IP6Address.IsValid("::"))
+
+  def testNotIsValid(self):
+    self.assertFalse(netutils.IP6Address.IsValid("0"))
+    self.assertFalse(netutils.IP6Address.IsValid(":1"))
+    self.assertFalse(netutils.IP6Address.IsValid("f" + (":f" * 6)))
+    self.assertFalse(netutils.IP6Address.IsValid("fffg" + (":ffff" * 7)))
+    self.assertFalse(netutils.IP6Address.IsValid("fffff" + (":ffff" * 7)))
+    self.assertFalse(netutils.IP6Address.IsValid("1" + (":1" * 8)))
+    self.assertFalse(netutils.IP6Address.IsValid("127.0.0.1"))
+
+  def testInNetwork(self):
+    self.assert_(netutils.IP6Address.InNetwork("::1/128", "::1"))
+
+  def testNotInNetwork(self):
+    self.assertFalse(netutils.IP6Address.InNetwork("2001:db8::1/128", "::1"))
+
+  def testIsLoopback(self):
+    self.assert_(netutils.IP6Address.IsLoopback("::1"))
+
+  def testNotIsLoopback(self):
+    self.assertFalse(netutils.IP6Address.IsLoopback("2001:db8::1"))
 
 
 class _BaseTcpPingTest:
@@ -323,23 +433,102 @@ class TestIP6TcpPingDeaf(unittest.TestCase, _BaseTcpPingDeafTest):
     _BaseTcpPingDeafTest.tearDown(self)
 
 
-class TestOwnIpAddress(unittest.TestCase):
-  """Testcase for OwnIpAddress"""
-
-  def testOwnLoopback(self):
-    """check having the loopback ip"""
-    self.failUnless(netutils.OwnIpAddress(constants.IP4_ADDRESS_LOCALHOST),
-                    "Should own the loopback address")
-
-  def testNowOwnAddress(self):
-    """check that I don't own an address"""
-
-    # Network 192.0.2.0/24 is reserved for test/documentation as per
-    # RFC 5737, so we *should* not have an address of this range... if
-    # this fails, we should extend the test to multiple addresses
-    DST_IP = "192.0.2.1"
-    self.failIf(netutils.OwnIpAddress(DST_IP),
-                "Should not own IP address %s" % DST_IP)
+class TestFormatAddress(unittest.TestCase):
+  """Testcase for FormatAddress"""
+
+  def testFormatAddressUnixSocket(self):
+    res1 = netutils.FormatAddress(("12352", 0, 0), family=socket.AF_UNIX)
+    self.assertEqual(res1, "pid=12352, uid=0, gid=0")
+
+  def testFormatAddressIP4(self):
+    res1 = netutils.FormatAddress(("127.0.0.1", 1234), family=socket.AF_INET)
+    self.assertEqual(res1, "127.0.0.1:1234")
+    res2 = netutils.FormatAddress(("192.0.2.32", None), family=socket.AF_INET)
+    self.assertEqual(res2, "192.0.2.32")
+
+  def testFormatAddressIP6(self):
+    res1 = netutils.FormatAddress(("::1", 1234), family=socket.AF_INET6)
+    self.assertEqual(res1, "[::1]:1234")
+    res2 = netutils.FormatAddress(("::1", None), family=socket.AF_INET6)
+    self.assertEqual(res2, "[::1]")
+    res2 = netutils.FormatAddress(("2001:db8::beef", "80"),
+                                  family=socket.AF_INET6)
+    self.assertEqual(res2, "[2001:db8::beef]:80")
+
+  def testFormatAddressWithoutFamily(self):
+    res1 = netutils.FormatAddress(("127.0.0.1", 1234))
+    self.assertEqual(res1, "127.0.0.1:1234")
+    res2 = netutils.FormatAddress(("::1", 1234))
+    self.assertEqual(res2, "[::1]:1234")
+
+
+  def testInvalidFormatAddress(self):
+    self.assertRaises(errors.ParameterError, netutils.FormatAddress,
+                      "127.0.0.1")
+    self.assertRaises(errors.ParameterError, netutils.FormatAddress,
+                      "127.0.0.1", family=socket.AF_INET)
+    self.assertRaises(errors.ParameterError, netutils.FormatAddress,
+                      ("::1"), family=socket.AF_INET )
+
+class TestIpParsing(testutils.GanetiTestCase):
+  """Test the code that parses the ip command output"""
+
+  def testIp4(self):
+    valid_addresses = [constants.IP4_ADDRESS_ANY,
+                       constants.IP4_ADDRESS_LOCALHOST,
+                       "192.0.2.1",     # RFC5737, IPv4 address blocks for docs
+                       "198.51.100.1",
+                       "203.0.113.1",
+                      ]
+    for addr in valid_addresses:
+      self.failUnless(re.search(netutils._IP_RE_TEXT, addr))
+
+  def testIp6(self):
+    valid_addresses = [constants.IP6_ADDRESS_ANY,
+                       constants.IP6_ADDRESS_LOCALHOST,
+                       "0:0:0:0:0:0:0:1", # other form for IP6_ADDRESS_LOCALHOST
+                       "0:0:0:0:0:0:0:0", # other form for IP6_ADDRESS_ANY
+                       "2001:db8:85a3::8a2e:370:7334", # RFC3849 IP6 docs block
+                       "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
+                       "0:0:0:0:0:FFFF:192.0.2.1",  # IPv4-compatible IPv6
+                       "::FFFF:192.0.2.1",
+                       "0:0:0:0:0:0:203.0.113.1",   # IPv4-mapped IPv6
+                       "::203.0.113.1",
+                      ]
+    for addr in valid_addresses:
+      self.failUnless(re.search(netutils._IP_RE_TEXT, addr))
+
+  def testParseIpCommandOutput(self):
+    # IPv4-only, fake loopback interface
+    tests = ["ip-addr-show-lo-ipv4.txt", "ip-addr-show-lo-oneline-ipv4.txt"]
+    for test_file in tests:
+      data = self._ReadTestData(test_file)
+      addr = netutils._GetIpAddressesFromIpOutput(data)
+      self.failUnless(len(addr[4]) == 1 and addr[4][0] == "127.0.0.1" and not
+                      addr[6])
+
+    # IPv6-only, fake loopback interface
+    tests = ["ip-addr-show-lo-ipv6.txt", "ip-addr-show-lo-ipv6.txt"]
+    for test_file in tests:
+      data = self._ReadTestData(test_file)
+      addr = netutils._GetIpAddressesFromIpOutput(data)
+      self.failUnless(len(addr[6]) == 1 and addr[6][0] == "::1" and not addr[4])
+
+    # IPv4 and IPv6, fake loopback interface
+    tests = ["ip-addr-show-lo.txt", "ip-addr-show-lo-oneline.txt"]
+    for test_file in tests:
+      data = self._ReadTestData(test_file)
+      addr = netutils._GetIpAddressesFromIpOutput(data)
+      self.failUnless(len(addr[6]) == 1 and addr[6][0] == "::1" and
+                      len(addr[4]) == 1 and addr[4][0] == "127.0.0.1")
+
+    # IPv4 and IPv6, dummy interface
+    data = self._ReadTestData("ip-addr-show-dummy0.txt")
+    addr = netutils._GetIpAddressesFromIpOutput(data)
+    self.failUnless(len(addr[6]) == 1 and
+                    addr[6][0] == "2001:db8:85a3::8a2e:370:7334" and
+                    len(addr[4]) == 1 and
+                    addr[4][0] == "192.0.2.1")
 
 
 if __name__ == "__main__":