From d1ff1cbb917613fe886bbe998b9df17d92e13823 Mon Sep 17 00:00:00 2001 From: Ramon van Alteren Date: Thu, 5 Jan 2012 13:43:27 +0100 Subject: [PATCH] Cleanup of unused code We don't use the test stuff, deleting I might replace this later on with a more generic testing framework --- pyjenkinsci/command_line/meta_test.py | 50 ------- pyjenkinsci/utils/bufwrapper.py | 39 ------ pyjenkinsci/utils/dates.py | 14 -- pyjenkinsci/utils/id.py | 16 --- pyjenkinsci/utils/junitxml.py | 189 -------------------------- pyjenkinsci/utils/xmlrunner.py | 244 ---------------------------------- pyjenkinsci/utils/xmlrunnertest.py | 144 -------------------- 7 files changed, 696 deletions(-) delete mode 100644 pyjenkinsci/command_line/meta_test.py delete mode 100644 pyjenkinsci/utils/bufwrapper.py delete mode 100644 pyjenkinsci/utils/dates.py delete mode 100644 pyjenkinsci/utils/id.py delete mode 100644 pyjenkinsci/utils/junitxml.py delete mode 100644 pyjenkinsci/utils/xmlrunner.py delete mode 100644 pyjenkinsci/utils/xmlrunnertest.py diff --git a/pyjenkinsci/command_line/meta_test.py b/pyjenkinsci/command_line/meta_test.py deleted file mode 100644 index c918339..0000000 --- a/pyjenkinsci/command_line/meta_test.py +++ /dev/null @@ -1,50 +0,0 @@ -import optparse -import os -import random -import logging -from utils.id import mk_id -from utils import junitxml - -log = logging.getLogger(__name__) - -class meta_test(object): - ATTEMPTS=3 - - @classmethod - def mkParser(cls): - parser = optparse.OptionParser() - - def __init__(self, opts=None): - self.opts = opts - - def testFunction(self): - if random.random() < 0.1: - raise AssertionError("The value was too small") - return 0 - - def __call__(self): - temp_dir = os.environ.get("TEMP", r"c:\temp" ) - output_dir = os.environ.get( "WORKSPACE", temp_dir ) - result_filepath = os.path.join( output_dir, "results.xml" ) - stream = open( result_filepath, "wb" ) - testsuite_name = mk_id() - ju = junitxml.junitxml( stream, testsuite_name) - - - classname = mk_id() - for i in xrange(0, self.ATTEMPTS ): - tr = ju.startTest( classname, mk_id() ) - try: - tr.run( self.testFunction ) - except Exception, e: - log.exception(e) - continue - - ju.write() - -def main( ): - logging.basicConfig() - return meta_test()() - -if __name__ == "__main__": - main() diff --git a/pyjenkinsci/utils/bufwrapper.py b/pyjenkinsci/utils/bufwrapper.py deleted file mode 100644 index 8addaaa..0000000 --- a/pyjenkinsci/utils/bufwrapper.py +++ /dev/null @@ -1,39 +0,0 @@ -from cStringIO import StringIO - -class bufwrapper( object ): - """ - Basic buffer-wrapper - wraps up an output stream with a buffer. - """ - def __init__( self, stream, buffer=None ): - self.stream = stream - - assert hasattr( self.stream, "write" ), "%s does not support write" % repr(stream) - - if buffer is None: - self.buf = StringIO() - else: - self.buf = buffer - - def get_and_clear( self ): - """ - Get the contents of the buffer and clear it. - """ - old_buffer = self.buf - self.buf = StringIO() - return old_buffer.getvalue() - - def flush( self ): - for item in [ self.stream, self.buf ]: - if hasattr( item, "flush" ) and callable( item.flush ): - item.flush() - - - def close(self): - self.stream.close() - - def write(self, txt ): - self.stream.write(txt) - self.buf.write(txt) - - def getvalue(self): - return self.buf.getvalue() diff --git a/pyjenkinsci/utils/dates.py b/pyjenkinsci/utils/dates.py deleted file mode 100644 index 121257e..0000000 --- a/pyjenkinsci/utils/dates.py +++ /dev/null @@ -1,14 +0,0 @@ -import datetime - -MICROSECONDS_PER_SECOND = 1000000.0 -SECONDS_PER_DAY = 86400 - -def timedelta_to_seconds( td ): - assert isinstance( td, datetime.timedelta ) - seconds = 0.0 - - seconds += td.days * SECONDS_PER_DAY - seconds += td.seconds - seconds += td.microseconds / MICROSECONDS_PER_SECOND - - return seconds diff --git a/pyjenkinsci/utils/id.py b/pyjenkinsci/utils/id.py deleted file mode 100644 index 097debb..0000000 --- a/pyjenkinsci/utils/id.py +++ /dev/null @@ -1,16 +0,0 @@ -""" -Generate random IDs. -""" -import random - -ID_VALID = "abcdefghijklmnopqrstuvwxyz0123456789" - -def mk_id(length=5, prefix=""): - idchars = [] - for count in range( 0, length ): - idchars.append( random.choice( ID_VALID ) ) - return "%s%s" % ( prefix, "".join( idchars ) ) - -if __name__ == "__main__": - for i in range(0, 50): - print repr( mk_id( i ) ) diff --git a/pyjenkinsci/utils/junitxml.py b/pyjenkinsci/utils/junitxml.py deleted file mode 100644 index f838241..0000000 --- a/pyjenkinsci/utils/junitxml.py +++ /dev/null @@ -1,189 +0,0 @@ -import logging -import datetime -import traceback -import sys - -try: - from xml.etree import ElementTree as ET -except Exception, e: - import elementtree.ElementTree as ET - -from utils.dates import timedelta_to_seconds - -log = logging.getLogger(__name__) - -class junitxml( object ): - - ERROR = "error" - FAILURE = "failure" - - def __init__( self, stream, testsuite_name="test", ): - """ - Set up a new stream - """ - assert isinstance( testsuite_name, str ) - - self.xml = ET.Element("testsuite") - self.stream = stream - - self.xml.attrib["name"] = testsuite_name - - self.count_errors = 0 - self.count_tests = 0 - self.count_failures = 0 - - def __repr__(self): - return "<%s.%s %s>" % (self.__class__.__module__, self.__class__.__name__, str(self)) - - def __str__(self): - return "Stream: %s, Tests: %i Errors: %i, Failures %i" % ( repr( self.stream ), - self.count_tests, - self.count_errors, - self.count_failures ) - - @classmethod - def get_error_strings( cls, e ): - str_error_type = "%s.%s" % ( e.__class__.__module__, e.__class__.__name__ ) - str_error_args = ",".join( [repr(ee) for ee in e.args] ) - str_doc = str( e.__doc__ ).strip() - - return str_error_type, str_error_args, str_doc - - def write(self, xml_declaration=True, encoding="utf-8"): - self.xml.attrib["errors"] = str( self.count_errors ) - self.xml.attrib["failures"] = str( self.count_failures ) - self.xml.attrib["tests"] = str( self.count_tests ) - - ET.ElementTree( self.xml ).write( self.stream, encoding=encoding, xml_declaration=xml_declaration ) - log.warn( "Wrote Junit-style XML log to %s" % self.stream ) - - def assertTrue(self, classname, testname, errmsg, fn, *args, **kwargs ): - """ - Map the interface onto an assert like statement. - Also returns the value so that we can do useful things with the result - """ - - _testname = testname.replace( ".", "_") # Dots are not permitted in names' - - def assert_fn( ): - if callable(fn): - assert fn( *args, **kwargs ), errmsg - else: - assert len(args) == 0 and len(kwargs) == 0, "Object being tested is not callable and cannot have arguments." - assert fn, "errmsg" - - tr = self.startTest(classname, _testname) - return tr.run( assert_fn ) - - def startTest( self, classname, testname, ): - return junitxml_transaction( self, classname, testname ) - - def passTest( self, classname, name, test_time ): - self.addPass( classname, name, test_time) - - def failTest(self, classname, name, test_time, error, tb, mode=FAILURE ): - """ - Add a error - """ - str_error, str_error_msg, str_doc = self.get_error_strings( error ) - enhanced_tb = "%s: %s\n\n( %s )\n\n%s" % ( repr(error), str_error_msg, str_doc, tb ) - tc = self.addPass( classname, name, test_time) - self.convertPassToFail( tc, str_error, enhanced_tb, mode=mode ) - - - def addPass(self, classname, name, test_time=0.0, ): - """ - Add a pass - """ - assert isinstance( classname, str ) - assert isinstance( name, str ) - assert isinstance( test_time, (int, float) ) - self.count_tests += 1 - testcase = ET.SubElement( self.xml, "testcase" ) - testcase.attrib["classname"] = classname - testcase.attrib["name"] = name - testcase.attrib["time"] = "%.2f" % test_time - - return testcase - - def convertPassToFail( self, tc, failtype="", tb="", mode=FAILURE ): - """ - Add a failure - """ - assert isinstance( failtype, str ) - assert isinstance( tb, str ), "Traceback should be a string, got %s" % repr(tb) - assert mode in [ self.FAILURE, self.ERROR ] - - if mode == self.FAILURE: - self.count_errors += 1 - else: - self.count_failures += 1 - - failure = ET.SubElement( tc, mode ) - failure.text = tb - failure.attrib["type"] = failtype - return failure - - -class junitxml_transaction( object ): - def __init__(self, jxml, classname, testname ): - assert isinstance( jxml, junitxml ) - self.jxml = jxml - self.classname = classname - self.testname = testname - self.start_time = datetime.datetime.now() - - def getRuntime(self): - return timedelta_to_seconds( datetime.datetime.now() - self.start_time ) - - def run( self, fn, *args, **kwargs ): - try: - result = fn( *args, **kwargs ) - self.jxml.addPass( self.classname, self.testname, self.getRuntime() ) - except Exception, e: - ex_type, ex_value, ex_tb = sys.exc_info() - - tb_formatted = traceback.format_exception( ex_type, ex_value, ex_tb ) - str_tb = "\n".join( tb_formatted ) - str_ex = "%s.%s" % ( ex_value.__class__.__module__, ex_value.__class__.__name__ ) - runtime = self.getRuntime() - - if isinstance(e, AssertionError): - self.jxml.failTest( self.classname, self.testname, runtime, e, str_tb, mode=self.jxml.FAILURE ) - else: - self.jxml.failTest( self.classname, self.testname, runtime, e, str_tb, mode=self.jxml.ERROR ) - - log.exception(e) - - raise e - return result - -if __name__ == "__main__": - import sys - import time - import random - - logging.basicConfig() - logging.getLogger("").setLevel( logging.INFO ) - fod = junitxml( stream=sys.stdout ) - - def fn_test( mode ): - - time.sleep( random.random( ) ) - - if mode=="pass": - return 1 - elif mode=="fail": - assert False - elif mode=="error": - {}["x"] - - for testname in [ "pass", "fail", "error" ]: - t = fod.startTest("a", testname, ) - try: - t.run( fn_test, testname ) - except Exception, e: - #log.exception(e) - pass - - fod.write() diff --git a/pyjenkinsci/utils/xmlrunner.py b/pyjenkinsci/utils/xmlrunner.py deleted file mode 100644 index 2292719..0000000 --- a/pyjenkinsci/utils/xmlrunner.py +++ /dev/null @@ -1,244 +0,0 @@ -""" -XML Test Runner for PyUnit -""" - -# Written by Sebastian Rittau and placed in -# the Public Domain. -from utils import bufwrapper - -__revision__ = "$Id: /mirror/jroger/python/stdlib/xmlrunner.py 3506 2006-07-27T09:12:39.629878Z srittau $" - -import sys -import time -import traceback -import unittest -import logging -from StringIO import StringIO -from xml.sax.saxutils import escape - -log = logging.getLogger() - -class faketest( object ): - """ - A fake test object for when you want to inject additional results into the XML stream. - """ - failureException = AssertionError - - def __init__( self, id, exc_info ): - self._id = id - self._exc_info = exc_info - - def id(self): - return self._id - - def run(self, result): - result.startTest(self) - result.addError(self, self._exc_info ) - ok = False - result.stopTest(self) - - def __call__(self, *args, **kwds): - return self.run(*args, **kwds) - - -class _TestInfo(object): - """Information about a particular test. - Used by _XmlTestResult.""" - - def __init__( self, test, time, ): - (self._class, self._method) = test.id().rsplit(".", 1) - self._time = time - self._error = None - self._failure = None - self._console = "" - - @staticmethod - def create_success(test, time): - """Create a _TestInfo instance for a successful test.""" - return _TestInfo(test, time) - - @staticmethod - def create_failure(test, time, failure, console=""): - """Create a _TestInfo instance for a failed test.""" - info = _TestInfo(test, time) - info._failure = failure - info.console = console - return info - - @staticmethod - def create_error(test, time, error, console="" ): - """Create a _TestInfo instance for an erroneous test.""" - info = _TestInfo(test, time) - info._error = error - info.console = console - return info - - def print_report(self, stream): - """Print information about this test case in XML format to the - supplied stream. - """ - stream.write(' ' % \ - { - "class": self._class, - "method": self._method, - "time": self._time, - }) - if self._failure is not None: - self._print_error(stream, 'failure', self._failure) - if self._error is not None: - self._print_error(stream, 'error', self._error) - stream.write('\n') - - def _print_error(self, stream, tagname, error): - """Print information from a failure or error to the supplied stream.""" - text = escape(str(error[1])) - stream.write('\n') - stream.write(' <%s type="%s">%s\n%s\n' \ - % (tagname, str(error[0]), text, self.console )) - tb_stream = StringIO() - traceback.print_tb(error[2], None, tb_stream) - stream.write(escape(tb_stream.getvalue())) - stream.write(' \n' % tagname) - stream.write(' ') - - -class _XmlTestResult(unittest.TestResult): - """A test result class that stores result as XML. - - Used by XmlTestRunner. - """ - - test_count = 0 - - @classmethod - def get_test_serial( cls ): - cls.test_count += 1 - return cls.test_count - - def __init__(self, classname, consolestream =None ): - unittest.TestResult.__init__(self) - self._test_name = classname - self._start_time = None - self._tests = [] - self._error = None - self._failure = None - self._consolestream = consolestream - - def startTest(self, test): - unittest.TestResult.startTest(self, test) - - sn = self.get_test_serial() - - log.info( "Test %i: %s" % ( sn, test.id() ) ) - self._error = None - self._failure = None - self._start_time = time.time() - - def stopTest(self, test, time_taken = None ): - if time_taken is not None: - time_taken = time.time() - self._start_time - - str_console = self._consolestream.get_and_clear() - - unittest.TestResult.stopTest(self, test) - if self._error: - info = _TestInfo.create_error(test, time_taken, self._error, console=str_console ) - log.error( "Error: %s" % test.id() ) - elif self._failure: - info = _TestInfo.create_failure(test, time_taken, self._failure, console=str_console ) - log.error( "Fail: %s" % test.id() ) - else: - info = _TestInfo.create_success(test, time_taken, ) - log.debug( "OK: %s" % test.id() ) - self._tests.append(info) - - def addError(self, test, err): - log.warn( "Error: %s" % test.id() ) - unittest.TestResult.addError(self, test, err) - self._error = err - - def addFailure(self, test, err): - log.warn( "Failure: %s" % test.id() ) - unittest.TestResult.addFailure(self, test, err) - self._failure = err - - def print_report(self, stream, time_taken, out, err): - """Prints the XML report to the supplied stream. - - The time the tests took to perform as well as the captured standard - output and standard error streams must be passed in. - """ - stream.write('\n' % \ - { - "n": self._test_name, - "t": self.testsRun, - "time": time_taken, - }) - for info in self._tests: - info.print_report(stream) - stream.write(' \n' % out) - stream.write(' \n' % err) - stream.write('\n') - - -class XmlTestRunner(object): - """A test runner that stores results in XML format compatible with JUnit. - - XmlTestRunner(stream=None) -> XML test runner - - The XML file is written to the supplied stream. If stream is None, the - results are stored in a file called TEST-..xml in the - current working directory (if not overridden with the path property), - where and are the module and class name of the test class. - """ - def __init__(self, stream=None ): - self._stream = stream - - @staticmethod - def get_test_class_name_from_testobj( obj_test ): - class_ = obj_test.__class__ - classname = class_.__module__ + "." + class_.__name__ - return classname - - - def run(self, test, result=None ): - """Run the given test case or test suite.""" - classname = self.get_test_class_name_from_testobj( test ) - assert not self._stream is None - stream = self._stream - - # TODO: Python 2.5: Use the with statement - old_stdout = sys.stdout - old_stderr = sys.stderr - sys.stdout = bufwrapper( old_stdout ) - sys.stderr = bufwrapper( old_stderr ) - - if result is None: - result = _XmlTestResult( classname, consolestream = sys.stdout ) - else: - log.debug("Using provided XML test result object.") - - start_time = time.time() - - try: - test(result) - try: - out_s = sys.stdout.getvalue() - except AttributeError: - out_s = "" - try: - err_s = sys.stderr.getvalue() - except AttributeError: - err_s = "" - finally: - sys.stdout = old_stdout - sys.stderr = old_stderr - - time_taken = time.time() - start_time - result.print_report(stream, time_taken, out_s, err_s) - if self._stream is None: - stream.close() - - return result diff --git a/pyjenkinsci/utils/xmlrunnertest.py b/pyjenkinsci/utils/xmlrunnertest.py deleted file mode 100644 index d245284..0000000 --- a/pyjenkinsci/utils/xmlrunnertest.py +++ /dev/null @@ -1,144 +0,0 @@ -import unittest -import sys -import re -from cStringIO import StringIO -from utils.xmlrunner import XmlTestRunner - -class XmlTestRunnerTest(unittest.TestCase): - def setUp(self): - self._stream = StringIO() - - def _try_test_run(self, test_class, expected): - """Run the test suite against the supplied test class and compare the - XML result against the expected XML string. Fail if the expected - string doesn't match the actual string. All time attribute in the - expected string should have the value "0.000". All error and failure - messages are reduced to "Foobar". - """ - runner = XmlTestRunner(self._stream) - runner.run(unittest.makeSuite(test_class)) - - got = self._stream.getvalue() - # Replace all time="X.YYY" attributes by time="0.000" to enable a - # simple string comparison. - got = re.sub(r'time="\d+\.\d+"', 'time="0.000"', got) - # Likewise, replace all failure and error messages by a simple "Foobar" - # string. - got = re.sub(r'(?s).*?', r'Foobar', got) - got = re.sub(r'(?s).*?', r'Foobar', got) - - self.assertEqual(expected, got) - - def test_no_tests(self): - """Regression test: Check whether a test run without any tests - matches a previous run.""" - class TestTest(unittest.TestCase): - pass - self._try_test_run(TestTest, """ - - - -""") - - def test_success(self): - """Regression test: Check whether a test run with a successful test - matches a previous run.""" - class TestTest(unittest.TestCase): - def test_foo(self): - pass - self._try_test_run(TestTest, """ - - - - -""") - - def test_failure(self): - """Regression test: Check whether a test run with a failing test - matches a previous run.""" - class TestTest(unittest.TestCase): - def test_foo(self): - self.assert_(False) - self._try_test_run(TestTest, """ - - Foobar - - - - -""") - - def test_error(self): - """Regression test: Check whether a test run with a erroneous test - matches a previous run.""" - class TestTest(unittest.TestCase): - def test_foo(self): - raise IndexError() - self._try_test_run(TestTest, """ - - Foobar - - - - -""") - - def test_stdout_capture(self): - """Regression test: Check whether a test run with output to stdout - matches a previous run.""" - class TestTest(unittest.TestCase): - def test_foo(self): - print "Test" - self._try_test_run(TestTest, """ - - - - -""") - - def test_stderr_capture(self): - """Regression test: Check whether a test run with output to stderr - matches a previous run.""" - class TestTest(unittest.TestCase): - def test_foo(self): - print >>sys.stderr, "Test" - self._try_test_run(TestTest, """ - - - - -""") - - class NullStream(object): - """A file-like object that discards everything written to it.""" - def write(self, buffer): - pass - - def test_unittests_changing_stdout(self): - """Check whether the XmlTestRunner recovers gracefully from unit tests - that change stdout, but don't change it back properly. - """ - class TestTest(unittest.TestCase): - def test_foo(self): - sys.stdout = XmlTestRunnerTest.NullStream() - - runner = XmlTestRunner(self._stream) - runner.run(unittest.makeSuite(TestTest)) - - def test_unittests_changing_stderr(self): - """Check whether the XmlTestRunner recovers gracefully from unit tests - that change stderr, but don't change it back properly. - """ - class TestTest(unittest.TestCase): - def test_foo(self): - sys.stderr = XmlTestRunnerTest.NullStream() - - runner = XmlTestRunner(self._stream) - runner.run(unittest.makeSuite(TestTest)) - - -if __name__ == "__main__": - suite = unittest.makeSuite(XmlTestRunnerTest) - unittest.TextTestRunner().run(suite) -- 2.7.4