testhod.py

HADOOP 0.18.0安装源代码头文件

#Licensed to the Apache Software Foundation (ASF) under one
#or more contributor license agreements.  See the NOTICE file
#distributed with this work for additional information
#regarding copyright ownership.  The ASF licenses this file
#to you under the Apache License, Version 2.0 (the
#"License"); you may not use this file except in compliance
#with the License.  You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

#Unless required by applicable law or agreed to in writing, software
#distributed under the License is distributed on an "AS IS" BASIS,
#WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#See the License for the specific language governing permissions and
#limitations under the License.
import unittest, getpass, os, sys, re, threading, time

myDirectory = os.path.realpath(sys.argv[0])
rootDirectory   = re.sub("/testing/.*", "", myDirectory)

sys.path.append(rootDirectory)

import tempfile
from testing.lib import BaseTestSuite, MockLogger, MockHadoopCluster
from hodlib.Hod.hod import hodRunner, hodState
from hodlib.Common.desc import NodePoolDesc

excludes = []

# Information about all clusters is written to a file called clusters.state.
from hodlib.Hod.hod import CLUSTER_DATA_FILE as TEST_CLUSTER_DATA_FILE, \
                           INVALID_STATE_FILE_MSGS

# Temp directory prefix
TMP_DIR_PREFIX=os.path.join('/tmp', 'hod-%s' % (getpass.getuser()))

# build a config object with all required keys for initializing hod.
def setupConf():
  cfg = {
          'hod' : {
                    'original-dir' : os.getcwd(),
                    'stream' : True,
                    # store all the info about clusters in this directory
                    'user_state' : '/tmp/hodtest',
                    'debug' : 3,
                    'java-home' : os.getenv('JAVA_HOME'),
                    'cluster' : 'dummy',
                    'cluster-factor' : 1.8,
                    'xrs-port-range' : (32768,65536),
                    'allocate-wait-time' : 3600,
                    'temp-dir' : '/tmp/hod'
                  },
          # just set everything to dummy. Need something to initialize the
          # node pool description object.
          'resource_manager' : {
                                 'id' : 'dummy',
                                 'batch-home' : 'dummy',
                                 'queue' : 'dummy',
                               }
        }
  cfg['nodepooldesc'] = NodePoolDesc(cfg['resource_manager'])
  return cfg

