From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from eggs.gnu.org ([140.186.70.92]:60005) by lists.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1RnBbQ-0006rF-KD for qemu-devel@nongnu.org; Tue, 17 Jan 2012 11:09:47 -0500 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1RnBbN-0006tp-Jf for qemu-devel@nongnu.org; Tue, 17 Jan 2012 11:09:40 -0500 Received: from mail-vx0-f173.google.com ([209.85.220.173]:56306) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1RnBbN-0006th-Ap for qemu-devel@nongnu.org; Tue, 17 Jan 2012 11:09:37 -0500 Received: by vcbfk14 with SMTP id fk14so1586034vcb.4 for ; Tue, 17 Jan 2012 08:09:36 -0800 (PST) Sender: Paolo Bonzini Message-ID: <4F159D3A.2030707@redhat.com> Date: Tue, 17 Jan 2012 17:09:30 +0100 From: Paolo Bonzini MIME-Version: 1.0 References: <1326479558-3016-1-git-send-email-aliguori@us.ibm.com> <1326479558-3016-3-git-send-email-aliguori@us.ibm.com> In-Reply-To: <1326479558-3016-3-git-send-email-aliguori@us.ibm.com> Content-Type: multipart/mixed; boundary="------------060301050100070709010608" Subject: Re: [Qemu-devel] [PATCH 3/6] qtest: add C version of test infrastructure List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: qemu-devel , Anthony Liguori This is a multi-part message in MIME format. --------------060301050100070709010608 Content-Type: text/plain; charset=ISO-8859-1; format=flowed Content-Transfer-Encoding: 7bit On 01/13/2012 07:32 PM, Anthony Liguori wrote: > This also includes a qtest wrapper script to make it easier to launch qtest > tests directly. > > Signed-off-by: Anthony Liguori Here is a Python test harness for qtest. I haven't tried merging them with the makefiles. Feel free to add my s-o-b and include the files in your patches. Paolo --------------060301050100070709010608 Content-Type: text/x-python; name="gtest_main.py" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="gtest_main.py" #! /usr/bin/env python # # GTester-compatible main program for pyunit # # Copyright (C) 2011 Red Hat, Inc. # Author: Paolo Bonzini # # Permission to use, copy, modify, and/or distribute this software for any # purpose with or without fee is hereby granted, provided that the above # copyright notice and this permission notice appear in all copies. # # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF # OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. # This file defines replacements for unittest.main and unittest.TextTestRunner. # They are command-line compatible with glib test suites, including gtester # support, and even in standalone mode the output is similar to gtest-based # C unit tests. The only unsupported features are -m and --GTestSkipCount. # # Use this instead of unittest.main: # # if __name__ == "__main__": # gtest_main.main() # run all tests import sys import os import types import fnmatch import random import re import struct import time import traceback import unittest from unittest.signals import registerResult, installHandler # Ugly implementation dependency... used by TestResult to skip internal # frames in the traceback __unittest = True def test_id(test, module='__main__'): """Retrieve a GTest-like path from a TestCase.""" id = test.id().replace('.', '/') if module is not None: id = id[len(module.__name__):] return id class SelectTestsSuite(unittest.TestSuite): """A wrapper for other TestSuites that processes -p and -s options. Perhaps it could also use decorators to implement -m?""" re = None module = None def __init__(self, tests=(), module='__main__', path='*', skipPath=None): super(unittest.TestSuite, self).__init__(tests=tests) self.module = module regex = '' if path != '*': regex = fnmatch.translate(path) if skipPath is not None: regex = '(?!' + fnmatch.translate(skipPath) + ')' + regex if regex != '': self.re = re.compile(regex) def __iter__(self): iter = super(unittest.TestSuite, self).__iter__() if self.re is not None: iter = (x for x in iter if self.accept(x)) return iter def addTest(self, test): if isinstance(test, unittest.TestSuite): self.addTests(test) else: unittest.TestSuite.addTest(self, test) def accept(self, test): id = test_id(test, self.module) return self.re is None or self.re.match(id) class GTestResult(unittest.TestResult): """A test result class that can print formatted text results to a stream and can talk to gtester using the glib test protocol. Roughly based on TextTestResult, used internally by instances of GAbstractTestRunner.""" separator1 = '=' * 70 + "\n" separator2 = '-' * 70 + "\n" def defaultSeed(self): """Return a default random number seed. GLib expects this to be the string 'R02S' followed by four zero-padded 32-bit integers. We need to return a properly formatted seed so that the value can be passed to gtester even when Python and C tests are mixed on the command line.""" try: s = os.urandom(16) a, b, c, d = struct.unpack(">IIII", s) except NotImplementedError: t = time.time() a = int(t) b = int((t - a) * 1000000) c = os.getpid() d = os.name = 'posix' and os.getppid() or 0 return "R02S{0:>08x}{1:>08x}{2:>08x}{3:>08x}".format(a,b,c,d) def __init__(self, log, stream, progName=None, module='__main__', verbosity=1, seed=None): super(GTestResult, self).__init__() self.stream = stream self.log = log self.showAll = verbosity > 1 self.quiet = verbosity == 0 self.module = module self.seed = seed or self.defaultSeed() self.progName = progName or os.path.basename(sys.argv[0]) # These methods implement the glib test protocol. G_TEST_LOG_NONE = 0 G_TEST_LOG_ERROR = 1 # s:msg G_TEST_LOG_START_BINARY = 2 # s:binaryname s:seed G_TEST_LOG_LIST_CASE = 3 # s:testpath G_TEST_LOG_SKIP_CASE = 4 # s:testpath, TODO G_TEST_LOG_START_CASE = 5 # s:testpath G_TEST_LOG_STOP_CASE = 6 # d:status d:nforks d:elapsed G_TEST_LOG_MIN_RESULT = 7 # s:blurb d:result G_TEST_LOG_MAX_RESULT = 8 # s:blurb d:result G_TEST_LOG_MESSAGE = 9 # s:blurb def pack_log(self, typ, strings=(), nums=()): out = struct.pack(">iiii", typ, len(strings), len(nums), 0) for s in strings: out = out + struct.pack(">i", len(s)) + s for n in nums: out = out + struct.pack(">d", float(n)) out = struct.pack(">i", len(out) + 4) + out self.log.write(out) def logStartBinary(self): self.pack_log(self.G_TEST_LOG_START_BINARY, (self.progName, self.seed)) def logStartCase(self, test): id = test_id(test, self.module) self.pack_log(self.G_TEST_LOG_START_CASE, (id,)) def logListCase(self, test): id = test_id(test, self.module) self.pack_log(self.G_TEST_LOG_LIST_CASE, (id,)) def logError(self, msg): self.pack_log(self.G_TEST_LOG_ERROR, (msg,)) def logMessage(self, msg): self.pack_log(self.G_TEST_LOG_MESSAGE, (msg,)) def logStopCase(self, status): try: elapsed = time.clock() - self.startCaseTime self.pack_log(self.G_TEST_LOG_STOP_CASE, nums=(status,0,elapsed)) except: # This happens when class setup fails. startCaseTime has not # been set, do nothing assert status != 0 pass def logException(self, test, kind, err, skip=1): """Return a representation for the exception passed to addError or addFailure.""" type, inst, trace=err try: file, line, func, text = traceback.extract_tb(trace, skip+1)[skip] msg = "%s:%s:%s:%s: %s\n" % (kind, file, line, type.__name__, inst) except: msg = "%s:%s: %s\n" % (kind, type.__name__, inst) self.write("**\n" + msg) self.logError(msg + self.separator1 + self._exc_info_to_string(err, test)) # These methods implement text output. def write(self, str): self.stream.write(str) def flush(self): self.stream.flush() # These methods implement the standard TestResult protocol. def startTestRun(self): self.logStartBinary() self.startTime = time.clock() def stopTestRun(self): self.stopTime = time.clock() def startTest(self, test): super(GTestResult, self).startTest(test) random.seed(self.seed) self.startCaseTime = time.clock() self.logStartCase(test) if not self.quiet: self.write(test_id(test, self.module)) self.write(": ") self.flush() def addSuccess(self, test): super(GTestResult, self).addSuccess(test) self.logStopCase(0) if not self.quiet: self.write("OK\n") def addError(self, test, err): self.logException(test, "ERROR", err) self.logStopCase(1) super(GTestResult, self).addError(test, err) def addFailure(self, test, err): self.logException(test, "FAIL", err) self.logStopCase(1) super(GTestResult, self).addFailure(test, err) def addSkip(self, test, reason): self.logStopCase(0) super(GTestResult, self).addSkip(test, reason) if not self.quiet: self.write("SKIP {0!r}\n".format(reason)) def addExpectedFailure(self, test, err): self.logException("XFAIL", err) self.logStopCase(0) super(GTestResult, self).addExpectedFailure(test, err) if not self.quiet: self.write("XFAIL\n") def addUnexpectedSuccess(self, test): self.logError("unexpected success") self.logStopCase(1) super(GTestResult, self).addUnexpectedSuccess(test) if not self.quiet: self.write("XPASS\n") # Additional methods used by GTestLister and GTestRunner. def listTest(self, test): super(GTestResult, self).startTest(test) self.logListCase(test) if not self.quiet: self.write(test_id(test, self.module)) self.write("\n") self.flush() super(GTestResult, self).addSuccess(test) def printErrors(self): self.printErrorList('ERROR', self.errors) self.printErrorList('FAIL', self.failures) self.write(self.separator2) def printErrorList(self, kind, errors): for test, err in errors: self.write(self.separator1) self.write("%s: %s\n" % (kind,test_id(test, self.module))) self.write(self.separator2) self.write("%s\n" % err) def printSummary(self): run = self.testsRun timeTaken = self.stopTime - self.startTime if not self.quiet: self.write("Ran %d test%s in %.3fs\n\n" % (run, run != 1 and "s" or "", timeTaken)) infos = [] failed = len(self.failures) if failed: infos.append("failures=%d" % failed) errored = len(self.errors) if errored: infos.append("errors=%d" % errored) skipped = len(self.skipped) if skipped: infos.append("skipped=%d" % skipped) expectedFails = len(self.expectedFailures) if expectedFails: infos.append("expected failures=%d" % expectedFails) unexpectedSuccesses = len(self.unexpectedSuccesses) if unexpectedSuccesses: infos.append("unexpected successes=%d" % unexpectedSuccesses) if not self.wasSuccessful(): self.write("FAILED (%s)\n" % (", ".join(infos),)) elif infos: self.write("OK (%s)\n" % (", ".join(infos),)) elif not self.quiet: self.write("OK\n") def printResults(self): if self.quiet or self.showAll: if self.showAll: self.write("\n") self.printErrors() self.printSummary() class GAbstractTestRunner(object): """A test runner class that interfaces to a GTestResult. Compared to e.g. TextTestRunner, it can pass down some data that is required by the glib test protocol (program name, random number seed).""" def __init__(self, log=None, stream=sys.stdout, verbosity=1, progName=None, module='__main__', failfast=True, seed=None, buffer=False): self.module = module self.verbosity = verbosity self.failfast = failfast self.buffer = buffer self.progName = progName self.seed = seed class _DummyStream(object): def write(self, s): pass self.log = log or _DummyStream() self.stream = log and _DummyStream() or stream def _makeResult(self): return GTestResult(stream=self.stream, log=self.log, module=self.module, seed=self.seed, progName=self.progName, verbosity=self.verbosity) def run(self, test): "Run the given test case or test suite." result = self._makeResult() registerResult(result) result.failfast = self.failfast result.buffer = self.buffer startTestRun = getattr(result, 'startTestRun', None) if startTestRun is not None: startTestRun() try: self.doRun(test, result) finally: stopTestRun = getattr(result, 'stopTestRun', None) if stopTestRun is not None: stopTestRun() return result def doRun(self, test, result): "Run the given test case or test suite." test(result) class GTestLister(GAbstractTestRunner): """A test runner class that only prints the names of test cases. in the suite. When run in standalone mode it prints out the names of tests that have been selected. When run in gtester mode, it logs the same information using the glib test protocol.""" def doRun(self, test, result): """Run the given test case or test suite (actually just ask the GTestResult to list the test case).""" for t in test: result.listTest(t) class GTestRunner(GAbstractTestRunner): """A test runner class that emits results in textual and GLib form. When run in standalone mode it prints out the names of tests as they are run, errors as they occur, and a summary of the results at the end of the test run, depending on the verbosity level. When run in gtester mode, it logs the entire run using the glib test protocol.""" def run(self, test): "Run the given test case or test suite." result = super(GTestRunner, self).run(test) result.printResults() return result USAGE = """\ Usage: %(progName)s [options] Help Options: -?, --help Show help options Test Options: -l List test cases available in a test executable -seed=RANDOMSEED Provide a random seed to reproduce test runs using random numbers -v, --verbose Run tests verbosely -q, --quiet Run tests quietly -p TESTPATH execute all tests matching TESTPATH -s TESTPATH skip all tests matching TESTPATH -m {perf|slow|thorough|quick} (unsupported) Execute tests according to mode -k, --keep-going Keep running after the first failure --runner=CLASS Use an alternative test runner (e.g. unittest.TextTestRunner) --GTestLogFD=N file descriptor for communication with GTester --GTestSkipCount=N (unsupported) gtester-specific argument -p and -s accept slash-separated paths. They work even in combination with --runner. -l and --GTestLogFD are only supported without --runner. """ class GTestProgram(object): """A command-line program that runs a set of tests, used to make test modules conveniently executable and to interface them with gtester.""" failfast = True progName = None seed = None exit = True logFD = None verbosity = 1 path = '*' skipPath = None def getModule(self, module): if isinstance(module, basestring): m = __import__(module) for part in module.split('.')[1:]: m = getattr(m, part) return m else: return module def getClass(self, name): if isinstance(name, basestring): parts = name.split('.') module = ".".join(parts[:-1]) m = self.getModule(module) return getattr(m, parts[-1]) else: return name def __init__(self, module='__main__', argv=None, testRunner=None, testLoader=unittest.defaultTestLoader, exit=True, seed=None): argv = argv or sys.argv self.module = self.getModule(module) self.exit = exit self.seed = seed self.testRunner = testRunner self.testLoader = testLoader self.progName = os.path.basename(argv[0]) self.parseArgs(argv) self.suite = self.createTests() self.runTests() def usageExit(self, msg=None): if msg: print msg print USAGE % {'progName': self.progName} if self.exit: sys.exit(2) def parseArgs(self, argv): import getopt long_opts = ['seed=', 'keep-going', 'verbose', 'quiet', 'GTestLogFD=', 'runner=', 'GTestSkipCount='] list = False try: options, args = getopt.getopt(argv[1:], '?hHlvqs:p:m:k', long_opts) for opt, value in options: # quirk in the gtest option parser... if opt[1] != '-' and value != '' and value[0] == '=': value = value[1:] if opt in ('-h','-H','-?','--help'): self.usageExit() if opt in ('-q','--quiet'): self.verbosity = 0 if opt in ('-v','--verbose'): self.verbosity = 2 if opt in ('-k','--keep-going'): self.failfast = False if opt in ('-l'): list = True if opt in ('-p'): self.path = value if opt in ('-s'): self.skipPath = value if opt in ('--seed'): self.seed = value if opt in ('--runner'): self.testRunner = self.getClass(value) if opt in ('--GTestLogFD'): self.logFD = int(value) except getopt.error, msg: self.usageExit(msg) if self.testRunner is None: self.testRunner = list and GTestLister or GTestRunner else: # Set the seed now for user-specified runners. GTestRunners # resets it for each testcase, so that results are reproducible # even if other tests change. random.seed(self.seed) def createTests(self): allTests = self.testLoader.loadTestsFromModule(self.module) return SelectTestsSuite(tests=allTests, path=self.path, skipPath=self.skipPath, module=self.module) def runTests(self): installHandler() if isinstance(self.testRunner, type) and \ issubclass(self.testRunner, GAbstractTestRunner): # pass GTest-specific options to GAbstractTestRunner subclasses testRunner = self.testRunner(verbosity=self.verbosity, failfast=self.failfast, module=self.module, progName=self.progName, seed=self.seed, log=self.logFD and os.fdopen(self.logFD, 'w')) elif isinstance(self.testRunner, (type, types.ClassType)): try: testRunner = self.testRunner(verbosity=self.verbosity, failfast=self.failfast) except TypeError: testRunner = self.testRunner() else: # it is assumed to be a TestRunner instance testRunner = self.testRunner self.result = testRunner.run(self.suite) if self.exit: sys.exit(not self.result.wasSuccessful()) main = GTestProgram --------------060301050100070709010608 Content-Type: text/x-python; name="qtest.py" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="qtest.py" #! /usr/bin/env python # # qtest harness for PyUnit # # Copyright (C) 2011 Red Hat, Inc. # Author: Paolo Bonzini # # This file is licensed under the terms of the GNU GPL, version 2 or later. # See the COPYING file in the top-level directory. import os import struct import socket import unittest class ClientDisconnected(Exception): pass class QTestCase(unittest.TestCase): @classmethod def supportedArches(): return ['all'] @classmethod def extraArgs(): return () @classmethod def isSupportedArch(self, arch): return any(x == 'all' or x == arch for x in self.supportedArches()) @classmethod def _raiseIRQ(self, irq): self._irqs.add(irq) @classmethod def _lowerIRQ(self, irq): self._irqs.discard(irq) @classmethod def setUpClass(self): qemu_binary = os.getenv('QTEST_QEMU_BINARY') arch = os.path.basename(qemu_binary).replace('qemu-system-','') if not self.isSupportedArch(arch): self._classSetupFailed = True return socket_path = '/tmp/qtest-%d.sock' % os.getpid() server_socket = socket.socket(socket.AF_UNIX) server_socket.bind(socket_path) server_socket.listen(1) self.pid = os.spawnl(os.P_NOWAIT, qemu_binary, qemu_binary, '-qtest', 'unix:%s' % socket_path, '-qtest-log', '/dev/null', '-machine', 'accel=qtest', *self.extraArgs()) self._socket, addr = server_socket.accept() self._irqs = set() self._buffer = '' self._reply = None server_socket.close() @classmethod def tearDownClass(self): self._socket.close() self._socket = None os.kill(self.pid, 15) os.waitpid(self.pid, 0) @classmethod def _qtestResponse(self, buf): tokens = buf.split() if tokens[0] == 'OK': self._reply = tokens return if tokens[0] == 'IRQ': if tokens[1] == 'raise': self._raiseIRQ(int(tokens[2])) else: self._lowerIRQ(int(tokens[2])) return @classmethod def waitForEvents(self, blocking=True): self._socket.setblocking(int(blocking)) while True: i = self._buffer.find('\n') if i != -1: response, self._buffer = self._buffer[:i], self._buffer[i+1:] self._qtestResponse(response) self._socket.setblocking(0) continue try: data = self._socket.recv(4096) self._buffer = self._buffer + data if (len(data) == 0): raise ClientDisconnected except: return @classmethod def _send(self, str): assert self._reply is None self._socket.sendall(str + '\n') while self._reply is None: self.waitForEvents() reply, self._reply = self._reply, None return reply # IRQ access def getIRQ(self, irq): self.waitForEvents(False) return irq in self._irqs def waitIRQ(self, irq, level=True): self.waitForEvents(False) while (irq in self._irqs) != level: self.waitForEvents(True) # PIO reads def inb(self, port): return int(self._send('inb 0x%x' % port)[1][2:], 16) def inw(self, port): return int(self._send('inw 0x%x' % port)[1][2:], 16) def inl(self, port): return int(self._send('inl 0x%x' % port)[1][2:], 16) # Memory reads def memread_unpack(self, addr, format): size = struct.calcsize(format) str = 'read 0x%x 0x%x' % (addr, size) reply = self._send(str)[1] bytes = reply[2:].decode('hex_codec') return struct.unpack(format, bytes) def memread(self, addr, size): return self.memread_unpack('%dB' % size) def ldb(self, addr): return self.memread_unpack(addr, 'B')[0] def ldw_le(self, addr): return self.memread_unpack(addr, 'H')[0] def ldl_be(self, addr): return self.memread_unpack(addr, '>I')[0] # PIO writes def outb(self, port, val): self._send('outb 0x%x 0x%x' % (port, val)) def outw(self, port, val): self._send('outw 0x%x 0x%x' % (port, val)) def outl(self, port, val): self._send('outl 0x%x 0x%x' % (port, val)) # Memory writes def memwrite_pack(self, addr, format, *data): bytes = struct.pack(format, *data) str = 'write 0x%x 0x%x 0x%s' % ( addr, len(bytes), bytes.encode('hex_codec')) self._send(str) def memwrite(self, addr, *data): self.memwrite_pack('%dB' % len(data), *data) def stb(self, addr, datum): self.memwrite_pack(addr, 'B', datum) def stw_le(self, addr, datum): self.memwrite_pack(addr, 'H', datum) def stl_be(self, addr, datum): self.memwrite_pack(addr, '>I', datum) --------------060301050100070709010608 Content-Type: text/x-python; name="rtc-test.py" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="rtc-test.py" #! /usr/bin/env python # # Sample qtest written in Python # # Copyright (C) 2011 Red Hat, Inc. # Author: Paolo Bonzini # # This file is licensed under the terms of the GNU GPL, version 2 or later. # See the COPYING file in the top-level directory. import time import qtest import gtest_main class QTestCase(qtest.QTestCase): @classmethod def supportedArches(self): return ['i386', 'x86_64'] @classmethod def extraArgs(self): return ('-vga', 'none') BASE = 0x70 RTC_SECONDS = 0 RTC_MINUTES = 2 RTC_HOURS = 4 def read(self, reg): self.outb(self.BASE, reg) return self.inb(self.BASE + 1) def write(self, reg, value): self.outb(self.BASE, reg) self.outb(self.BASE + 1, value) def testRead(self): self.write(self.RTC_HOURS, 0x01) self.write(self.RTC_MINUTES, 0x15) self.write(self.RTC_SECONDS, 0x29) assert self.read(self.RTC_MINUTES) == 0x15 assert self.read(self.RTC_HOURS) == 0x01 def testUpdate(self): self.write(self.RTC_HOURS, 0x01) self.write(self.RTC_MINUTES, 0x15) self.write(self.RTC_SECONDS, 0x29) while self.read(self.RTC_SECONDS) == 0x29: time.sleep(0.2) assert self.read(self.RTC_SECONDS) == 0x30 if __name__ == '__main__': gtest_main.main() --------------060301050100070709010608--