Synnefo » snf-ganeti » ganeti-local

  if what.get(constants.NV_FILE_STORAGE_PATH):

    pathresult = filestorage.CheckFileStoragePath(

        what[constants.NV_FILE_STORAGE_PATH])

    if pathresult:

      result[constants.NV_FILE_STORAGE_PATH] = pathresult

  def _VerifyStoragePaths(self, ninfo, nresult):

    """Verifies (file) storage paths.

    @type ninfo: L{objects.Node}

    @param ninfo: the node to check

    @param nresult: the remote results for the node

    """

    cluster = self.cfg.GetClusterInfo()

    if cluster.IsFileStorageEnabled():

      self._ErrorIf(

          constants.NV_FILE_STORAGE_PATH in nresult,

          constants.CV_ENODEFILESTORAGEPATHUNUSABLE, ninfo.name,

          "The configured file storage path is unusable: %s" %

          nresult.get(constants.NV_FILE_STORAGE_PATH))

      self._VerifyStoragePaths(node_i, nresult)

CV_ENODEFILESTORAGEPATHUNUSABLE = \

  (CV_TNODE, "ENODEFILESTORAGEPATHUNUSABLE", "File storage path unusable")

  CV_ENODEFILESTORAGEPATHUNUSABLE,

def _CheckFileStoragePath(path, allowed, exact_match_ok=False):

  @type exact_match_ok: bool

  @param exact_match_ok: whether or not it is okay when the path is exactly

      equal to an allowed path and not a subdir of it

    if exact_match_ok:

      if os.path.normpath(i) == os.path.normpath(path):

        break

    path, _filename=pathutils.FILE_STORAGE_PATHS_FILE,

    exact_match_ok=False):

  _CheckFileStoragePath(path, allowed, exact_match_ok=exact_match_ok)

def _CheckFileStoragePathExistance(path):

  """Checks whether the given path is usable on the file system.

  This checks wether the path is existing, a directory and writable.

  @type path: string

  @param path: path to check

  """

  if not os.path.isdir(path):

    raise errors.FileStoragePathError("Path '%s' is not exisiting or not a"

                                      " directory." % path)

  if not os.access(path, os.W_OK):

    raise errors.FileStoragePathError("Path '%s' is not writable" % path)