# Test class that defines methods to test invalid arguments to hod operations.
class test_InvalidArgsOperations(unittest.TestCase):
  def setUp(self):

    self.cfg = setupConf()
    # initialize the mock objects
    self.log = MockLogger()
    self.cluster = MockHadoopCluster()

    # Use the test logger. This will be used for test verification.
    self.client = hodRunner(self.cfg, log=self.log, cluster=self.cluster)
    # Create the hodState object to set the test state you want.
    self.state = hodState(self.cfg['hod']['user_state'])
    if not os.path.exists(self.cfg['hod']['user_state']):
      os.path.mkdir(self.cfg['hod']['user_state'])
    p = os.path.join(self.cfg['hod']['user_state'], '%s.state' % TEST_CLUSTER_DATA_FILE)
    # ensure cluster data file exists, so write works in the tests.
    f = open(p, 'w')
    f.close()
  
  def tearDown(self):
    # clean up cluster data file and directory
    p = os.path.join(self.cfg['hod']['user_state'], '%s.state' % TEST_CLUSTER_DATA_FILE)
    os.remove(p)
    os.rmdir(self.cfg['hod']['user_state'])

  # Test that list works with deleted cluster directories - more than one entries which are invalid.
  def testListInvalidDirectory(self):
    userState = { os.path.join(TMP_DIR_PREFIX, 'testListInvalidDirectory1') : '123.dummy.id1', 
                  os.path.join(TMP_DIR_PREFIX, 'testListInvalidDirectory2') : '123.dummy.id2' }
    self.__setupClusterState(userState)
    self.client._op_list(['list'])
    # assert that required errors are logged.
    for clusterDir in userState.keys():
      self.assertTrue(self.log.hasMessage('cluster state unknown\t%s\t%s' \
                            % (userState[clusterDir], clusterDir), 'info'))

    # simulate a test where a directory is deleted, and created again, without deallocation
    clusterDir = os.path.join(TMP_DIR_PREFIX, 'testListEmptyDirectory')
    os.makedirs(clusterDir)
    self.assertTrue(os.path.isdir(clusterDir))
    userState = { clusterDir : '123.dummy.id3' }
    self.__setupClusterState(userState, False)
    self.client._op_list(['list'])
    self.assertTrue(self.log.hasMessage('cluster state unknown\t%s\t%s' \
                          % (userState[clusterDir], clusterDir), 'info'))
    os.rmdir(clusterDir)
    
  # Test that info works with a deleted cluster directory
  def testInfoInvalidDirectory(self):
    clusterDir = os.path.join(TMP_DIR_PREFIX, 'testInfoInvalidDirectory')
    userState = { clusterDir : '456.dummy.id' }
    self.__setupClusterState(userState)
    self.client._op_info(['info', clusterDir])
    self.assertTrue(self.log.hasMessage("Cannot find information for cluster with id '%s' in previously allocated cluster directory '%s'." % (userState[clusterDir], clusterDir), 'critical'))

    # simulate a test where a directory is deleted, and created again, without deallocation
    clusterDir = os.path.join(TMP_DIR_PREFIX, 'testInfoEmptyDirectory')
    os.makedirs(clusterDir)
    self.assertTrue(os.path.isdir(clusterDir))
    userState = { clusterDir : '456.dummy.id1' }
    self.__setupClusterState(userState, False)
    self.client._op_info(['info', clusterDir])
    self.assertTrue(self.log.hasMessage("Cannot find information for cluster with id '%s' in previously allocated cluster directory '%s'." % (userState[clusterDir], clusterDir), 'critical'))
    os.rmdir(clusterDir)

  # Test info works with an invalid cluster directory
  def testInfoNonExistentDirectory(self):
    clusterDir = '/tmp/hod/testInfoNonExistentDirectory'
    self.client._op_info(['info', clusterDir])
    self.assertTrue(self.log.hasMessage("Invalid hod.clusterdir(--hod.clusterdir or -d). %s : No such directory" % (clusterDir), 'critical'))

  # Test that deallocation works on a deleted cluster directory
  # by clearing the job, and removing the state
  def testDeallocateInvalidDirectory(self):
    clusterDir = os.path.join(TMP_DIR_PREFIX,'testDeallocateInvalidDirectory')
    jobid = '789.dummy.id'
    userState = { clusterDir : jobid }
    self.__setupClusterState(userState)
    self.client._op_deallocate(['deallocate', clusterDir])
    # verify job was deleted
    self.assertTrue(self.cluster.wasOperationPerformed('delete_job', jobid))
    # verify appropriate message was logged.
    self.assertTrue(self.log.hasMessage("Cannot find information for cluster with id '%s' in previously allocated cluster directory '%s'." % (userState[clusterDir], clusterDir), 'critical'))
    self.assertTrue(self.log.hasMessage("Freeing resources allocated to the cluster.", 'critical'))
    # verify that the state information was cleared.
    userState = self.state.read(TEST_CLUSTER_DATA_FILE)
    self.assertFalse(clusterDir in userState.keys())
 
    # simulate a test where a directory is deleted, and created again, without deallocation
    clusterDir = os.path.join(TMP_DIR_PREFIX,'testDeallocateEmptyDirectory')
    os.makedirs(clusterDir)
    self.assertTrue(os.path.isdir(clusterDir))
    jobid = '789.dummy.id1'
    userState = { clusterDir : jobid }
    self.__setupClusterState(userState, False)
    self.client._op_deallocate(['deallocate', clusterDir])
    # verify job was deleted
    self.assertTrue(self.cluster.wasOperationPerformed('delete_job', jobid))
    # verify appropriate message was logged.
    self.assertTrue(self.log.hasMessage("Cannot find information for cluster with id '%s' in previously allocated cluster directory '%s'." % (userState[clusterDir], clusterDir), 'critical'))
    self.assertTrue(self.log.hasMessage("Freeing resources allocated to the cluster.", 'critical'))
    # verify that the state information was cleared.
    userState = self.state.read(TEST_CLUSTER_DATA_FILE)
    self.assertFalse(clusterDir in userState.keys())
    os.rmdir(clusterDir)
     
  # Test that deallocation works on a nonexistent directory.
  def testDeallocateNonExistentDirectory(self):
    clusterDir = os.path.join(TMP_DIR_PREFIX,'testDeallocateNonExistentDirectory')
    self.client._op_deallocate(['deallocate', clusterDir])
    # there should be no call..
    self.assertFalse(self.cluster.wasOperationPerformed('delete_job', None))
    self.assertTrue(self.log.hasMessage("Invalid hod.clusterdir(--hod.clusterdir or -d). %s : No such directory" % (clusterDir), 'critical'))

  # Test that allocation on an previously deleted directory fails.    
  def testAllocateOnDeletedDirectory(self):
    clusterDir = os.path.join(TMP_DIR_PREFIX, 'testAllocateOnDeletedDirectory')
    os.makedirs(clusterDir)
    self.assertTrue(os.path.isdir(clusterDir))
    jobid = '1234.abc.com'
    userState = { clusterDir : jobid }
    self.__setupClusterState(userState, False)
    self.client._op_allocate(['allocate', clusterDir, '3'])
    self.assertTrue(self.log.hasMessage("Found a previously allocated cluster at "\
                      "cluster directory '%s'. HOD cannot determine if this cluster "\
                      "can be automatically deallocated. Deallocate the cluster if it "\
                      "is unused." % (clusterDir), 'critical'))
    os.rmdir(clusterDir)

  def __setupClusterState(self, clusterStateMap, verifyDirIsAbsent=True):
    for clusterDir in clusterStateMap.keys():
      # ensure directory doesn't exist, just in case.
      if verifyDirIsAbsent:
        self.assertFalse(os.path.exists(clusterDir))
    # set up required state.
    self.state.write(TEST_CLUSTER_DATA_FILE, clusterStateMap)
    # verify everything is stored correctly.
    state = self.state.read(TEST_CLUSTER_DATA_FILE)
    for clusterDir in clusterStateMap.keys():
      self.assertTrue(clusterDir in state.keys())
      self.assertEquals(clusterStateMap[clusterDir], state[clusterDir])

