#! /usr/bin/env python # # Copyright (C) 2009 Nexedi SA # # 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. import unittest import tempfile import logging import time import os # list of test modules # each of them have to import its TestCase classes UNIT_TEST_MODULES = [ # generic parts 'neo.tests.testBootstrap', 'neo.tests.testConnection', 'neo.tests.testEvent', 'neo.tests.testHandler', 'neo.tests.testNodes', 'neo.tests.testProtocol', 'neo.tests.testPT', # master application 'neo.tests.master.testClientHandler', 'neo.tests.master.testElectionHandler', 'neo.tests.master.testMasterApp', 'neo.tests.master.testMasterPT', 'neo.tests.master.testRecoveryHandler', 'neo.tests.master.testStorageHandler', 'neo.tests.master.testVerificationHandler', # storage application 'neo.tests.storage.testClientHandler', 'neo.tests.storage.testInitializationHandler', 'neo.tests.storage.testMasterHandler', 'neo.tests.storage.testStorageApp', 'neo.tests.storage.testStorageHandler', 'neo.tests.storage.testStorageMySQLdb', 'neo.tests.storage.testVerificationHandler', # client application 'neo.tests.client.testClientApp', 'neo.tests.client.testClientHandler', 'neo.tests.client.testConnectionPool', 'neo.tests.client.testDispatcher', ] FUNC_TEST_MODULES = [ ('neo.tests.functional.testZODB', 'check'), 'neo.tests.functional.testMaster', 'neo.tests.functional.testCluster', 'neo.tests.functional.testStorage', 'neo.tests.functional.testImportExport', ] # configuration UNIT_TESTS = True FUNCTIONAL_TESTS = True FUNCTIONAL_TESTS = False SEND_REPORT = True SEND_REPORT = False CONSOLE_LOG = False ATTACH_LOG = False # for ZODB test, only the client side is logged LOG_FILE = 'neo.log' SENDER = 'gregory@nexedi.com' RECIPIENTS = ['gregory@nexedi.com'] #['neo-report@erp5.org'] SMTP_SERVER = ( "mail.nexedi.com", "25") # override logging configuration to send all messages to a file logger = logging.getLogger() logger.setLevel(logging.INFO) handler = logging.FileHandler(LOG_FILE, 'w+') format='[%(module)12s:%(levelname)s:%(lineno)3d] %(message)s' formatter = logging.Formatter(format) handler.setFormatter(formatter) logger.addHandler(handler) # enabled console logging if desired if CONSOLE_LOG: handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) class NeoTestRunner(unittest.TestResult): """ Custom result class to build report with statistics per module """ def __init__(self): unittest.TestResult.__init__(self) self.modulesStats = {} self.failedImports = {} self.lastStart = None self.temp_directory = tempfile.mkdtemp(prefix='neo_') os.environ['TEMP'] = self.temp_directory print "Base directory : %s" % (self.temp_directory, ) def run(self, name, modules): suite = unittest.TestSuite() loader = unittest.defaultTestLoader for test_module in modules: # load prefix if supplied if isinstance(test_module, tuple): test_module, prefix = test_module loader.testMethodPrefix = prefix else: loader.testMethodPrefix = 'test' try: test_module = __import__(test_module, globals(), locals(), ['*']) except ImportError, err: # TODO: include import errors in report self.failedImports[test_module] = err print "Import of %s failed : %s" % (test_module, err) continue suite.addTests(loader.loadTestsFromModule(test_module)) suite.run(self) class ModuleStats(object): run = 0 errors = 0 success = 0 failures = 0 time = 0.0 def _getModuleStats(self, test): module = test.__class__.__module__ module = tuple(module.split('.')) try: return self.modulesStats[module] except KeyError: self.modulesStats[module] = self.ModuleStats() return self.modulesStats[module] def _updateTimer(self, stats): stats.time += time.time() - self.lastStart def startTest(self, test): print test.__class__.__module__, test._TestCase__testMethodName unittest.TestResult.startTest(self, test) module = test.__class__.__name__ method = test._TestCase__testMethodName logging.info(" * TEST %s" % test) stats = self._getModuleStats(test) stats.run += 1 self.lastStart = time.time() def addSuccess(self, test): unittest.TestResult.addSuccess(self, test) stats = self._getModuleStats(test) stats.success += 1 self._updateTimer(stats) def addError(self, test, err): unittest.TestResult.addError(self, test, err) stats = self._getModuleStats(test) stats.errors += 1 self._updateTimer(stats) def addFailure(self, test, err): unittest.TestResult.addFailure(self, test, err) stats = self._getModuleStats(test) stats.failures += 1 self._updateTimer(stats) def _buildSystemInfo(self): import platform import datetime success = self.testsRun - len(self.errors) - len(self.failures) s = """ Date : %s Node : %s Processor : %s (%s) System : %s (%s) Directory : %s Status : %7.3f%% """ % ( datetime.date.today().isoformat(), platform.node(), platform.processor(), platform.architecture()[0], platform.system(), platform.release(), self.temp_directory, success * 100.0 / self.testsRun, ) return s def _buildSummary(self): # visual header = "%25s | run | success | errors | fails | time \n" % 'Test Module' separator = "%25s-+---------+---------+---------+---------+----------\n" % ('-' * 25) format = "%25s | %3s | %3s | %3s | %3s | %6.2fs \n" group_f = "%25s | | | | | \n" # header s = ' ' * 30 + ' NEO TESTS REPORT' s += '\n\n' s += self._buildSystemInfo() s += '\n' + header + separator group = None t_success = 0 # for each test case for k, v in sorted(self.modulesStats.items()): # display group below its content _group = '.'.join(k[:-1]) if group is None: group = _group if _group != group: s += separator + group_f % group + separator group = _group # test case stats t_success += v.success run, success = v.run or '.', v.success or '.' errors, failures = v.errors or '.', v.failures or '.' name = k[-1].lstrip('test') args = (name, run, success, errors, failures, v.time) s += format % args # the last group s += separator + group_f % group + separator # the final summary errors, failures = len(self.errors) or '.', len(self.failures) or '.' args = ("Summary", self.testsRun, t_success, errors, failures, self.time) s += format % args + separator + '\n' return s def _buildErrors(self): s = '\n' if len(self.errors): s += '\nERRORS:\n' for test, trace in self.errors: s += "%s\n" % test s += "-------------------------------------------------------------\n" s += trace s += "-------------------------------------------------------------\n" s += '\n' if len(self.failures): s += '\nFAILURES:\n' for test, trace in self.failures: s += "%s\n" % test s += "-------------------------------------------------------------\n" s += trace s += "-------------------------------------------------------------\n" s += '\n' return s def _buildWarnings(self): s = '\n' if self.failedImports: s += 'Failed imports :\n' for module, err in self.failedImports.items(): s += '%s:\n%s' % (module, err) s += '\n' return s def build(self): self.time = sum([s.time for s in self.modulesStats.values()]) args = (self.testsRun, len(self.errors), len(self.failures)) self.subject = "Neo : %s Tests, %s Errors, %s Failures" % args self.summary = self._buildSummary() self.errors = self._buildErrors() self.warnings = self._buildWarnings() def sendReport(self): """ Send a mail with the report summary """ import smtplib from email.MIMEMultipart import MIMEMultipart from email.MIMEText import MIMEText # build the email msg = MIMEMultipart() msg['Subject'] = self.subject msg['From'] = SENDER msg['To'] = ', '.join(RECIPIENTS) #msg.preamble = self.subject msg.epilogue = '' # Add custom headers for client side filtering msg['X-ERP5-Tests'] = 'NEO' if self.wasSuccessful(): msg['X-ERP5-Tests-Status'] = 'OK' # write the body body = MIMEText(self.summary + self.warnings + self.errors) msg.attach(body) # attach the log file if ATTACH_LOG: log = MIMEText(file(LOG_FILE, 'r').read()) log.add_header('Content-Disposition', 'attachment', filename=LOG_FILE) msg.attach(log) # Send the email via our own SMTP server. s = smtplib.SMTP() s.connect(*SMTP_SERVER) mail = msg.as_string() for recipient in RECIPIENTS: try: s.sendmail(SENDER, recipient, mail) except smtplib.SMTPRecipientsRefused: print "Mail for %s fails" % recipient s.close() if __name__ == "__main__": if not UNIT_TESTS and not FUNCTIONAL_TESTS: raise RuntimeError('Nothing to run') # run and build the report runner = NeoTestRunner() if UNIT_TESTS: runner.run('Unit tests', UNIT_TEST_MODULES) if FUNCTIONAL_TESTS: runner.run('Functional tests', FUNC_TEST_MODULES) runner.build() print runner.errors print runner.warnings print runner.summary # send a mail if SEND_REPORT: runner.sendReport()