Synnefo » snf-ganeti

	test/ganeti.hypervisor.hv_kvm_unittest.py \

_KVM_NETWORK_SCRIPT = constants.SYSCONFDIR + "/ganeti/kvm-vif-bridge"

def _WriteNetScript(instance, nic, index):

  """Write a script to connect a net interface to the proper bridge.

  This can be used by any qemu-type hypervisor.

  @type instance: L{objects.Instance}

  @param instance: Instance object

  @type nic: L{objects.NIC}

  @param nic: NIC object

  @type index: int

  @param index: NIC index

  @return: Script

  @rtype: string

  """

  if instance.tags:

    tags = " ".join(instance.tags)

  else:

    tags = ""

  buf = StringIO()

  sw = utils.ShellWriter(buf)

  sw.Write("#!/bin/sh")

  sw.Write("# this is autogenerated by Ganeti, please do not edit")

  sw.Write("export PATH=$PATH:/sbin:/usr/sbin")

  sw.Write("export INSTANCE=%s", utils.ShellQuote(instance.name))

  sw.Write("export MAC=%s", utils.ShellQuote(nic.mac))

  sw.Write("export MODE=%s",

           utils.ShellQuote(nic.nicparams[constants.NIC_MODE]))

  sw.Write("export INTERFACE=\"$1\"")

  sw.Write("export TAGS=%s", utils.ShellQuote(tags))

  if nic.ip:

    sw.Write("export IP=%s", utils.ShellQuote(nic.ip))

  if nic.nicparams[constants.NIC_LINK]:

    sw.Write("export LINK=%s",

             utils.ShellQuote(nic.nicparams[constants.NIC_LINK]))

  if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:

    sw.Write("export BRIDGE=%s",

             utils.ShellQuote(nic.nicparams[constants.NIC_LINK]))

  # TODO: make this configurable at ./configure time

  sw.Write("if [ -x %s ]; then", utils.ShellQuote(_KVM_NETWORK_SCRIPT))

  sw.IncIndent()

  try:

    sw.Write("# Execute the user-specific vif file")

    sw.Write(_KVM_NETWORK_SCRIPT)

  finally:

    sw.DecIndent()

  sw.Write("else")

  sw.IncIndent()

  try:

    sw.Write("ifconfig $INTERFACE 0.0.0.0 up")

    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:

      sw.Write("# Connect the interface to the bridge")

      sw.Write("brctl addif $BRIDGE $INTERFACE")

    elif nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_ROUTED:

      if not nic.ip:

        raise errors.HypervisorError("nic/%d is routed, but has no IP"

                                     " address" % index)

      sw.Write("# Route traffic targeted at the IP to the interface")

      if nic.nicparams[constants.NIC_LINK]:

        sw.Write("while ip rule del dev $INTERFACE; do :; done")

        sw.Write("ip rule add dev $INTERFACE table $LINK")

        sw.Write("ip route replace $IP table $LINK proto static"

                 " dev $INTERFACE")

      else:

        sw.Write("ip route replace $IP proto static dev $INTERFACE")

      interface_v4_conf = "/proc/sys/net/ipv4/conf/$INTERFACE"

      sw.Write(" if [ -d %s ]; then", interface_v4_conf)

      sw.IncIndent()

      try:

        sw.Write("echo 1 > %s/proxy_arp", interface_v4_conf)

        sw.Write("echo 1 > %s/forwarding", interface_v4_conf)

      finally:

        sw.DecIndent()

      sw.Write("fi")

      interface_v6_conf = "/proc/sys/net/ipv6/conf/$INTERFACE"

      sw.Write("if [ -d %s ]; then", interface_v6_conf)

      sw.IncIndent()

      try:

        sw.Write("echo 1 > %s/proxy_ndp", interface_v6_conf)

        sw.Write("echo 1 > %s/forwarding", interface_v6_conf)

      finally:

        sw.DecIndent()

      sw.Write("fi")

  finally:

    sw.DecIndent()

  sw.Write("fi")

  return buf.getvalue()

  @staticmethod

  def _WriteNetScriptFile(instance, seq, nic):

    script = _WriteNetScript(instance, nic, seq)

      tmpfile.write(script)

        script = self._WriteNetScriptFile(instance, nic_seq, nic)

#!/usr/bin/python

#

# Copyright (C) 2010 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing the hypervisor.hv_kvm module"""

import unittest

from ganeti import constants

from ganeti import compat

from ganeti import objects

from ganeti import errors

from ganeti.hypervisor import hv_kvm

import testutils

class TestWriteNetScript(unittest.TestCase):

  def testBridged(self):

    inst = objects.Instance(name="inst1.example.com", tags=[])

    nic = objects.NIC(mac="01:23:45:67:89:0A",

                      nicparams={

                        constants.NIC_MODE: constants.NIC_MODE_BRIDGED,

                        constants.NIC_LINK: "",

                        })

    script = hv_kvm._WriteNetScript(inst, nic, 0)

    self.assert_(isinstance(script, basestring))

  def testBridgedWithTags(self):

    inst = objects.Instance(name="inst1.example.com", tags=["Hello", "World"])

    nic = objects.NIC(mac="01:23:45:67:89:0A",

                      nicparams={

                        constants.NIC_MODE: constants.NIC_MODE_BRIDGED,

                        constants.NIC_LINK: "",

                        })

    script = hv_kvm._WriteNetScript(inst, nic, 0)

    self.assert_(isinstance(script, basestring))

  def testRouted(self):

    inst = objects.Instance(name="inst2.example.com", tags=[])

    nic = objects.NIC(mac="A0:98:76:54:32:10",

                      ip="192.0.2.4",

                      nicparams={

                        constants.NIC_MODE: constants.NIC_MODE_ROUTED,

                        constants.NIC_LINK: "eth0",

                        })

    script = hv_kvm._WriteNetScript(inst, nic, 4)

    self.assert_(isinstance(script, basestring))

  def testRoutedNoIpAddress(self):

    inst = objects.Instance(name="eiphei1e.example.com", tags=[])

    nic = objects.NIC(mac="93:28:76:54:32:10",

                      nicparams={

                        constants.NIC_MODE: constants.NIC_MODE_ROUTED,

                        constants.NIC_LINK: "",

                        })

    self.assertRaises(errors.HypervisorError, hv_kvm._WriteNetScript,

                      inst, nic, 2)

if __name__ == "__main__":

  testutils.GanetiTestProgram()