class test_InvalidHodStateFiles(unittest.TestCase):
  def setUp(self):
    self.rootDir = '/tmp/hod-%s' % getpass.getuser()
    self.cfg = setupConf() # creat a conf
    # Modify hod.user_state
    self.cfg['hod']['user_state'] = tempfile.mkdtemp(dir=self.rootDir,
                              prefix='HodTestSuite.test_InvalidHodStateFiles_')
    self.log = MockLogger() # mock logger
    self.cluster = MockHadoopCluster() # mock hadoop cluster
    self.client = hodRunner(self.cfg, log=self.log, cluster=self.cluster)
    self.state = hodState(self.cfg['hod']['user_state'])
    self.statePath = os.path.join(self.cfg['hod']['user_state'], '%s.state' % \
                                  TEST_CLUSTER_DATA_FILE)
    self.clusterDir = tempfile.mkdtemp(dir=self.rootDir,
                              prefix='HodTestSuite.test_InvalidHodStateFiles_')
  
  def testOperationWithInvalidStateFile(self):
    jobid = '1234.hadoop.apache.org'
    # create user state file with invalid permissions
    stateFile = open(self.statePath, "w")
    os.chmod(self.statePath, 000) # has no read/write permissions
    self.client._hodRunner__cfg['hod']['operation'] = \
                                             "info %s" % self.clusterDir
    ret = self.client.operation()
    os.chmod(self.statePath, 700) # restore permissions
    stateFile.close()
    os.remove(self.statePath)

    # print self.log._MockLogger__logLines
    self.assertTrue(self.log.hasMessage(INVALID_STATE_FILE_MSGS[0] % \
                          os.path.realpath(self.statePath), 'critical'))
    self.assertEquals(ret, 1)
    
  def testAllocateWithInvalidStateFile(self):
    jobid = '1234.hadoop.apache.org'
    # create user state file with invalid permissions
    stateFile = open(self.statePath, "w")
    os.chmod(self.statePath, 0400) # has no write permissions
    self.client._hodRunner__cfg['hod']['operation'] = \
                                        "allocate %s %s" % (self.clusterDir, '3')
    ret = self.client.operation()
    os.chmod(self.statePath, 700) # restore permissions
    stateFile.close()
    os.remove(self.statePath)

    # print self.log._MockLogger__logLines
    self.assertTrue(self.log.hasMessage(INVALID_STATE_FILE_MSGS[2] % \
                        os.path.realpath(self.statePath), 'critical'))
    self.assertEquals(ret, 1)
  
  def testAllocateWithInvalidStateStore(self):
    jobid = '1234.hadoop.apache.org'
    self.client._hodRunner__cfg['hod']['operation'] = \
                                      "allocate %s %s" % (self.clusterDir, 3)

    ###### check with no executable permissions ######
    stateFile = open(self.statePath, "w") # create user state file
    os.chmod(self.cfg['hod']['user_state'], 0600) 
    ret = self.client.operation()
    os.chmod(self.cfg['hod']['user_state'], 0700) # restore permissions
    stateFile.close()
    os.remove(self.statePath)
    # print self.log._MockLogger__logLines
    self.assertTrue(self.log.hasMessage(INVALID_STATE_FILE_MSGS[0] % \
                          os.path.realpath(self.statePath), 'critical'))
    self.assertEquals(ret, 1)
    
    ###### check with no write permissions ######
    stateFile = open(self.statePath, "w") # create user state file
    os.chmod(self.cfg['hod']['user_state'], 0500) 
    ret = self.client.operation()
    os.chmod(self.cfg['hod']['user_state'], 0700) # restore permissions
    stateFile.close()
    os.remove(self.statePath)
    # print self.log._MockLogger__logLines
    self.assertTrue(self.log.hasMessage(INVALID_STATE_FILE_MSGS[0] % \
                          os.path.realpath(self.statePath), 'critical'))
    self.assertEquals(ret, 1)

  def tearDown(self):
    if os.path.exists(self.clusterDir): os.rmdir(self.clusterDir)
    if os.path.exists(self.cfg['hod']['user_state']):
      os.rmdir(self.cfg['hod']['user_state'])


class HodTestSuite(BaseTestSuite):
  def __init__(self):
    # suite setup
    BaseTestSuite.__init__(self, __name__, excludes)
    pass
  
  def cleanUp(self):
    # suite tearDown
    pass

def RunHodTests():
  # modulename_suite
  suite = HodTestSuite()
  testResult = suite.runTests()
  suite.cleanUp()
  return testResult

if __name__ == "__main__":
  RunHodTests()