#!/usr/bin/python
#

# Copyright (C) 2010 Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Script for unittesting the netutils module"""

import os
import re
import shutil
import socket
import tempfile
import unittest

import testutils
from ganeti import constants
from ganeti import errors
from ganeti import netutils
from ganeti import serializer
from ganeti import utils


def _GetSocketCredentials(path):
  """Connect to a Unix socket and return remote credentials.

  """
  sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  try:
    sock.settimeout(10)
    sock.connect(path)
    return netutils.GetSocketCredentials(sock)
  finally:
    sock.close()


class TestGetSocketCredentials(unittest.TestCase):
  def setUp(self):
    self.tmpdir = tempfile.mkdtemp()
    self.sockpath = utils.PathJoin(self.tmpdir, "sock")

    self.listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    self.listener.settimeout(10)
    self.listener.bind(self.sockpath)
    self.listener.listen(1)

  def tearDown(self):
    self.listener.shutdown(socket.SHUT_RDWR)
    self.listener.close()
    shutil.rmtree(self.tmpdir)

  def test(self):
    (c2pr, c2pw) = os.pipe()

    # Start child process
    child = os.fork()
    if child == 0:
      try:
        data = serializer.DumpJson(_GetSocketCredentials(self.sockpath))

        os.write(c2pw, data)
        os.close(c2pw)

        os._exit(0)
      finally:
        os._exit(1)

    os.close(c2pw)

    # Wait for one connection
    (conn, _) = self.listener.accept()
    conn.recv(1)
    conn.close()

    # Wait for result
    result = os.read(c2pr, 4096)
    os.close(c2pr)

    # Check child's exit code
    (_, status) = os.waitpid(child, 0)
    self.assertFalse(os.WIFSIGNALED(status))
    self.assertEqual(os.WEXITSTATUS(status), 0)

    # Check result
    (pid, uid, gid) = serializer.LoadJson(result)
    self.assertEqual(pid, os.getpid())
    self.assertEqual(uid, os.getuid())
    self.assertEqual(gid, os.getgid())


class TestHostname(unittest.TestCase):
  """Testing case for Hostname"""

  def testUppercase(self):
    data = "AbC.example.com"
    self.assertEqual(netutils.Hostname.GetNormalizedName(data), data.lower())

  def testTooLongName(self):
    data = "a.b." + "c" * 255
    self.assertRaises(errors.OpPrereqError,
                      netutils.Hostname.GetNormalizedName, data)

  def testTrailingDot(self):
    data = "a.b.c"
    self.assertEqual(netutils.Hostname.GetNormalizedName(data + "."), data)

  def testInvalidName(self):
    data = [
      "a b",
      "a/b",
      ".a.b",
      "a..b",
      ]
    for value in data:
      self.assertRaises(errors.OpPrereqError,
                        netutils.Hostname.GetNormalizedName, value)

  def testValidName(self):
    data = [
      "a.b",
      "a-b",
      "a_b",
      "a.b.c",
      ]
    for value in data:
      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)

  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:
  """Base class for TcpPing tests against listen(2)ing port"""
  family = None
  address = None

  def setUp(self):
    self.listener = socket.socket(self.family, socket.SOCK_STREAM)
    self.listener.bind((self.address, 0))
    self.listenerport = self.listener.getsockname()[1]
    self.listener.listen(1)

  def tearDown(self):
    self.listener.shutdown(socket.SHUT_RDWR)
    del self.listener
    del self.listenerport

  def testTcpPingToLocalHostAccept(self):
    self.assert_(netutils.TcpPing(self.address,
                                  self.listenerport,
                                  timeout=constants.TCP_PING_TIMEOUT,
                                  live_port_needed=True,
                                  source=self.address,
                                  ),
                 "failed to connect to test listener")

    self.assert_(netutils.TcpPing(self.address, self.listenerport,
                                  timeout=constants.TCP_PING_TIMEOUT,
                                  live_port_needed=True),
                 "failed to connect to test listener (no source)")


class TestIP4TcpPing(unittest.TestCase, _BaseTcpPingTest):
  """Testcase for IPv4 TCP version of ping - against listen(2)ing port"""
  family = socket.AF_INET
  address = constants.IP4_ADDRESS_LOCALHOST

  def setUp(self):
    unittest.TestCase.setUp(self)
    _BaseTcpPingTest.setUp(self)

  def tearDown(self):
    unittest.TestCase.tearDown(self)
    _BaseTcpPingTest.tearDown(self)