def CheckFileStoragePath(

    path, _allowed_paths_file=pathutils.FILE_STORAGE_PATHS_FILE):

  """Checks whether the path exists and is acceptable to use.

  @type path: string

  @param path: path to check

  @rtype: string

  @returns: error message if the path is not ready to use

  """

  try:

    CheckFileStoragePathAcceptance(path, _filename=_allowed_paths_file,

                                   exact_match_ok=True)

  except errors.FileStoragePathError, e:

    return e.message

  if not os.path.isdir(path):

    return "Path '%s' is not exisiting or not a directory." % path

  if not os.access(path, os.W_OK):

    return "Path '%s' is not writable" % path

  [ ("CvECLUSTERCFG",                  'C.cvEclustercfgCode)

  , ("CvECLUSTERCERT",                 'C.cvEclustercertCode)

  , ("CvECLUSTERFILECHECK",            'C.cvEclusterfilecheckCode)

  , ("CvECLUSTERDANGLINGNODES",        'C.cvEclusterdanglingnodesCode)

  , ("CvECLUSTERDANGLINGINST",         'C.cvEclusterdanglinginstCode)

  , ("CvEINSTANCEBADNODE",             'C.cvEinstancebadnodeCode)

  , ("CvEINSTANCEDOWN",                'C.cvEinstancedownCode)

  , ("CvEINSTANCELAYOUT",              'C.cvEinstancelayoutCode)

  , ("CvEINSTANCEMISSINGDISK",         'C.cvEinstancemissingdiskCode)

  , ("CvEINSTANCEFAULTYDISK",          'C.cvEinstancefaultydiskCode)

  , ("CvEINSTANCEWRONGNODE",           'C.cvEinstancewrongnodeCode)

  , ("CvEINSTANCESPLITGROUPS",         'C.cvEinstancesplitgroupsCode)

  , ("CvEINSTANCEPOLICY",              'C.cvEinstancepolicyCode)

  , ("CvENODEDRBD",                    'C.cvEnodedrbdCode)

  , ("CvENODEDRBDHELPER",              'C.cvEnodedrbdhelperCode)

  , ("CvENODEFILECHECK",               'C.cvEnodefilecheckCode)

  , ("CvENODEHOOKS",                   'C.cvEnodehooksCode)

  , ("CvENODEHV",                      'C.cvEnodehvCode)

  , ("CvENODELVM",                     'C.cvEnodelvmCode)

  , ("CvENODEN1",                      'C.cvEnoden1Code)

  , ("CvENODENET",                     'C.cvEnodenetCode)

  , ("CvENODEOS",                      'C.cvEnodeosCode)

  , ("CvENODEORPHANINSTANCE",          'C.cvEnodeorphaninstanceCode)

  , ("CvENODEORPHANLV",                'C.cvEnodeorphanlvCode)

  , ("CvENODERPC",                     'C.cvEnoderpcCode)

  , ("CvENODESSH",                     'C.cvEnodesshCode)

  , ("CvENODEVERSION",                 'C.cvEnodeversionCode)

  , ("CvENODESETUP",                   'C.cvEnodesetupCode)

  , ("CvENODETIME",                    'C.cvEnodetimeCode)

  , ("CvENODEOOBPATH",                 'C.cvEnodeoobpathCode)

  , ("CvENODEUSERSCRIPTS",             'C.cvEnodeuserscriptsCode)

  , ("CvENODEFILESTORAGEPATHS",        'C.cvEnodefilestoragepathsCode)

  , ("CvENODEFILESTORAGEPATHUNUSABLE", 'C.cvEnodefilestoragepathunusableCode)

import os

import shutil

import tempfile

    filestorage.GetFileStorageSpaceInfo("/")

class TestCheckFileStoragePath(unittest.TestCase):

  def _WriteAllowedFile(self, allowed_paths_filename, allowed_paths):

    allowed_paths_file = open(allowed_paths_filename, 'w')

    allowed_paths_file.write('\n'.join(allowed_paths))

    allowed_paths_file.close()

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    self.allowed_paths = [os.path.join(self.tmpdir, "allowed")]

    for path in self.allowed_paths:

      os.mkdir(path)

    self.allowed_paths_filename = os.path.join(self.tmpdir, "allowed-path-file")

    self._WriteAllowedFile(self.allowed_paths_filename, self.allowed_paths)

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def testCheckFileStoragePathExistance(self):

    filestorage._CheckFileStoragePathExistance(self.tmpdir)

  def testCheckFileStoragePathExistanceFail(self):

    path = os.path.join(self.tmpdir, "does/not/exist")

    self.assertRaises(errors.FileStoragePathError,

        filestorage._CheckFileStoragePathExistance, path)

  def testCheckFileStoragePathNotWritable(self):

    path = os.path.join(self.tmpdir, "isnotwritable/")

    os.mkdir(path)

    os.chmod(path, 0)

    self.assertRaises(errors.FileStoragePathError,

        filestorage._CheckFileStoragePathExistance, path)

    os.chmod(path, 777)

  def testCheckFileStoragePath(self):

    path = os.path.join(self.allowed_paths[0], "allowedsubdir")

    os.mkdir(path)

    result = filestorage.CheckFileStoragePath(

        path, _allowed_paths_file=self.allowed_paths_filename)

    self.assertEqual(None, result)

  def testCheckFileStoragePathNotAllowed(self):

    path = os.path.join(self.tmpdir, "notallowed")

    result = filestorage.CheckFileStoragePath(

        path, _allowed_paths_file=self.allowed_paths_filename)

    self.assertTrue("not acceptable" in result)