blob: 578d812cc0fe801de9fb291cb17a8343ffc6aa16 [file] [log] [blame]
#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
#Unless required by applicable law or agreed to in writing, software
#distributed under the License is distributed on an "AS IS" BASIS,
#See the License for the specific language governing permissions and
#limitations under the License.
import unittest, re, sys
class BaseTestSuite():
def __init__(self, name, excludes): = name
self.excludes = excludes
def runTests(self):
# Create a runner
self.runner = unittest.TextTestRunner()
# Get all the test-case classes
# From module import *
mod = __import__(, fromlist=['*'])
modItemsList = dir(mod)
allsuites = []
# Create all the test suites
for modItem in modItemsList:
if"^test_", modItem):
# Yes this is a test class
if modItem not in self.excludes:
test_class = getattr(mod, modItem)
# Create a master suite to be run.
alltests = unittest.TestSuite(tuple(allsuites))
# Run the master test suite.
runner =
if(runner.wasSuccessful()): return 0
printLine( "%s test(s) failed." % runner.failures.__len__())
printLine( "%s test(s) threw errors." % runner.errors.__len__())
return runner.failures.__len__() + runner.errors.__len__()
def cleanUp(self):
# suite tearDown
def printLine(str):
print >>sys.stderr, str
def printSeparator():
str = ""
for i in range(0,79):
str = str + "*"
print >>sys.stderr, "\n", str, "\n"
# This class captures all log messages logged by hodRunner and other classes.
# It is then used to verify that certain log messages have come. This is one
# way to validate that messages printed to the logger are correctly written.
class MockLogger:
def __init__(self):
self.__logLines = {}
def info(self, message):
self.__logLines[message] = 'info'
def critical(self, message):
self.__logLines[message] = 'critical'
def warn(self, message):
self.__logLines[message] = 'warn'
def debug(self, message):
# don't track debug lines.
# verify a certain message has been logged at the defined level of severity.
def hasMessage(self, message, level):
if not self.__logLines.has_key(message):
return False
return self.__logLines[message] == level
# Stub class to test cluster manipulation operations.
class MockHadoopCluster:
def __init__(self):
# store the operations received.
self.__operations = {}
def delete_job(self, jobid):
self.__operations['delete_job'] = [jobid]
def is_cluster_deallocated(self, dummy):
return False
def wasOperationPerformed(self, operation, args):
if self.__operations.has_key(operation):
actualArgs = self.__operations[operation]
for arg in actualArgs:
if arg not in args:
return True
return False