class TestIP6TcpPing(unittest.TestCase, _BaseTcpPingTest):
  """Testcase for IPv6 TCP version of ping - against listen(2)ing port"""
  family = socket.AF_INET6
  address = constants.IP6_ADDRESS_LOCALHOST

  def setUp(self):
    unittest.TestCase.setUp(self)
    _BaseTcpPingTest.setUp(self)

  def tearDown(self):
    unittest.TestCase.tearDown(self)
    _BaseTcpPingTest.tearDown(self)


class _BaseTcpPingDeafTest:
  """Base class for TcpPing tests against non listen(2)ing port"""
  family = None
  address = None

  def setUp(self):
    self.deaflistener = socket.socket(self.family, socket.SOCK_STREAM)
    self.deaflistener.bind((self.address, 0))
    self.deaflistenerport = self.deaflistener.getsockname()[1]

  def tearDown(self):
    del self.deaflistener
    del self.deaflistenerport

  def testTcpPingToLocalHostAcceptDeaf(self):
    self.assertFalse(netutils.TcpPing(self.address,
                                      self.deaflistenerport,
                                      timeout=constants.TCP_PING_TIMEOUT,
                                      live_port_needed=True,
                                      source=self.address,
                                      ), # need successful connect(2)
                     "successfully connected to deaf listener")

    self.assertFalse(netutils.TcpPing(self.address,
                                      self.deaflistenerport,
                                      timeout=constants.TCP_PING_TIMEOUT,
                                      live_port_needed=True,
                                      ), # need successful connect(2)
                     "successfully connected to deaf listener (no source)")

  def testTcpPingToLocalHostNoAccept(self):
    self.assert_(netutils.TcpPing(self.address,
                                  self.deaflistenerport,
                                  timeout=constants.TCP_PING_TIMEOUT,
                                  live_port_needed=False,
                                  source=self.address,
                                  ), # ECONNREFUSED is OK
                 "failed to ping alive host on deaf port")

    self.assert_(netutils.TcpPing(self.address,
                                  self.deaflistenerport,
                                  timeout=constants.TCP_PING_TIMEOUT,
                                  live_port_needed=False,
                                  ), # ECONNREFUSED is OK
                 "failed to ping alive host on deaf port (no source)")


class TestIP4TcpPingDeaf(unittest.TestCase, _BaseTcpPingDeafTest):
  """Testcase for IPv4 TCP version of ping - against non listen(2)ing port"""
  family = socket.AF_INET
  address = constants.IP4_ADDRESS_LOCALHOST

  def setUp(self):
    self.deaflistener = socket.socket(self.family, socket.SOCK_STREAM)
    self.deaflistener.bind((self.address, 0))
    self.deaflistenerport = self.deaflistener.getsockname()[1]

  def tearDown(self):
    del self.deaflistener
    del self.deaflistenerport


class TestIP6TcpPingDeaf(unittest.TestCase, _BaseTcpPingDeafTest):
  """Testcase for IPv6 TCP version of ping - against non listen(2)ing port"""
  family = socket.AF_INET6
  address = constants.IP6_ADDRESS_LOCALHOST

  def setUp(self):
    unittest.TestCase.setUp(self)
    _BaseTcpPingDeafTest.setUp(self)

  def tearDown(self):
    unittest.TestCase.tearDown(self)
    _BaseTcpPingDeafTest.tearDown(self)


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 = testutils.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 = testutils.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 = testutils.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 = testutils.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")


class TestValidatePortNumber(unittest.TestCase):
  """Test netutils.ValidatePortNumber"""

  def testPortNumberInt(self):
    self.assertRaises(ValueError, lambda: \
      netutils.ValidatePortNumber(500000))
    self.assertEqual(netutils.ValidatePortNumber(5000), 5000)

  def testPortNumberStr(self):
    self.assertRaises(ValueError, lambda: \
      netutils.ValidatePortNumber("pinky bunny"))
    self.assertEqual(netutils.ValidatePortNumber("5000"), 5000)

if __name__ == "__main__":
  testutils.GanetiTestProgram()