reporter.py revision 440416c42460ff291475a273205375c8ddb30333
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich# -*- coding: utf-8 -*-
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich# $Id$
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich# pylint: disable=C0302
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich"""
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen RabinovichTestdriver reporter module.
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich"""
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich__copyright__ = \
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich"""
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen RabinovichCopyright (C) 2010-2014 Oracle Corporation
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen RabinovichThis file is part of VirtualBox Open Source Edition (OSE), as
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovichavailable from http://www.virtualbox.org. This file is free software;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichyou can redistribute it and/or modify it under the terms of the GNU
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen RabinovichGeneral Public License (GPL) as published by the Free Software
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen RabinovichFoundation, in version 2 as it comes in the "COPYING" file of the
76438e9b6959cc0bd7e35fb9cf5a11c87a37f744Allen RabinovichVirtualBox OSE distribution. VirtualBox OSE is distributed in the
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovichhope that it will be useful, but WITHOUT ANY WARRANTY of any kind.
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen RabinovichThe contents of this file may alternatively be used under the terms
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovichof the Common Development and Distribution License Version 1.0
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich(CDDL) only, as it comes in the "COPYING.CDDL" file of the
7211944935145abdba8a3ea3b17621ec264d8e21Allen RabinovichVirtualBox OSE distribution, in which case the provisions of the
7211944935145abdba8a3ea3b17621ec264d8e21Allen RabinovichCDDL are applicable instead of those of the GPL.
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen RabinovichYou may elect to license modified versions of this file under the
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichterms and conditions of either the GPL or the CDDL or both.
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich"""
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich__version__ = "$Revision$"
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich# Standard Python imports.
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovichimport array
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport datetime
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovichimport errno
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport os
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovichimport os.path
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport sys
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport time
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport threading
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichimport traceback
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich# Validation Kit imports.
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichfrom common import utils;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich## test reporter instance
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichg_oReporter = None;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichg_sReporterName = None;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichg_oLock = threading.Lock();
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichclass PythonLoggingStream(object):
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich Python logging => testdriver/reporter.py stream.
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich def write(self, sText):
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """Writes python log message to our stream."""
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich if g_oReporter != None:
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich sText = sText.rstrip("\r\n");
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich #g_oReporter.log(0, 'python: %s' % (sText), utils.getCallerName(), utils.getTimePrefix());
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich return True;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich def flush(self):
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """Flushes the stream."""
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich return True;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovichclass ReporterBase(object):
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich Base class for the reporters.
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich """
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich def __init__(self):
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich self.iVerbose = 1;
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich self.iDebug = 0;
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich self.cErrors = 0;
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich self.fTimedOut = False; # Once set, it trickles all the way up.
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich self.atTests = [];
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich self.sName = os.path.splitext(os.path.basename(sys.argv[0]))[0];
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich # Hook into the python logging.
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich import logging;
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich logging.basicConfig(stream = PythonLoggingStream(),
ec749fdeacb8e11cb7a9dd6dcebd1538b953059eAllen Rabinovich level = logging.DEBUG,
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich format = '%(name)-12s %(levelname)-8s %(message)s');
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich #
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich # Introspection and configuration.
70837a7c5871ea73a84ab67d8c34fcd9d2daa31cAllen Rabinovich #
66900b43227bf441b6a6b1084af2c96ce2c747c8Allen Rabinovich
7211944935145abdba8a3ea3b17621ec264d8e21Allen Rabinovich def isLocal(self):
"""Is this a local reporter?"""
return False;
def incVerbosity(self):
"""Increases the verbosity level."""
self.iVerbose += 1;
def incDebug(self):
"""Increases the debug level."""
self.iDebug += 1;
#
# Generic logging.
#
def log(self, iLevel, sText, sCaller, sTsPrf):
"""
Writes the specfied text to the log if iLevel is less or requal
to iVerbose.
"""
_ = iLevel; _ = sText; _ = sCaller; _ = sTsPrf;
return 0;
#
# XML output from the reporter.
#
def _xmlEscAttr(self, sValue):
"""Escapes an XML attribute value."""
sValue = sValue.replace('&', '&');
sValue = sValue.replace('<', '&lt;');
sValue = sValue.replace('>', '&gt;');
#sValue = sValue.replace('\'', '&apos;');
sValue = sValue.replace('"', '&quot;');
sValue = sValue.replace('\n', '&#xA');
sValue = sValue.replace('\r', '&#xD');
return sValue;
def _xmlWrite(self, asText, fIndent = True):
"""XML output function for the reporter."""
_ = asText; _ = fIndent;
return None;
def xmlFlush(self, fRetry = False, fForce = False):
"""Flushes XML output if buffered."""
_ = fRetry; _ = fForce;
return None;
#
# XML output from child.
#
def subXmlStart(self, oFileWrapper):
"""Called by the file wrapper when the first bytes are written to the test pipe."""
_ = oFileWrapper;
return None;
def subXmlWrite(self, oFileWrapper, sRawXml, sCaller):
"""Called by the file wrapper write method for test pipes."""
return self.log(0, 'raw xml%s: %s' % (oFileWrapper.sPrefix, sRawXml), sCaller, utils.getTimePrefix());
def subXmlEnd(self, oFileWrapper):
"""Called by the file wrapper __del__ method for test pipes."""
_ = oFileWrapper;
return None;
#
# File output.
#
def addLogFile(self, oSrcFile, sSrcFilename, sAltName, sDescription, sKind, sCaller, sTsPrf):
"""
Adds the file to the report.
Returns True on success, False on failure.
"""
_ = oSrcFile; _ = sSrcFilename; _ = sAltName; _ = sDescription; _ = sKind; _ = sCaller; _ = sTsPrf;
return True;
#
# Test reporting
#
def _testGetFullName(self):
"""
Mangles the test names in atTest into a single name to make it easier
to spot where we are.
"""
sName = '';
for t in self.atTests:
if sName != '':
sName += ', ';
sName += t[0];
return sName;
def testIncErrors(self):
"""Increates the error count."""
self.cErrors += 1;
return self.cErrors;
def testSetTimedOut(self):
"""Sets time out indicator for the current test and increases the error counter."""
self.fTimedOut = True;
self.cErrors += 1;
return None;
def testStart(self, sName, sCaller):
""" Starts a new test, may be nested. """
(sTsPrf, sTsIso) = utils.getTimePrefixAndIsoTimestamp();
self._xmlWrite([ '<Test timestamp="%s" name="%s">' % (sTsIso, self._xmlEscAttr(sName),), ]);
self.atTests.append((sName, self.cErrors, self.fTimedOut));
self.fTimedOut = False;
return self.log(1, '%-50s: TESTING' % (self._testGetFullName()), sCaller, sTsPrf);
def testValue(self, sName, sValue, sUnit, sCaller):
""" Reports a benchmark value or something simiarlly useful. """
(sTsPrf, sTsIso) = utils.getTimePrefixAndIsoTimestamp();
self._xmlWrite([ '<Value timestamp="%s" name="%s" unit="%s" value="%s"/>'
% (sTsIso, self._xmlEscAttr(sName), self._xmlEscAttr(sUnit), self._xmlEscAttr(sValue)), ]);
return self.log(0, ' %-48s: %12s %s' % (sName, sValue, sUnit), sCaller, sTsPrf);
def testFailure(self, sDetails, sCaller):
""" Reports a failure. """
(sTsPrf, sTsIso) = utils.getTimePrefixAndIsoTimestamp();
self.cErrors = self.cErrors + 1;
self._xmlWrite([ '<FailureDetails timestamp="%s" text="%s"/>' % (sTsIso, self._xmlEscAttr(sDetails),), ]);
return self.log(0, sDetails, sCaller, sTsPrf);
def testDone(self, fSkipped, sCaller):
"""
Marks the current test as DONE, pops it and maks the next test on the
stack current.
Returns (name, errors).
"""
(sTsPrf, sTsIso) = utils.getTimePrefixAndIsoTimestamp();
sFullName = self._testGetFullName();
# safe pop
if len(self.atTests) <= 0:
self.log(0, 'testDone on empty test stack!', sCaller, sTsPrf);
return ('internal error', 0);
fTimedOut = self.fTimedOut;
sName, cErrorsStart, self.fTimedOut = self.atTests.pop();
# log + xml.
cErrors = self.cErrors - cErrorsStart;
if cErrors == 0:
if fSkipped is not True:
self._xmlWrite([ ' <Passed timestamp="%s"/>' % (sTsIso,), '</Test>' ],);
self.log(1, '%-50s: PASSED' % (sFullName,), sCaller, sTsPrf);
else:
self._xmlWrite([ ' <Skipped timestamp="%s"/>' % (sTsIso,), '</Test>' ]);
self.log(1, '%-50s: SKIPPED' % (sFullName,), sCaller, sTsPrf);
elif fTimedOut:
self._xmlWrite([ ' <TimedOut timestamp="%s" errors="%d"/>' % (sTsIso, cErrors), '</Test>' ]);
self.log(0, '%-50s: TIMED-OUT - %d errors' % (sFullName, cErrors), sCaller, sTsPrf);
else:
self._xmlWrite([ ' <Failed timestamp="%s" errors="%d"/>' % (sTsIso, cErrors), '</Test>' ]);
self.log(0, '%-50s: FAILED - %d errors' % (sFullName, cErrors), sCaller, sTsPrf);
# Flush buffers when reaching the last test.
if len(self.atTests) == 0:
self.xmlFlush(fRetry = True);
return (sName, cErrors);
def testErrorCount(self):
"""
Returns the number of errors accumulated by the current test.
"""
cTests = len(self.atTests);
if cTests <= 0:
return self.cErrors;
return self.cErrors - self.atTests[cTests - 1][1];
def testCleanup(self, sCaller):
"""
Closes all open test as failed.
Returns True if no open tests, False if there were open tests.
"""
if len(self.atTests) == 0:
return True;
for _ in range(len(self.atTests)):
self.testFailure('Test not closed by test drver', sCaller)
self.testDone(False, sCaller);
return False;
class LocalReporter(ReporterBase):
"""
Local reporter instance.
"""
def __init__(self):
ReporterBase.__init__(self);
self.oLogFile = None;
self.oXmlFile = None;
self.fXmlOk = True;
self.iSubXml = 0;
self.iOtherFile = 0;
self.fnGetIsoTimestamp = utils.getIsoTimestamp; # Hack to get a timestamp in __del__.
self.oStdErr = sys.stderr; # Hack for __del__ output.
#
# Figure the main log directory.
#
try:
import user;
self.sDefLogDir = os.path.abspath(os.path.join(user.home, "VBoxTestLogs"));
except:
self.sDefLogDir = os.path.abspath("VBoxTestLogs");
try:
sLogDir = os.path.abspath(os.environ.get('TESTBOX_REPORTER_LOG_DIR', self.sDefLogDir));
if not os.path.isdir(sLogDir):
os.makedirs(sLogDir, 0750);
except:
sLogDir = self.sDefLogDir;
if not os.path.isdir(sLogDir):
os.makedirs(sLogDir, 0750);
#
# Make a subdirectory for this test run.
#
sTs = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H-%M-%S.log');
self.sLogDir = sLogDir = os.path.join(sLogDir, '%s-%s' % (sTs, self.sName));
try:
os.makedirs(self.sLogDir, 0750);
except:
self.sLogDir = '%s-%s' % (self.sLogDir, os.getpid());
os.makedirs(self.sLogDir, 0750);
#
# Open the log file and write a header.
#
sLogName = os.path.join(self.sLogDir, 'testsuite.log');
sTsIso = utils.getIsoTimestamp();
self.oLogFile = utils.openNoInherit(sLogName, "w");
self.oLogFile.write(('Created log file at %s.\nRunning: %s' % (sTsIso, sys.argv)).encode('utf-8'));
#
# Open the xml log file and write the mandatory introduction.
#
# Note! This is done here and not in the base class because the remote
# logger doesn't really need this. It doesn't need the outer
# test wrapper either.
#
sXmlName = os.path.join(self.sLogDir, 'testsuite.xml');
self.oXmlFile = utils.openNoInherit(sXmlName, "w");
self._xmlWrite([ '<?xml version="1.0" encoding="UTF-8" ?>',
'<Test timestamp="%s" name="%s">' % (sTsIso, self._xmlEscAttr(self.sName),), ],
fIndent = False);
def __del__(self):
"""Ends and completes the log files."""
try: sTsIso = self.fnGetIsoTimestamp();
except Exception, oXcpt:
sTsIso = str(oXcpt);
if self.oLogFile is not None:
try:
self.oLogFile.write(('\nThe End %s\n' % (sTsIso,)).encode('utf-8'));
self.oLogFile.close();
except: pass;
self.oLogFile = None;
if self.oXmlFile is not None:
self._closeXml(sTsIso);
self.oXmlFile = None;
def _closeXml(self, sTsIso):
"""Closes the XML file."""
if self.oXmlFile is not None:
# pop the test stack
while len(self.atTests) > 0:
sName, cErrorsStart, self.fTimedOut = self.atTests.pop();
self._xmlWrite([ '<End timestamp="%s" errors="%d"/>' % (sTsIso, self.cErrors - cErrorsStart,),
'</%s>' % (sName,), ]);
# The outer one is not on the stack.
self._xmlWrite([ ' <End timestamp="%s"/>' % (sTsIso,),
'</Test>', ], fIndent = False);
try:
self.oXmlFile.close();
self.oXmlFile = None;
except:
pass;
def _xmlWrite(self, asText, fIndent = True):
"""Writes to the XML file."""
for sText in asText:
if fIndent:
sIndent = ''.ljust((len(self.atTests) + 1) * 2);
sText = sIndent + sText;
sText += '\n';
try:
self.oXmlFile.write(sText.encode('utf-8'));
except:
if self.fXmlOk:
traceback.print_exc();
self.fXmlOk = False;
return False;
return True;
#
# Overridden methods.
#
def isLocal(self):
"""Is this a local reporter?"""
return True;
def log(self, iLevel, sText, sCaller, sTsPrf):
if iLevel <= self.iVerbose:
# format it.
if self.iDebug > 0:
sLogText = '%s %30s: %s' % (sTsPrf, sCaller, sText);
else:
sLogText = '%s %s' % (sTsPrf, sText);
# output it.
sAscii = sLogText.encode('ascii', 'replace');
if self.iDebug == 0:
print >> self.oStdErr, '%s: %s' % (self.sName, sAscii)
else:
print >> self.oStdErr, '%s' % (sAscii)
sLogText += '\n';
try:
self.oLogFile.write(sLogText.encode('utf-8'));
except:
pass;
return 0;
def addLogFile(self, oSrcFile, sSrcFilename, sAltName, sDescription, sKind, sCaller, sTsPrf):
# Figure the destination filename.
iOtherFile = self.iOtherFile;
self.iOtherFile += 1;
sDstFilename = os.path.join(self.sLogDir, 'other-%d-%s.log' \
% (iOtherFile, os.path.splitext(os.path.basename(sSrcFilename))[0]));
self.log(0, 'Other log file: %s - %s (%s)' % (sDstFilename, sDescription, sSrcFilename), sCaller, sTsPrf);
# Open the destination file and copy over the data.
fRc = True;
try:
oDstFile = utils.openNoInherit(sDstFilename, 'w');
except Exception, oXcpt:
self.log(0, 'error opening %s: %s' % (sDstFilename, oXcpt), sCaller, sTsPrf);
else:
while True:
try:
abBuf = oSrcFile.read(65536);
except Exception, oXcpt:
fRc = False;
self.log(0, 'error reading %s: %s' % (sSrcFilename, oXcpt), sCaller, sTsPrf);
else:
try:
oDstFile.write(abBuf);
except Exception, oXcpt:
fRc = False;
self.log(0, 'error writing %s: %s' % (sDstFilename, oXcpt), sCaller, sTsPrf);
else:
if len(abBuf) > 0:
continue;
break;
oDstFile.close();
# Leave a mark in the XML log.
self._xmlWrite('<LogFile timestamp="%s" filename="%s" source="%s" kind="%s" ok="%s">%s</LogFile>\n'
% (utils.getIsoTimestamp(), self._xmlEscAttr(os.path.basename(sDstFilename)), self._xmlEscAttr(sSrcFilename), \
self._xmlEscAttr(sKind), fRc, self._xmlEscAttr(sDescription)) );
_ = sAltName;
return fRc;
def subXmlStart(self, oFileWrapper):
# Open a new file and just include it from the main XML.
iSubXml = self.iSubXml;
self.iSubXml += 1;
sSubXmlName = os.path.join(self.sLogDir, 'sub-%d.xml' % (iSubXml,));
try:
oFileWrapper.oSubXmlFile = utils.openNoInherit(sSubXmlName, "w");
except:
errorXcpt('open(%s)' % oFileWrapper.oSubXmlName);
oFileWrapper.oSubXmlFile = None;
else:
self._xmlWrite('<Include timestamp="%s" filename="%s"/>\n'
% (utils.getIsoTimestamp(), self._xmlEscAttr(os.path.basename(sSubXmlName))));
return None;
def subXmlWrite(self, oFileWrapper, sRawXml, sCaller):
if oFileWrapper.oSubXmlFile is not None:
try:
oFileWrapper.oSubXmlFile.write(sRawXml);
except:
pass;
if sCaller is None: pass; # pychecker - NOREF
return None;
def subXmlEnd(self, oFileWrapper):
if oFileWrapper.oSubXmlFile is not None:
try:
oFileWrapper.oSubXmlFile.close();
oFileWrapper.oSubXmlFile = None;
except:
pass;
return None;
class RemoteReporter(ReporterBase):
"""
Reporter that talks to the test manager server.
"""
## The XML sync min time (seconds).
kcSecXmlFlushMin = 30;
## The XML sync max time (seconds).
kcSecXmlFlushMax = 120;
## The XML sync idle time before flushing (seconds).
kcSecXmlFlushIdle = 5;
## The XML sync line count threshold.
kcLinesXmlFlush = 512;
## The retry timeout.
kcSecTestManagerRetryTimeout = 120;
## The request timeout.
kcSecTestManagerRequestTimeout = 30;
def __init__(self):
ReporterBase.__init__(self);
self.sTestManagerUrl = os.environ.get('TESTBOX_MANAGER_URL');
self.sTestBoxUuid = os.environ.get('TESTBOX_UUID');
self.idTestBox = int(os.environ.get('TESTBOX_ID'));
self.idTestSet = int(os.environ.get('TESTBOX_TEST_SET_ID'));
self._asXml = [];
self._secTsXmlFlush = utils.timestampSecond();
self._secTsXmlLast = self._secTsXmlFlush;
self._fXmlFlushing = False;
self.oOutput = sys.stdout; # Hack for __del__ output.
self.fFlushEachLine = True;
self.fDebugXml = 'TESTDRIVER_REPORTER_DEBUG_XML' in os.environ;
# Prepare the TM connecting.
import urlparse;
import httplib;
import urllib;
from common import constants;
self._fnUrlEncode = urllib.urlencode;
self._fnUrlParseQs = urlparse.parse_qs;
self._oParsedTmUrl = urlparse.urlparse(self.sTestManagerUrl);
if sys.version_info[0] >= 3 \
or (sys.version_info[0] == 2 and sys.version_info[1] >= 6):
if self._oParsedTmUrl.scheme == 'https': # pylint: disable=E1101
self._fnTmConnect = lambda: httplib.HTTPSConnection(self._oParsedTmUrl.hostname,
timeout = self.kcSecTestManagerRequestTimeout);
else:
self._fnTmConnect = lambda: httplib.HTTPConnection( self._oParsedTmUrl.hostname,
timeout = self.kcSecTestManagerRequestTimeout);
else:
if self._oParsedTmUrl.scheme == 'https': # pylint: disable=E1101
self._fnTmConnect = lambda: httplib.HTTPSConnection(self._oParsedTmUrl.hostname);
else:
self._fnTmConnect = lambda: httplib.HTTPConnection( self._oParsedTmUrl.hostname);
self._dHttpHeader = \
{
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
'User-Agent': 'TestDriverReporter/%s.0 (%s, %s)' % (__version__, utils.getHostOs(), utils.getHostArch(),),
'Accept': 'text/plain,application/x-www-form-urlencoded',
'Accept-Encoding': 'identity',
'Cache-Control': 'max-age=0',
#'Connection': 'keep-alive',
};
dParams = {
constants.tbreq.ALL_PARAM_TESTBOX_UUID: self.sTestBoxUuid,
constants.tbreq.ALL_PARAM_TESTBOX_ID: self.idTestBox,
constants.tbreq.RESULT_PARAM_TEST_SET_ID: self.idTestSet,
};
self._sTmServerPath = '/%s/testboxdisp.py?%s' \
% ( self._oParsedTmUrl.path.strip('/'), # pylint: disable=E1101
urllib.urlencode(dParams), );
def __del__(self):
"""Flush pending log messages?"""
if len(self._asXml) > 0:
self._xmlDoFlush(self._asXml, fRetry = True, fDtor = True);
def _writeOutput(self, sText):
""" Does the actual writing and flushing. """
print >> self.oOutput, sText.encode('ascii', 'replace');
if self.fFlushEachLine: self.oOutput.flush();
return None;
#
# Talking to TM.
#
def _processTmStatusResponse(self, oConn, sOperation, fClose = True):
"""
Processes HTTP reponse from the test manager.
Returns True, False or None. None should be retried, the others not.
May raise exception on HTTP issue (retry ok).
"""
import httplib;
from common import constants;
# Read the response and (optionally) close the connection.
oResponse = oConn.getresponse();
try:
sRspBody = oResponse.read();
except httplib.IncompleteRead, oXcpt:
self._writeOutput('%s: %s: Warning: httplib.IncompleteRead: %s [expected %s, got %s]'
% (utils.getTimePrefix(), sOperation, oXcpt, oXcpt.expected, len(oXcpt.partial),));
sRspBody = oXcpt.partial;
if fClose is True:
try: oConn.close();
except: pass;
# Check the content type.
sContentType = oResponse.getheader('Content-Type');
if sContentType is not None and sContentType == 'application/x-www-form-urlencoded; charset=utf-8':
# Parse the body and check the RESULT parameter.
dResponse = self._fnUrlParseQs(sRspBody, strict_parsing = True);
sResult = dResponse.get(constants.tbresp.ALL_PARAM_RESULT, None);
if isinstance(sResult, list):
sResult = sResult[0] if len(sResult) == 1 else '%d results' % (len(sResult),);
if sResult is not None:
if sResult == constants.tbresp.STATUS_ACK:
return True;
if sResult == constants.tbresp.STATUS_NACK:
self._writeOutput('%s: %s: Failed (%s). (dResponse=%s)'
% (utils.getTimePrefix(), sOperation, sResult, dResponse,));
return False;
self._writeOutput('%s: %s: Failed - dResponse=%s' % (utils.getTimePrefix(), sOperation, dResponse,));
else:
self._writeOutput('%s: %s: Unexpected Content-Type: %s' % (utils.getTimePrefix(), sOperation, sContentType,));
self._writeOutput('%s: %s: Body: %s' % (utils.getTimePrefix(), sOperation, sRspBody,));
return None;
def _doUploadFile(self, oSrcFile, sSrcFilename, sDescription, sKind, sMime):
""" Uploads the given file to the test manager. """
# Prepare header and url.
dHeader = dict(self._dHttpHeader);
dHeader['Content-Type'] = 'application/octet-stream';
self._writeOutput('%s: _doUploadFile: sHeader=%s' % (utils.getTimePrefix(), dHeader,));
oSrcFile.seek(0, 2);
self._writeOutput('%s: _doUploadFile: size=%d' % (utils.getTimePrefix(), oSrcFile.tell(),));
oSrcFile.seek(0);
from common import constants;
sUrl = self._sTmServerPath + '&' \
+ self._fnUrlEncode({ constants.tbreq.UPLOAD_PARAM_NAME: os.path.basename(sSrcFilename),
constants.tbreq.UPLOAD_PARAM_DESC: sDescription,
constants.tbreq.UPLOAD_PARAM_KIND: sKind,
constants.tbreq.UPLOAD_PARAM_MIME: sMime,
constants.tbreq.ALL_PARAM_ACTION: constants.tbreq.UPLOAD,
});
self._writeOutput('%s: _doUploadFile: url="%s"' % (utils.getTimePrefix(), sUrl));
# Retry loop.
secStart = utils.timestampSecond();
while True:
try:
oConn = self._fnTmConnect();
oConn.request('POST', sUrl, oSrcFile.read(), dHeader);
fRc = self._processTmStatusResponse(oConn, '_doUploadFile', fClose = True);
oConn.close();
if fRc is not None:
return fRc;
except:
logXcpt('warning: exception during UPLOAD request');
if utils.timestampSecond() - secStart >= self.kcSecTestManagerRetryTimeout:
self._writeOutput('%s: _doUploadFile: Timed out.' % (utils.getTimePrefix(),));
break;
try: oSrcFile.seek(0);
except:
logXcpt();
break;
self._writeOutput('%s: _doUploadFile: Retrying...' % (utils.getTimePrefix(), ));
time.sleep(2);
return False;
def _xmlDoFlush(self, asXml, fRetry = False, fDtor = False):
"""
The code that does the actual talking to the server.
Used by both xmlFlush and __del__.
"""
secStart = utils.timestampSecond();
while True:
fRc = None;
try:
# Post.
from common import constants;
sPostBody = self._fnUrlEncode({constants.tbreq.XML_RESULT_PARAM_BODY: '\n'.join(asXml),});
oConn = self._fnTmConnect();
oConn.request('POST',
self._sTmServerPath + ('&%s=%s' % (constants.tbreq.ALL_PARAM_ACTION, constants.tbreq.XML_RESULTS)),
sPostBody,
self._dHttpHeader);
fRc = self._processTmStatusResponse(oConn, '_xmlDoFlush', fClose = True);
if fRc is True:
if self.fDebugXml:
self._writeOutput('_xmlDoFlush:\n%s' % ('\n'.join(asXml),));
return (None, False);
if fRc is False:
self._writeOutput('_xmlDoFlush: Failed - we should abort the test, really.');
return (None, True);
except Exception, oXcpt:
if not fDtor:
logXcpt('warning: exception during XML_RESULTS request');
else:
self._writeOutput('warning: exception during XML_RESULTS request: %s' % (oXcpt,));
if fRetry is not True \
or utils.timestampSecond() - secStart >= self.kcSecTestManagerRetryTimeout:
break;
time.sleep(2);
return (asXml, False);
#
# Overridden methods.
#
def isLocal(self):
return False;
def log(self, iLevel, sText, sCaller, sTsPrf):
if iLevel <= self.iVerbose:
if self.iDebug > 0:
sLogText = '%s %30s: %s' % (sTsPrf, sCaller, sText);
else:
sLogText = '%s %s: %s' % (sTsPrf, self.sName, sText);
self._writeOutput(sLogText);
return 0;
def addLogFile(self, oSrcFile, sSrcFilename, sAltName, sDescription, sKind, sCaller, sTsPrf):
fRc = True;
if sKind in [ 'text', 'log', ] or sKind.startswith('log/'):
self.log(0, '*** Uploading "%s" - KIND: "%s" - DESC: "%s" ***'
% (sSrcFilename, sKind, sDescription), sCaller, sTsPrf);
g_oLock.release();
self._doUploadFile(oSrcFile, sAltName, sDescription, sKind, 'text/plain');
g_oLock.acquire();
elif sKind.startswith('screenshot/'):
self.log(0, '*** Uploading "%s" - KIND: "%s" - DESC: "%s" ***'
% (sSrcFilename, sKind, sDescription), sCaller, sTsPrf);
g_oLock.release();
self._doUploadFile(oSrcFile, sAltName, sDescription, sKind, 'image/png');
g_oLock.acquire();
else:
self.log(0, '*** UNKNOWN FILE "%s" - KIND "%s" - DESC "%s" ***'
% (sSrcFilename, sKind, sDescription), sCaller, sTsPrf);
return fRc;
def xmlFlush(self, fRetry = False, fForce = False):
"""
Flushes the XML back log. Called with the lock held, may leave it
while communicating with the server.
"""
if not self._fXmlFlushing:
asXml = self._asXml;
self._asXml = [];
if len(asXml) > 0 or fForce is True:
self._fXmlFlushing = True;
g_oLock.release();
(asXml, fIncErrors) = self._xmlDoFlush(asXml, fRetry = fRetry);
g_oLock.acquire();
if fIncErrors:
self.testIncErrors();
self._fXmlFlushing = False;
if asXml is None:
self._secTsXmlFlush = utils.timestampSecond();
else:
self._asXml = asXml + self._asXml;
return True;
self._secTsXmlFlush = utils.timestampSecond();
return False;
def _xmlFlushIfNecessary(self):
"""Flushes the XML back log if necessary."""
tsNow = utils.timestampSecond();
cSecs = tsNow - self._secTsXmlFlush;
cSecsLast = tsNow - self._secTsXmlLast;
self._secTsXmlLast = tsNow;
# Absolute flush thresholds.
if cSecs >= self.kcSecXmlFlushMax:
return self.xmlFlush();
if len(self._asXml) >= self.kcLinesXmlFlush:
return self.xmlFlush();
# Flush if idle long enough.
if cSecs >= self.kcSecXmlFlushMin \
and cSecsLast >= self.kcSecXmlFlushIdle:
return self.xmlFlush();
return False;
def _xmlWrite(self, asText, fIndent = True):
"""XML output function for the reporter."""
self._asXml += asText;
self._xmlFlushIfNecessary();
_ = fIndent; # No pretty printing, thank you.
return None;
def subXmlStart(self, oFileWrapper):
oFileWrapper.sXmlBuffer = '';
return None;
def subXmlWrite(self, oFileWrapper, sRawXml, sCaller):
oFileWrapper.sXmlBuffer += sRawXml;
_ = sCaller;
return None;
def subXmlEnd(self, oFileWrapper):
sRawXml = oFileWrapper.sXmlBuffer;
## @todo should validate the document here and maybe auto terminate things. Adding some hints to have the server do
# this instead.
g_oLock.acquire();
self._asXml += [ '<PushHint testdepth="%d"/>' % (len(self.atTests),),
sRawXml,
'<PopHint testdepth="%d"/>' % (len(self.atTests),),];
self._xmlFlushIfNecessary();
g_oLock.release();
return None;
#
# Helpers
#
def logXcptWorker(iLevel, fIncErrors, sPrefix="", sText=None, cFrames=1):
"""
Log an exception, optionally with a preceeding message and more than one
call frame.
"""
g_oLock.acquire();
if fIncErrors:
g_oReporter.testIncErrors();
## @todo skip all this if iLevel is too high!
# Try get exception info.
sTsPrf = utils.getTimePrefix();
try:
oType, oValue, oTraceback = sys.exc_info();
except:
oType = oValue = oTraceback = None;
if oType is not None:
# Try format the info
try:
rc = 0;
sCaller = utils.getCallerName(oTraceback.tb_frame);
if sText is not None:
rc = g_oReporter.log(iLevel, "%s%s" % (sPrefix, sText), sCaller, sTsPrf);
asInfo = [];
try:
asInfo = asInfo + traceback.format_exception_only(oType, oValue);
if cFrames is not None and cFrames <= 1:
asInfo = asInfo + traceback.format_tb(oTraceback, 1);
else:
asInfo.append('Traceback:')
asInfo = asInfo + traceback.format_tb(oTraceback, cFrames);
asInfo.append('Stack:')
asInfo = asInfo + traceback.format_stack(oTraceback.tb_frame.f_back, cFrames);
except:
g_oReporter.log(0, 'internal-error: Hit exception #2! %s' % (traceback.format_exc()), sCaller, sTsPrf);
if len(asInfo) > 0:
# Do the logging.
for sItem in asInfo:
asLines = sItem.splitlines();
for sLine in asLines:
rc = g_oReporter.log(iLevel, '%s%s' % (sPrefix, sLine), sCaller, sTsPrf);
else:
g_oReporter.log(iLevel, 'No exception info...', sCaller, sTsPrf);
rc = -3;
except:
g_oReporter.log(0, 'internal-error: Hit exception! %s' % (traceback.format_exc()), None, sTsPrf);
rc = -2;
else:
g_oReporter.log(0, 'internal-error: No exception! %s'
% (utils.getCallerName(iFrame=3)), utils.getCallerName(iFrame=3), sTsPrf);
rc = -1;
g_oLock.release();
return rc;
#
# The public Classes
#
class FileWrapper(object):
""" File like class for TXS EXEC and similar. """
def __init__(self, sPrefix):
self.sPrefix = sPrefix;
def read(self, cb):
"""file.read"""
_ = cb;
return "";
def write(self, sText):
"""file.write"""
if isinstance(sText, array.array):
try:
sText = sText.tostring();
except:
pass;
g_oLock.acquire();
try:
sTsPrf = utils.getTimePrefix();
sCaller = utils.getCallerName();
asLines = sText.splitlines();
for sLine in asLines:
g_oReporter.log(0, '%s: %s' % (self.sPrefix, sLine), sCaller, sTsPrf);
except:
traceback.print_exc();
g_oLock.release();
return None;
class FileWrapperTestPipe(object):
""" File like class for the test pipe (TXS EXEC and similar). """
def __init__(self):
self.sPrefix = '';
self.fStarted = False;
self.fClosed = False;
def __del__(self):
self.close();
def close(self):
""" file.close """
if self.fStarted is True and self.fClosed is False:
self.fClosed = True;
try: g_oReporter.subXmlEnd(self);
except:
try: traceback.print_exc();
except: pass;
return True;
def read(self, cb = None):
"""file.read"""
_ = cb;
return "";
def write(self, sText):
"""file.write"""
# lazy start.
if self.fStarted is not True:
try:
g_oReporter.subXmlStart(self);
except:
traceback.print_exc();
self.fStarted = True;
if isinstance(sText, array.array):
try:
sText = sText.tostring();
except:
pass;
try:
g_oReporter.subXmlWrite(self, sText, utils.getCallerName());
except:
traceback.print_exc();
return None;
#
# The public APIs.
#
def log(sText):
"""Writes the specfied text to the log."""
g_oLock.acquire();
try:
rc = g_oReporter.log(1, sText, utils.getCallerName(), utils.getTimePrefix());
except:
rc = -1;
g_oLock.release();
return rc;
def logXcpt(sText=None, cFrames=1):
"""
Log an exception, optionally with a preceeding message and more than one
call frame.
"""
return logXcptWorker(1, False, "", sText, cFrames);
def log2(sText):
"""Log level 2: Writes the specfied text to the log."""
g_oLock.acquire();
try:
rc = g_oReporter.log(2, sText, utils.getCallerName(), utils.getTimePrefix());
except:
rc = -1;
g_oLock.release();
return rc;
def log2Xcpt(sText=None, cFrames=1):
"""
Log level 2: Log an exception, optionally with a preceeding message and
more than one call frame.
"""
return logXcptWorker(2, False, "", sText, cFrames);
def maybeErr(fIsError, sText):
""" Maybe error or maybe normal log entry. """
if fIsError is True:
return error(sText);
return log(sText);
def maybeErrXcpt(fIsError, sText=None, cFrames=1):
""" Maybe error or maybe normal log exception entry. """
if fIsError is True:
return errorXcpt(sText, cFrames);
return logXcpt(sText, cFrames);
def maybeLog(fIsNotError, sText):
""" Maybe error or maybe normal log entry. """
if fIsNotError is not True:
return error(sText);
return log(sText);
def maybeLogXcpt(fIsNotError, sText=None, cFrames=1):
""" Maybe error or maybe normal log exception entry. """
if fIsNotError is not True:
return errorXcpt(sText, cFrames);
return logXcpt(sText, cFrames);
def error(sText):
"""
Writes the specfied error message to the log.
This will add an error to the current test.
Always returns False for the convenience of methods returning boolean
success indicators.
"""
g_oLock.acquire();
g_oReporter.testIncErrors();
try:
g_oReporter.log(0, 'error: %s' % (sText), utils.getCallerName(), utils.getTimePrefix());
except:
pass;
g_oLock.release();
return False;
def errorXcpt(sText=None, cFrames=1):
"""
Log an error caused by an exception. If sText is given, it will preceed
the exception information. cFrames can be used to display more stack.
This will add an error to the current test.
Always returns False for the convenience of methods returning boolean
success indicators.
"""
logXcptWorker(0, True, "error: ", sText, cFrames);
return False;
def errorTimeout(sText):
"""
Flags the current test as having timed out and writes the specified message to the log.
This will add an error to the current test.
Always returns False for the convenience of methods returning boolean
success indicators.
"""
g_oLock.acquire();
g_oReporter.testSetTimedOut();
try:
g_oReporter.log(0, 'timeout-error: %s' % (sText), utils.getCallerName(), utils.getTimePrefix());
except:
pass;
g_oLock.release();
return False;
def fatal(sText):
"""
Writes a fatal error to the log.
This will add an error to the current test.
Always returns False for the convenience of methods returning boolean
success indicators.
"""
g_oLock.acquire();
g_oReporter.testIncErrors();
try:
g_oReporter.log(0, 'fatal error: %s' % (sText), utils.getCallerName(), utils.getTimePrefix());
except:
pass
g_oLock.release();
return False;
def fatalXcpt(sText=None, cFrames=1):
"""
Log a fatal error caused by an exception. If sText is given, it will
preceed the exception information. cFrames can be used to display more
stack.
This will add an error to the current test.
Always returns False for the convenience of methods returning boolean
success indicators.
"""
logXcptWorker(1, True, "fatal error: ", sText, cFrames);
return False;
def addLogFile(sFilename, sKind, sDescription = '', sAltName = None):
"""
Adds the specified log file to the report if the file exists.
The sDescription is a free form description of the log file.
The sKind parameter is for adding some machine parsable hint what kind of
log file this really is.
Returns True on success, False on failure (no ENOENT errors are logged).
"""
sTsPrf = utils.getTimePrefix();
sCaller = utils.getCallerName();
fRc = False;
if sAltName is None:
sAltName = sFilename;
try:
oSrcFile = utils.openNoInherit(sFilename, 'rb');
except IOError, oXcpt:
if oXcpt.errno != errno.ENOENT:
logXcpt('addLogFile(%s,%s,%s)' % (sFilename, sDescription, sKind));
else:
logXcpt('addLogFile(%s,%s,%s) IOError' % (sFilename, sDescription, sKind));
except:
logXcpt('addLogFile(%s,%s,%s)' % (sFilename, sDescription, sKind));
else:
g_oLock.acquire();
fRc = g_oReporter.addLogFile(oSrcFile, sFilename, sAltName, sDescription, sKind, sCaller, sTsPrf);
g_oLock.release();
oSrcFile.close();
return fRc;
def isLocal():
"""Is this a local reporter?"""
return g_oReporter.isLocal()
def incVerbosity():
"""Increases the verbosity level."""
return g_oReporter.incVerbosity()
def incDebug():
"""Increases the debug level."""
return g_oReporter.incDebug()
def getErrorCount():
"""
Get the current error count for the entire test run.
"""
g_oLock.acquire();
cErrors = g_oReporter.cErrors;
g_oLock.release();
return cErrors;
#
# Test reporting, a bit similar to RTTestI*.
#
def testStart(sName):
"""
Starts a new test (pushes it).
"""
g_oLock.acquire();
rc = g_oReporter.testStart(sName, utils.getCallerName());
g_oLock.release();
return rc;
def testValue(sName, sValue, sUnit):
"""
Reports a benchmark value or something simiarlly useful.
"""
g_oLock.acquire();
rc = g_oReporter.testValue(sName, str(sValue), sUnit, utils.getCallerName());
g_oLock.release();
return rc;
def testFailure(sDetails):
"""
Reports a failure.
We count these calls and testDone will use them to report PASSED or FAILED.
Returns False so that a return False line can be saved.
"""
g_oLock.acquire();
g_oReporter.testFailure(sDetails, utils.getCallerName());
g_oLock.release();
return False;
def testFailureXcpt(sDetails = ''):
"""
Reports a failure with exception.
We count these calls and testDone will use them to report PASSED or FAILED.
Returns False so that a return False line can be saved.
"""
# Extract exception info.
try:
oType, oValue, oTraceback = sys.exc_info();
except:
oType = oValue, oTraceback = None;
if oType is not None:
sCaller = utils.getCallerName(oTraceback.tb_frame);
sXcpt = ' '.join(traceback.format_exception_only(oType, oValue));
else:
sCaller = utils.getCallerName();
sXcpt = 'No exception at %s' % (sCaller,);
# Use testFailure to do the work.
g_oLock.acquire();
if sDetails == '':
g_oReporter.testFailure('Exception: %s' % (sXcpt,), sCaller);
else:
g_oReporter.testFailure('%s: %s' % (sDetails, sXcpt), sCaller);
g_oLock.release();
return False;
def testDone(fSkipped = False):
"""
Completes the current test (pops it), logging PASSED / FAILURE.
Returns a tuple with the name of the test and its error count.
"""
g_oLock.acquire();
rc = g_oReporter.testDone(fSkipped, utils.getCallerName());
g_oLock.release();
return rc;
def testErrorCount():
"""
Gets the error count of the current test.
Returns the number of errors.
"""
g_oLock.acquire();
cErrors = g_oReporter.testErrorCount();
g_oLock.release();
return cErrors;
def testCleanup():
"""
Closes all open tests with a generic error condition.
Returns True if no open tests, False if something had to be closed with failure.
"""
g_oLock.acquire();
fRc = g_oReporter.testCleanup(utils.getCallerName());
g_oReporter.xmlFlush(fRetry = False, fForce = True);
g_oLock.release();
return fRc;
#
# Sub XML stuff.
#
def addSubXmlFile(sFilename):
"""
Adds a sub-xml result file to the party.
"""
fRc = False;
try:
oSrcFile = utils.openNoInherit(sFilename, 'r');
except IOError, oXcpt:
if oXcpt.errno != errno.ENOENT:
logXcpt('addSubXmlFile(%s)' % (sFilename,));
except:
logXcpt('addSubXmlFile(%s)' % (sFilename,));
else:
try:
oWrapper = FileWrapperTestPipe()
oWrapper.write(oSrcFile.read());
oWrapper.close();
except:
logXcpt('addSubXmlFile(%s)' % (sFilename,));
oSrcFile.close();
return fRc;
#
# Other useful debugging tools.
#
def logAllStacks(cFrames = None):
"""
Logs the stacks of all python threads.
"""
sTsPrf = utils.getTimePrefix();
sCaller = utils.getCallerName();
g_oLock.acquire();
cThread = 0;
for idThread, oStack in sys._current_frames().items(): # >=2.5, a bit ugly - pylint: disable=W0212
try:
if cThread > 0:
g_oReporter.log(1, '', sCaller, sTsPrf);
g_oReporter.log(1, 'Thread %s (%#x)' % (idThread, idThread), sCaller, sTsPrf);
try:
asInfo = traceback.format_stack(oStack, cFrames);
except:
g_oReporter.log(1, ' Stack formatting failed w/ exception', sCaller, sTsPrf);
else:
for sInfo in asInfo:
asLines = sInfo.splitlines();
for sLine in asLines:
g_oReporter.log(1, sLine, sCaller, sTsPrf);
except:
pass;
cThread += 1;
g_oLock.release();
return None;
def checkTestManagerConnection():
"""
Checks the connection to the test manager.
Returns True if the connection is fine, False if not, None if not remote
reporter.
Note! This as the sideeffect of flushing XML.
"""
g_oLock.acquire();
fRc = g_oReporter.xmlFlush(fRetry = False, fForce = True);
g_oLock.release();
return fRc;
def flushall():
"""
Flushes all output streams, both standard and logger related.
"""
try: sys.stdout.flush();
except: pass;
try: sys.stderr.flush();
except: pass;
# Note! Current no logger specific streams to flush.
return True;
#
# Module initialization.
#
def _InitReporterModule():
"""
Instantiate the test reporter.
"""
global g_oReporter, g_sReporterName
g_sReporterName = os.getenv("TESTBOX_REPORTER", "local");
if g_sReporterName == "local":
g_oReporter = LocalReporter();
elif g_sReporterName == "remote":
g_oReporter = RemoteReporter();
else:
print >> sys.stderr, os.path.basename(__file__) + ": Unknown TESTBOX_REPORTER value: '" + g_sReporterName + "'";
raise Exception("Unknown TESTBOX_REPORTER value '" + g_sReporterName + "'");
if __name__ != "checker": # pychecker avoidance.
_InitReporterModule();