Synnefo » snf-ganeti » ganeti-local

	test/ganeti.client.gnt_instance_unittest.py \

import logging

from ganeti import ssh

from ganeti import objects

  cl = GetClient()

  try:

    cluster_name = cl.QueryConfigValues(["cluster_name"])[0]

    console_data = SubmitOpCode(op, opts=opts, cl=cl)

  finally:

    # Ensure client connection is closed while external commands are run

    cl.Close()

  del cl

  return _DoConsole(objects.InstanceConsole.FromDict(console_data),

                    opts.show_command, cluster_name)

def _DoConsole(console, show_command, cluster_name, feedback_fn=ToStdout,

               _runcmd_fn=utils.RunCmd):

  """Acts based on the result of L{opcodes.OpConnectConsole}.

  @type console: L{objects.InstanceConsole}

  @param console: Console object

  @type show_command: bool

  @param show_command: Whether to just display commands

  @type cluster_name: string

  @param cluster_name: Cluster name as retrieved from master daemon

  """

  assert console.Validate()

  if console.kind == constants.CONS_MESSAGE:

    feedback_fn(console.message)

  elif console.kind == constants.CONS_VNC:

    feedback_fn("Instance %s has VNC listening on %s:%s (display %s),"

                " URL <vnc://%s:%s/>",

                console.instance, console.host, console.port,

                console.display, console.host, console.port)

  elif console.kind == constants.CONS_SSH:

    # Convert to string if not already one

    if isinstance(console.command, basestring):

      cmd = console.command

    else:

      cmd = utils.ShellQuoteArgs(console.command)

    srun = ssh.SshRunner(cluster_name=cluster_name)

    ssh_cmd = srun.BuildCmd(console.host, console.user, cmd,

                            batch=True, quiet=False, tty=True)

    if show_command:

      feedback_fn(utils.ShellQuoteArgs(ssh_cmd))

    else:

      result = _runcmd_fn(ssh_cmd, interactive=True)

      if result.failed:

        logging.error("Console command \"%s\" failed with reason '%s' and"

                      " output %r", result.cmd, result.fail_reason,

                      result.output)

        raise errors.OpExecError("Connection to console of instance %s failed,"

                                 " please check cluster configuration" %

                                 console.instance)

    raise errors.GenericError("Unknown console type '%s'" % console.kind)

    console = objects.InstanceConsole(instance=instance.name,

                                      kind=constants.CONS_SSH,

                                      host=node,

                                      user="root",

                                      command=console_cmd)

    assert console.Validate()

    return console.ToDict()

#: Console as SSH command

CONS_SSH = "ssh"

#: Console as VNC server

CONS_VNC = "vnc"

#: Display a message for console access

CONS_MESSAGE = "msg"

#: All console types

CONS_ALL = frozenset([CONS_SSH, CONS_VNC, CONS_MESSAGE])

class InstanceConsole(ConfigObject):

  """Object describing how to access the console of an instance.

  """

  __slots__ = [

    "instance",

    "kind",

    "message",

    "host",

    "port",

    "user",

    "command",

    "display",

    ]

  def Validate(self):

    """Validates contents of this object.

    """

    assert self.kind in constants.CONS_ALL, "Unknown console type"

    assert self.instance, "Missing instance name"

    assert self.message or self.kind in [constants.CONS_SSH, constants.CONS_VNC]

    assert self.host or self.kind == constants.CONS_MESSAGE

    assert self.port or self.kind in [constants.CONS_MESSAGE,

                                      constants.CONS_SSH]

    assert self.user or self.kind in [constants.CONS_MESSAGE,

                                      constants.CONS_VNC]

    assert self.command or self.kind in [constants.CONS_MESSAGE,

                                         constants.CONS_VNC]

    assert self.display or self.kind in [constants.CONS_MESSAGE,

                                         constants.CONS_SSH]

    return True

#!/usr/bin/python

#

# Copyright (C) 2011 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 ganeti.client.gnt_instance"""

import unittest

from ganeti import constants

from ganeti import utils

from ganeti import errors

from ganeti import objects

from ganeti.client import gnt_instance

import testutils

class TestConsole(unittest.TestCase):

  def setUp(self):

    self._output = []

    self._cmds = []

    self._next_cmd_exitcode = 0

  def _Test(self, *args):

    return gnt_instance._DoConsole(*args,

                                   feedback_fn=self._Feedback,

                                   _runcmd_fn=self._FakeRunCmd)

  def _Feedback(self, msg, *args):

    if args:

      msg = msg % args

    self._output.append(msg)

  def _FakeRunCmd(self, cmd, interactive=None):

    self.assertTrue(interactive)

    self.assertTrue(isinstance(cmd, list))

    self._cmds.append(cmd)

    return utils.RunResult(self._next_cmd_exitcode, None, "", "", "cmd",

                           utils._TIMEOUT_NONE, 5)

  def testMessage(self):

    cons = objects.InstanceConsole(instance="inst98.example.com",

                                   kind=constants.CONS_MESSAGE,

                                   message="Hello World")

    self.assertEqual(self._Test(cons, False, "cluster.example.com"),

                     constants.EXIT_SUCCESS)

    self.assertEqual(len(self._cmds), 0)

    self.assertEqual(self._output, ["Hello World"])

  def testVnc(self):

    cons = objects.InstanceConsole(instance="inst1.example.com",

                                   kind=constants.CONS_VNC,

                                   host="node1.example.com",

                                   port=5901,

                                   display=1)

    self.assertEqual(self._Test(cons, False, "cluster.example.com"),

                     constants.EXIT_SUCCESS)

    self.assertEqual(len(self._cmds), 0)

    self.assertEqual(len(self._output), 1)

    self.assertTrue(" inst1.example.com " in self._output[0])

    self.assertTrue(" node1.example.com:5901 " in self._output[0])

    self.assertTrue("vnc://node1.example.com:5901/" in self._output[0])

  def testSshShow(self):

    cons = objects.InstanceConsole(instance="inst31.example.com",

                                   kind=constants.CONS_SSH,

                                   host="node93.example.com",

                                   user="user_abc",

                                   command="xm console x.y.z")

    self.assertEqual(self._Test(cons, True, "cluster.example.com"),

                     constants.EXIT_SUCCESS)

    self.assertEqual(len(self._cmds), 0)

    self.assertEqual(len(self._output), 1)

    self.assertTrue(" user_abc@node93.example.com " in self._output[0])

    self.assertTrue("'xm console x.y.z'" in self._output[0])

  def testSshRun(self):

    cons = objects.InstanceConsole(instance="inst31.example.com",

                                   kind=constants.CONS_SSH,

                                   host="node93.example.com",

                                   user="user_abc",

                                   command=["xm", "console", "x.y.z"])

    self.assertEqual(self._Test(cons, False, "cluster.example.com"),

                     constants.EXIT_SUCCESS)

    self.assertEqual(len(self._cmds), 1)

    self.assertEqual(len(self._output), 0)

    # This is very important to prevent escapes from the console

    self.assertTrue("-oEscapeChar=none" in self._cmds[0])

  def testSshRunFail(self):

    cons = objects.InstanceConsole(instance="inst31.example.com",

                                   kind=constants.CONS_SSH,

                                   host="node93.example.com",

                                   user="user_abc",

                                   command=["xm", "console", "x.y.z"])

    self._next_cmd_exitcode = 100

    self.assertRaises(errors.OpExecError, self._Test,

                      cons, False, "cluster.example.com")

    self.assertEqual(len(self._cmds), 1)

    self.assertEqual(len(self._output), 0)

if __name__ == "__main__":

  testutils.GanetiTestProgram()