Synnefo » snf-ganeti » ganeti-local

	lib/runtime.py \

	test/ganeti.runtime_unittest.py \

	  echo "ADMIN_GROUP = '$(ADMIN_GROUP)'"; \

	  echo "MASTERD_GROUP = '$(MASTERD_GROUP)'"; \

	  echo "RAPI_GROUP = '$(RAPI_GROUP)'"; \

	  echo "CONFD_USER = '$(CONFD_USER)'"; \

	  echo "CONFD_GROUP = '$(CONFD_GROUP)'"; \

	  echo "NODED_USER = '$(NODED_USER)'"; \

ADMIN_GROUP = _autoconf.ADMIN_GROUP

MASTERD_GROUP = _autoconf.MASTERD_GROUP

RAPI_GROUP = _autoconf.RAPI_GROUP

CONFD_USER = _autoconf.CONFD_USER

CONFD_GROUP = _autoconf.CONFD_GROUP

NODED_USER = _autoconf.NODED_USER

#

# 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.

"""Module implementing configuration details at runtime.

"""

import grp

import pwd

import threading

from ganeti import constants

from ganeti import errors

_priv = None

_priv_lock = threading.Lock()

def GetUid(user, _getpwnam):

  """Retrieve the uid from the database.

  @type user: string

  @param user: The username to retrieve

  @return: The resolved uid

  """

  try:

    return _getpwnam(user).pw_uid

  except KeyError, err:

    raise errors.ConfigurationError("User '%s' not found (%s)" % (user, err))

def GetGid(group, _getgrnam):

  """Retrieve the gid from the database.

  @type group: string

  @param group: The group name to retrieve

  @return: The resolved gid

  """

  try:

    return _getgrnam(group).gr_gid

  except KeyError, err:

    raise errors.ConfigurationError("Group '%s' not found (%s)" % (group, err))

class GetentResolver:

  """Resolves Ganeti uids and gids by name.

  @ivar masterd_uid: The resolved uid of the masterd user

  @ivar masterd_gid: The resolved gid of the masterd group

  @ivar confd_uid: The resolved uid of the confd user

  @ivar confd_gid: The resolved gid of the confd group

  @ivar rapi_uid: The resolved uid of the rapi user

  @ivar rapi_gid: The resolved gid of the rapi group

  @ivar noded_uid: The resolved uid of the noded user

  @ivar daemons_gid: The resolved gid of the daemons group

  @ivar admin_gid: The resolved gid of the admin group

  """

  def __init__(self, _getpwnam=pwd.getpwnam, _getgrnam=grp.getgrnam):

    """Initialize the resolver.

    """

    # Daemon pairs

    self.masterd_uid = GetUid(constants.MASTERD_USER, _getpwnam)

    self.masterd_gid = GetGid(constants.MASTERD_GROUP, _getgrnam)

    self.confd_uid = GetUid(constants.CONFD_USER, _getpwnam)

    self.confd_gid = GetGid(constants.CONFD_GROUP, _getgrnam)

    self.rapi_uid = GetUid(constants.RAPI_USER, _getpwnam)

    self.rapi_gid = GetGid(constants.RAPI_GROUP, _getgrnam)

    self.noded_uid = GetUid(constants.NODED_USER, _getpwnam)

    # Misc Ganeti groups

    self.daemons_gid = GetGid(constants.DAEMONS_GROUP, _getgrnam)

    self.admin_gid = GetGid(constants.ADMIN_GROUP, _getgrnam)

def GetEnts(resolver=GetentResolver):

  """Singleton wrapper around resolver instance.

  As this method is accessed by multiple threads at the same time

  we need to take thread-safty carefully

  """

  # We need to use the global keyword here

  global _priv # pylint: disable-msg=W0603

  if not _priv:

    _priv_lock.acquire()

    try:

      if not _priv:

        # W0621: Redefine '_priv' from outer scope (used for singleton)

        _priv = resolver() # pylint: disable-msg=W0621

    finally:

      _priv_lock.release()

  return _priv

#!/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 ganeti.runtime"""

from ganeti import constants

from ganeti import errors

from ganeti import runtime

import testutils

class _EntStub:

  def __init__(self, uid=None, gid=None):

    self.pw_uid = uid

    self.gr_gid = gid

def _StubGetpwnam(user):

  users = {

    constants.MASTERD_USER: _EntStub(uid=0),

    constants.CONFD_USER: _EntStub(uid=1),

    constants.RAPI_USER: _EntStub(uid=2),

    constants.NODED_USER: _EntStub(uid=3),

    }

  return users[user]

def _StubGetgrnam(group):

  groups = {

    constants.MASTERD_GROUP: _EntStub(gid=0),

    constants.CONFD_GROUP: _EntStub(gid=1),

    constants.RAPI_GROUP: _EntStub(gid=2),

    constants.DAEMONS_GROUP: _EntStub(gid=3),

    constants.ADMIN_GROUP: _EntStub(gid=4),

    }

  return groups[group]

def _RaisingStubGetpwnam(user):

  raise KeyError("user not found")

def _RaisingStubGetgrnam(group):

  raise KeyError("group not found")

class ResolverStubRaising(object):

  def __init__(self):

    raise errors.ConfigurationError("No entries")

class TestErrors(testutils.GanetiTestCase):

  def testEverythingSuccessful(self):

    resolver = runtime.GetentResolver(_getpwnam=_StubGetpwnam,

                                      _getgrnam=_StubGetgrnam)

    self.assertEqual(resolver.masterd_uid,

                     _StubGetpwnam(constants.MASTERD_USER).pw_uid)

    self.assertEqual(resolver.masterd_gid,

                     _StubGetgrnam(constants.MASTERD_GROUP).gr_gid)

    self.assertEqual(resolver.confd_uid,

                     _StubGetpwnam(constants.CONFD_USER).pw_uid)

    self.assertEqual(resolver.confd_gid,

                     _StubGetgrnam(constants.CONFD_GROUP).gr_gid)

    self.assertEqual(resolver.rapi_uid,

                     _StubGetpwnam(constants.RAPI_USER).pw_uid)

    self.assertEqual(resolver.rapi_gid,

                     _StubGetgrnam(constants.RAPI_GROUP).gr_gid)

    self.assertEqual(resolver.noded_uid,

                     _StubGetpwnam(constants.NODED_USER).pw_uid)

    self.assertEqual(resolver.daemons_gid,

                     _StubGetgrnam(constants.DAEMONS_GROUP).gr_gid)

    self.assertEqual(resolver.admin_gid,

                     _StubGetgrnam(constants.ADMIN_GROUP).gr_gid)

  def testUserNotFound(self):

    self.assertRaises(errors.ConfigurationError, runtime.GetentResolver,

                      _getpwnam=_RaisingStubGetpwnam, _getgrnam=_StubGetgrnam)

  def testGroupNotFound(self):

    self.assertRaises(errors.ConfigurationError, runtime.GetentResolver,

                      _getpwnam=_StubGetpwnam, _getgrnam=_RaisingStubGetgrnam)

  def testUserNotFoundGetEnts(self):

    self.assertRaises(errors.ConfigurationError, runtime.GetEnts,

                      resolver=ResolverStubRaising)

if __name__ == "__main__":

  testutils.GanetiTestProgram()