openstack/deb-python-pyngus
Code Issues Proposed changes

Add unit test infrastructure, plus a few bugfixes and PEP8 cleanups.

Browse Source
This commit is contained in:
Kenneth Giusti 2014-03-19 13:27:02 -04:00
parent 33dfb83748
commit 684a31d057
19 changed files with 917 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -38,3 +38,4 @@ nosetests.xml
# emacs crap
*~
MANIFEST

9
examples/python/recv.py
View File

@ -19,15 +19,14 @@
#
""" Minimal message receive example code."""
import optparse, sys, time, uuid
import re, socket, select, errno
import optparse
import sys
import uuid
from proton import Message
import fusion
from utils import connect_socket
from utils import get_host_port
from utils import process_connection
from utils import SEND_STATUS
def main(argv=None):
@ -76,6 +75,7 @@ def main(argv=None):
self.done = False
self.message = None
self.handle = None
def message_received(self, receiver, message, handle):
self.done = True
self.message = message
@ -112,4 +112,3 @@ def main(argv=None):
if __name__ == "__main__":
sys.exit(main())

27
examples/python/rpc-client.py
View File

@ -22,9 +22,15 @@ form: {'method': '<name of method on server>', 'args': {<map of name=value
arguments for the call} }
"""
import optparse, sys, time, uuid
import re, socket, select, errno
import errno
import logging
import optparse
import re
import socket
import select
import sys
import time
import uuid
from proton import Message
import fusion
@ -33,7 +39,6 @@ LOG = logging.getLogger()
LOG.addHandler(logging.StreamHandler())
class MyConnection(fusion.ConnectionEventHandler):
def __init__(self, name, container, properties):
@ -340,7 +345,7 @@ def main(argv=None):
if not addr:
raise Exception("Could not translate address '%s'" % opts.server)
my_socket = socket.socket(addr[0][0], addr[0][1], addr[0][2])
my_socket.setblocking(0) # 0=non-blocking
my_socket.setblocking(0) # 0=non-blocking
try:
my_socket.connect(addr[0][4])
except socket.error, e:
@ -356,18 +361,17 @@ def main(argv=None):
if opts.ca:
conn_properties["x-ssl-ca-file"] = opts.ca
my_connection = MyConnection( "to-server", container, conn_properties)
my_connection = MyConnection("to-server", container, conn_properties)
# Create the RPC caller
method = {'method': method_info[0],
'args': dict([(method_info[i], method_info[i+1])
for i in range(1, len(method_info), 2)])}
my_caller = my_connection.create_caller( method,
"my-source-address",
"my-target-address",
receiver_properties={},
sender_properties={})
my_caller = my_connection.create_caller(method,
"my-source-address",
"my-target-address",
receiver_properties={},
sender_properties={})
try:
my_connection.connect(my_socket)
repeat = 0
@ -407,4 +411,3 @@ def main(argv=None):
if __name__ == "__main__":
sys.exit(main())

29
examples/python/rpc-server.py
View File

@ -27,9 +27,15 @@ The server replies to the client using a map that contains a copy of the method
map sent in the request.
"""
import optparse, sys, time, uuid
import re, socket, select, errno
import errno
import logging
import optparse
import re
import socket
import select
import sys
import time
import uuid
#import gc
from guppy import hpy
@ -50,7 +56,7 @@ receiver_links = {}
reply_senders = {}
# database of all active SocketConnections
socket_connections = {} # indexed by name
socket_connections = {} # indexed by name
class SocketConnection(fusion.ConnectionEventHandler):
@ -84,8 +90,7 @@ class SocketConnection(fusion.ConnectionEventHandler):
def process_input(self):
"""Called when socket is read-ready"""
try:
rc = fusion.read_socket_input(self.connection,
self.socket)
fusion.read_socket_input(self.connection, self.socket)
except Exception as e:
LOG.error("Exception on socket read: %s", str(e))
# may be redundant if closed cleanly:
@ -268,7 +273,7 @@ class MyReceiverLink(fusion.ReceiverEventHandler):
correlation_id = message.correlation_id
method_map = message.body
if (not isinstance(method_map, dict) or
'method' not in method_map):
'method' not in method_map):
LOG.error("no method given, map=%s", str(method_map))
self._link.message_rejected(handle, "Bad format")
else:
@ -291,7 +296,7 @@ class MyReceiverLink(fusion.ReceiverEventHandler):
if self._link.capacity == 0:
LOG.debug("increasing credit...")
self._link.add_capacity( 5 )
self._link.add_capacity(5)
def main(argv=None):
@ -333,7 +338,7 @@ def main(argv=None):
if not addr:
raise Exception("Could not translate address '%s'" % opts.address)
my_socket = socket.socket(addr[0][0], addr[0][1], addr[0][2])
my_socket.setblocking(0) # 0=non-blocking
my_socket.setblocking(0) # 0=non-blocking
try:
my_socket.bind((host, port))
my_socket.listen(10)
@ -368,12 +373,15 @@ def main(argv=None):
timeout = None
if timers:
deadline = timers[0].next_tick # 0 == next expiring timer
deadline = timers[0].next_tick # 0 == next expiring timer
now = time.time()
timeout = 0 if deadline <= now else deadline - now
LOG.debug("select() start (t=%s)", str(timeout))
readable, writable, ignore = select.select(readfd, writefd, [], timeout)
readable, writable, ignore = select.select(readfd,
writefd,
[],
timeout)
LOG.debug("select() returned")
worked = []
@ -434,4 +442,3 @@ def main(argv=None):
if __name__ == "__main__":
sys.exit(main())

7
examples/python/send.py
View File

@ -19,8 +19,9 @@
#
""" Minimal message send example code."""
import optparse, sys, time, uuid
import re, socket, select, errno
import optparse
import sys
import uuid
from proton import Message
import fusion
@ -87,6 +88,7 @@ def main(argv=None):
def __init__(self):
self.done = False
self.status = None
def __call__(self, link, handle, status, error):
self.done = True
self.status = status
@ -116,4 +118,3 @@ def main(argv=None):
if __name__ == "__main__":
sys.exit(main())

18
examples/python/server.py
View File

@ -19,9 +19,12 @@
#
"""A simple server that consumes and produces messages."""
import optparse, sys, time, uuid
import re, socket, select, errno
import logging
import optparse
import select
import sys
import time
import uuid
from proton import Message
import fusion
@ -71,8 +74,7 @@ class SocketConnection(fusion.ConnectionEventHandler):
def process_input(self):
"""Called when socket is read-ready"""
try:
rc = fusion.read_socket_input(self.connection,
self.socket)
fusion.read_socket_input(self.connection, self.socket)
except Exception as e:
LOG.error("Exception on socket read: %s", str(e))
# may be redundant if closed cleanly:
@ -194,6 +196,7 @@ class MySenderLink(fusion.SenderEventHandler):
# send another message:
self.send_message()
class MyReceiverLink(fusion.ReceiverEventHandler):
"""Receive messages, and drop them."""
def __init__(self, socket_conn, handle, rx_addr=None):
@ -232,6 +235,7 @@ class MyReceiverLink(fusion.ReceiverEventHandler):
print("Message received on Receiver link %s, message=%s"
% (self.receiver_link.name, str(message)))
def main(argv=None):
_usage = """Usage: %prog [options]"""
@ -277,13 +281,14 @@ def main(argv=None):
timeout = None
if timers:
deadline = timers[0].next_tick # [0] == next expiring timer
deadline = timers[0].next_tick # [0] == next expiring timer
now = time.time()
timeout = 0 if deadline <= now else deadline - now
LOG.debug("select() start (t=%s)", str(timeout))
readfd.append(my_socket)
readable, writable, ignore = select.select(readfd, writefd, [], timeout)
readable, writable, ignore = select.select(readfd, writefd,
[], timeout)
LOG.debug("select() returned")
worked = []
@ -344,4 +349,3 @@ def main(argv=None):
if __name__ == "__main__":
sys.exit(main())

17
examples/python/utils.py
View File

@ -18,11 +18,15 @@
#
"""Utilities used by the Examples"""
import optparse, sys, time, uuid
import re, socket, select, errno
import errno
import re
import socket
import select
import time
import fusion
def get_host_port(server_address):
"""Parse the hostname and port out of the server_address."""
regex = re.compile(r"^amqp://([a-zA-Z0-9.]+)(:([\d]+))?$")
@ -39,7 +43,8 @@ def connect_socket(host, port, blocking=True):
"""Create a TCP connection to the server."""
addr = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)
if not addr:
raise Exception("Could not translate address '%s:%s'" % (host, str(port)))
raise Exception("Could not translate address '%s:%s'"
% (host, str(port)))
my_socket = socket.socket(addr[0][0], addr[0][1], addr[0][2])
if not blocking:
my_socket.setblocking(0)
@ -50,11 +55,13 @@ def connect_socket(host, port, blocking=True):
raise
return my_socket
def server_socket(host, port, backlog=10):
"""Create a TCP listening socket for a server."""
addr = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)
if not addr:
raise Exception("Could not translate address '%s:%s'" % (host, str(port)))
raise Exception("Could not translate address '%s:%s'"
% (host, str(port)))
my_socket = socket.socket(addr[0][0], addr[0][1], addr[0][2])
my_socket.setblocking(0) # 0=non-blocking
try:
@ -65,6 +72,7 @@ def server_socket(host, port, backlog=10):
raise
return my_socket
def process_connection(connection, my_socket):
"""Handle I/O and Timers on a single Connection."""
work = False
@ -108,4 +116,3 @@ SEND_STATUS = {
fusion.SenderLink.RELEASED: "RELEASED",
fusion.SenderLink.MODIFIED: "MODIFIED"
}

12
python/__init__.py
View File

@ -16,9 +16,9 @@
# specific language governing permissions and limitations
# under the License.
#
from container import Container, ContainerEventHandler
from connection import Connection, ConnectionEventHandler
from link import ReceiverLink, ReceiverEventHandler
from link import SenderLink, SenderEventHandler
from sockets import read_socket_input
from sockets import write_socket_output
from fusion.container import Container, ContainerEventHandler
from fusion.connection import Connection, ConnectionEventHandler
from fusion.link import ReceiverLink, ReceiverEventHandler
from fusion.link import SenderLink, SenderEventHandler
from fusion.sockets import read_socket_input
from fusion.sockets import write_socket_output

6
python/connection.py
View File

@ -23,7 +23,7 @@ __all__ = [
import logging
import time
from link import _SessionProxy
from fusion.link import _SessionProxy
import proton
@ -74,6 +74,7 @@ class ConnectionEventHandler(object):
"""SASL exchange complete."""
LOG.debug("sasl_done (ignored)")
class Connection(object):
"""A Connection to a peer."""
EOS = -1 # indicates 'I/O stream closed'
@ -295,7 +296,7 @@ class Connection(object):
next_pn_link = pn_link.next(self._REMOTE_REQ)
session = pn_link.session.context
if (pn_link.is_sender and
pn_link.name not in self._sender_links):
pn_link.name not in self._sender_links):
LOG.debug("Remotely initiated Sender needs init")
link = session.request_sender(pn_link)
self._sender_links[pn_link.name] = link
@ -357,7 +358,6 @@ class Connection(object):
if self._handler:
self._handler.connection_closed(self)
# DEBUG LINK "LEAK"
# count = 0
# link = self._pn_connection.link_head(0)

2
python/container.py
View File

@ -22,7 +22,7 @@ __all__ = [
import heapq
import logging
from connection import Connection
from fusion.connection import Connection
LOG = logging.getLogger(__name__)

3
python/link.py
View File

@ -28,6 +28,7 @@ import proton
LOG = logging.getLogger(__name__)
class _Link(object):
"""A generic Link base class."""
@ -263,6 +264,7 @@ _Link._STATE_MAP = [ # {event: (next-state, action), ...}
#_STATE_REJECTED:
{}]
class SenderEventHandler(object):
def sender_active(self, sender_link):
LOG.debug("sender_active (ignored)")
@ -561,6 +563,7 @@ class ReceiverLink(_Link):
{"source-address":
pn_link.remote_source.address})
class _SessionProxy(object):
"""Corresponds to a Proton Session object."""
def __init__(self, connection, pn_session=None):

2
python/sockets.py
View File

@ -27,7 +27,7 @@ import errno
import logging
import socket
from connection import Connection
from fusion.connection import Connection
LOG = logging.getLogger(__name__)

3
test-requirements.txt Normal file
View File

@ -0,0 +1,3 @@
pep8>=1.4.5
pyflakes>=0.7.3
flake8>=2.1.0

22
tests/python/README.md Normal file
View File

@ -0,0 +1,22 @@
# Unit Tests #
The unit test are run using the `tox` command from the [tox automation
project](https://testrun.org/tox/latest) for more detail. It is
assumed tox is available on your system.
To run the tests:
1. The Proton Python binding must be installed [1]
2. From the topmost directory (that contains the tox.ini file) execute the *tox* command
3. That's it!
The unit tests may be run without tox, but your environment must be
set up so that the module is able to be imported by the test scripts.
For example, setting the PYTHONPATH environment variable to include
the path to the build directory.
[1] You can still run the tests without installing Proton. You can
run the tests using a build of the Proton sources instead. In order
to do this you **MUST** source the config.sh file supplied in the
Proton sources prior to running the unit tests. This script adds the
Proton Python bindings to the search path used by the unit tests.

690
tests/python/test-runner Executable file
View File

@ -0,0 +1,690 @@
#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
# KGIUSTI: mooched from the Proton project's python-test runner
# TODO: summarize, test harness preconditions (e.g. broker is alive)
from __future__ import print_function
import logging, optparse, os, struct, sys, time, traceback, types, cgi
from fnmatch import fnmatchcase as match
from getopt import GetoptError
from logging import getLogger, StreamHandler, Formatter, Filter, \
WARN, DEBUG, ERROR
levels = {
"DEBUG": DEBUG,
"WARN": WARN,
"ERROR": ERROR
}
sorted_levels = [(v, k) for k, v in levels.items()]
sorted_levels.sort()
sorted_levels = [v for k, v in sorted_levels]
parser = optparse.OptionParser(usage="usage: %prog [options] PATTERN ...",
description="Run tests matching the specified PATTERNs.")
parser.add_option("-l", "--list", action="store_true", default=False,
help="list tests instead of executing them")
parser.add_option("-f", "--log-file", metavar="FILE", help="log output to FILE")
parser.add_option("-v", "--log-level", metavar="LEVEL", default="WARN",
help="only display log messages of LEVEL or higher severity: "
"%s (default %%default)" % ", ".join(sorted_levels))
parser.add_option("-c", "--log-category", metavar="CATEGORY", action="append",
dest="log_categories", default=[],
help="log only categories matching CATEGORY pattern")
parser.add_option("-m", "--module", action="append", default=[],
dest="modules", help="add module to test search path")
parser.add_option("-i", "--ignore", action="append", default=[],
help="ignore tests matching IGNORE pattern")
parser.add_option("-I", "--ignore-file", metavar="IFILE", action="append",
default=[],
help="ignore tests matching patterns in IFILE")
parser.add_option("-H", "--halt-on-error", action="store_true", default=False,
dest="hoe", help="halt if an error is encountered")
parser.add_option("-t", "--time", action="store_true", default=False,
help="report timing information on test run")
parser.add_option("-D", "--define", metavar="DEFINE", dest="defines",
action="append", default=[], help="define test parameters")
parser.add_option("-x", "--xml", metavar="XML", dest="xml",
help="write test results in Junit style xml suitable for use by CI tools etc")
parser.add_option("-a", "--always-colorize", action="store_true", dest="always_colorize", default=False,
help="always colorize the test results rather than relying on terminal tty detection. Useful when invoked from Jython/Maven.")
parser.add_option("-n", metavar="count", dest="count", type=int, default=1,
help="run the tests <count> times")
parser.add_option("-b", "--bare", action="store_true", default=False,
help="Run bare, i.e. don't capture stack traces. This is useful under Jython as " +
"captured stack traces do not include the Java portion of the stack," +
"whereas non captured stack traces do.")
class Config:
def __init__(self):
self.defines = {}
self.log_file = None
self.log_level = WARN
self.log_categories = []
opts, args = parser.parse_args()
includes = []
excludes = ["*__*__"]
config = Config()
list_only = opts.list
for d in opts.defines:
try:
idx = d.index("=")
name = d[:idx]
value = d[idx+1:]
config.defines[name] = value
except ValueError:
config.defines[d] = None
config.log_file = opts.log_file
config.log_level = levels[opts.log_level.upper()]
config.log_categories = opts.log_categories
excludes.extend([v.strip() for v in opts.ignore])
for v in opts.ignore_file:
f = open(v)
for line in f:
line = line.strip()
if line.startswith("#"):
continue
excludes.append(line)
f.close()
for a in args:
includes.append(a.strip())
if not includes:
includes.append("*")
def is_ignored(path):
for p in excludes:
if match(path, p):
return True
return False
def is_included(path):
if is_ignored(path):
return False
for p in includes:
if match(path, p):
return True
return False
def is_smart():
return sys.stdout.isatty() and os.environ.get("TERM", "dumb") != "dumb"
try:
import fcntl, termios
def width():
if is_smart():
s = struct.pack("HHHH", 0, 0, 0, 0)
fd_stdout = sys.stdout.fileno()
x = fcntl.ioctl(fd_stdout, termios.TIOCGWINSZ, s)
rows, cols, xpx, ypx = struct.unpack("HHHH", x)
return cols
else:
try:
return int(os.environ.get("COLUMNS", "80"))
except ValueError:
return 80
WIDTH = width()
def resize(sig, frm):
global WIDTH
WIDTH = width()
import signal
signal.signal(signal.SIGWINCH, resize)
except ImportError:
WIDTH = 80
def vt100_attrs(*attrs):
return "\x1B[%sm" % ";".join(map(str, attrs))
vt100_reset = vt100_attrs(0)
KEYWORDS = {"pass": (32,),
"skip": (33,),
"fail": (31,),
"start": (34,),
"total": (34,),
"ignored": (33,),
"selected": (34,),
"elapsed": (34,),
"average": (34,)}
def colorize_word(word, text=None):
if text is None:
text = word
return colorize(text, *KEYWORDS.get(word, ()))
def colorize(text, *attrs):
if attrs and (is_smart() or opts.always_colorize):
return "%s%s%s" % (vt100_attrs(*attrs), text, vt100_reset)
else:
return text
def indent(text):
lines = text.split("\n")
return " %s" % "\n ".join(lines)
# Write a 'minimal' Junit xml style report file suitable for use by CI tools such as Jenkins.
class JunitXmlStyleReporter:
def __init__(self, file):
self.f = open(file, "w");
def begin(self):
self.f.write('<?xml version="1.0" encoding="UTF-8" ?>\n')
self.f.write('<testsuite name="pythontests">\n')
def report(self, name, result):
parts = name.split(".")
method = parts[-1]
module = '.'.join(parts[0:-1])
self.f.write('<testcase classname="%s" name="%s" time="%f">\n' % (module, method, result.time))
if result.failed:
escaped_type = cgi.escape(str(result.exception_type))
escaped_message = cgi.escape(str(result.exception_message))
self.f.write('<failure type="%s" message="%s">\n' % (escaped_type, escaped_message))
self.f.write('<![CDATA[\n')
self.f.write(result.formatted_exception_trace)
self.f.write(']]>\n')
self.f.write('</failure>\n')
if result.skipped:
self.f.write('<skipped/>\n')
self.f.write('</testcase>\n')
def end(self):
self.f.write('</testsuite>\n')
self.f.close()
class Interceptor:
def __init__(self):
self.newline = False
self.indent = False
self.passthrough = True
self.dirty = False
self.last = None
def begin(self):
self.newline = True
self.indent = True
self.passthrough = False
self.dirty = False
self.last = None
def reset(self):
self.newline = False
self.indent = False
self.passthrough = True
class StreamWrapper:
def __init__(self, interceptor, stream, prefix=" "):
self.interceptor = interceptor
self.stream = stream
self.prefix = prefix
def fileno(self):
return self.stream.fileno()
def isatty(self):
return self.stream.isatty()
def write(self, s):
if self.interceptor.passthrough:
self.stream.write(s)
return
if s:
self.interceptor.dirty = True
if self.interceptor.newline:
self.interceptor.newline = False
self.stream.write(" %s\n" % colorize_word("start"))
self.interceptor.indent = True
if self.interceptor.indent:
self.stream.write(self.prefix)
if s.endswith("\n"):
s = s.replace("\n", "\n%s" % self.prefix)[:-2]
self.interceptor.indent = True
else:
s = s.replace("\n", "\n%s" % self.prefix)
self.interceptor.indent = False
self.stream.write(s)
if s:
self.interceptor.last = s[-1]
def flush(self):
self.stream.flush()
interceptor = Interceptor()
out_wrp = StreamWrapper(interceptor, sys.stdout)
err_wrp = StreamWrapper(interceptor, sys.stderr)
out = sys.stdout
err = sys.stderr
sys.stdout = out_wrp
sys.stderr = err_wrp
class PatternFilter(Filter):
def __init__(self, *patterns):
Filter.__init__(self, patterns)
self.patterns = patterns
def filter(self, record):
if not self.patterns:
return True
for p in self.patterns:
if match(record.name, p):
return True
return False
root = getLogger()
handler = StreamHandler(sys.stdout)
filter = PatternFilter(*config.log_categories)
handler.addFilter(filter)
handler.setFormatter(Formatter("%(asctime)s %(levelname)s %(message)s"))
root.addHandler(handler)
root.setLevel(WARN)
log = getLogger("unit.test")
PASS = "pass"
SKIP = "skip"
FAIL = "fail"
class Runner:
def __init__(self):
self.exception = None
self.exception_phase_name = None
self.skip = False
def passed(self):
return not self.exception
def skipped(self):
return self.skip
def failed(self):
return self.exception and not self.skip
def halt(self):
"""determines if the overall test execution should be allowed to continue to the next phase"""
return self.exception or self.skip
def run(self, phase_name, phase):
"""invokes a test-phase method (which can be the test method itself or a setup/teardown
method). If the method raises an exception the exception is examined to see if the
exception should be classified as a 'skipped' test"""
# we don't try to catch exceptions for jython because currently a
# jython bug will prevent the java portion of the stack being
# stored with the exception info in the sys module
if opts.bare:
phase()
else:
try:
phase()
except KeyboardInterrupt:
raise
except:
self.exception_phase_name = phase_name
self.exception = sys.exc_info()
exception_type = self.exception[0]
self.skip = getattr(exception_type, "skipped", False)
def status(self):
if self.passed():
return PASS
elif self.skipped():
return SKIP
elif self.failed():
return FAIL
else:
return None
def get_formatted_exception_trace(self):
if self.exception:
if self.skip:
# format skipped tests without a traceback
output = indent("".join(traceback.format_exception_only(*self.exception[:2]))).rstrip()
else:
output = "Error during %s:" % self.exception_phase_name
output += indent("".join(traceback.format_exception(*self.exception))).rstrip()
return output
def get_exception_type(self):
if self.exception:
return self.exception[0]
else:
return None
def get_exception_message(self):
if self.exception:
return self.exception[1]
else:
return None
ST_WIDTH = 8
def run_test(name, test, config):
patterns = filter.patterns
level = root.level
filter.patterns = config.log_categories
root.setLevel(config.log_level)
parts = name.split(".")
line = None
output = ""
for part in parts:
if line:
if len(line) + len(part) >= (WIDTH - ST_WIDTH - 1):
output += "%s. \\\n" % line
line = " %s" % part
else:
line = "%s.%s" % (line, part)
else:
line = part
if line:
output += "%s %s" % (line, (((WIDTH - ST_WIDTH) - len(line))*"."))
sys.stdout.write(output)
sys.stdout.flush()
interceptor.begin()
start = time.time()
try:
runner = test()
finally:
interceptor.reset()
end = time.time()
if interceptor.dirty:
if interceptor.last != "\n":
sys.stdout.write("\n")
sys.stdout.write(output)
print(" %s" % colorize_word(runner.status()))
if runner.failed() or runner.skipped():
print(runner.get_formatted_exception_trace())
root.setLevel(level)
filter.patterns = patterns
return TestResult(end - start,
runner.passed(),
runner.skipped(),
runner.failed(),
runner.get_exception_type(),
runner.get_exception_message(),
runner.get_formatted_exception_trace())
class TestResult:
def __init__(self, time, passed, skipped, failed, exception_type, exception_message, formatted_exception_trace):
self.time = time
self.passed = passed
self.skipped = skipped
self.failed = failed
self.exception_type = exception_type
self.exception_message = exception_message
self.formatted_exception_trace = formatted_exception_trace
class FunctionTest:
def __init__(self, test):
self.test = test
def name(self):
return "%s.%s" % (self.test.__module__, self.test.__name__)
def run(self):
return run_test(self.name(), self._run, config)
def _run(self):
runner = Runner()
runner.run("test", lambda: self.test(config))
return runner
def __repr__(self):
return "FunctionTest(%r)" % self.test
class MethodTest:
def __init__(self, cls, method):
self.cls = cls
self.method = method
def name(self):
return "%s.%s.%s" % (self.cls.__module__, self.cls.__name__, self.method)
def run(self):
return run_test(self.name(), self._run, config)
def _run(self):
runner = Runner()
inst = self.cls(self.method)
test = getattr(inst, self.method)
if hasattr(inst, "configure"):
runner.run("configure", lambda: inst.configure(config))
if runner.halt(): return runner
if hasattr(inst, "setUp"):
runner.run("setup", inst.setUp)
if runner.halt(): return runner
elif hasattr(inst, "setup"):
runner.run("setup", inst.setup)
if runner.halt(): return runner
runner.run("test", test)
if hasattr(inst, "tearDown"):
runner.run("teardown", inst.tearDown)
elif hasattr(inst, "teardown"):
runner.run("teardown", inst.teardown)
return runner
def __repr__(self):
return "MethodTest(%r, %r)" % (self.cls, self.method)
class PatternMatcher:
def __init__(self, *patterns):
self.patterns = patterns
def matches(self, name):
for p in self.patterns:
if match(name, p):
return True
return False
class FunctionScanner(PatternMatcher):
def inspect(self, obj):
return type(obj) == types.FunctionType and self.matches(obj.__name__)
def descend(self, func):
# the None is required for older versions of python
return; yield None
def extract(self, func):
yield FunctionTest(func)
class ClassScanner(PatternMatcher):
def inspect(self, obj):
return type(obj) in (types.ClassType, types.TypeType) and self.matches(obj.__name__)
def descend(self, cls):
# the None is required for older versions of python
return; yield None
def extract(self, cls):
names = dir(cls)
names.sort()
for name in names:
obj = getattr(cls, name)
t = type(obj)
if t == types.MethodType and name.startswith("test"):
yield MethodTest(cls, name)
class ModuleScanner:
def inspect(self, obj):
return type(obj) == types.ModuleType
def descend(self, obj):
names = dir(obj)
names.sort()
for name in names:
yield getattr(obj, name)
def extract(self, obj):
# the None is required for older versions of python
return; yield None
class Harness:
def __init__(self):
self.scanners = [
ModuleScanner(),
ClassScanner("*Test", "*Tests", "*TestCase"),
FunctionScanner("test_*")
]
self.tests = []
self.scanned = []
def scan(self, *roots):
objects = list(roots)
while objects:
obj = objects.pop(0)
for s in self.scanners:
if s.inspect(obj):
self.tests.extend(s.extract(obj))
for child in s.descend(obj):
if not (child in self.scanned or child in objects):
objects.append(child)
self.scanned.append(obj)
modules = opts.modules
if not modules:
modules.extend(["unit_tests"])
h = Harness()
for name in modules:
m = __import__(name, None, None, ["dummy"])
h.scan(m)
filtered = [t for t in h.tests if is_included(t.name())]
ignored = [t for t in h.tests if is_ignored(t.name())]
total = len(filtered) + len(ignored)
if opts.xml and not list_only:
xmlr = JunitXmlStyleReporter(opts.xml);
xmlr.begin();
else:
xmlr = None
def runthrough():
passed = 0
failed = 0
skipped = 0
start = time.time()
for t in filtered:
if list_only:
print(t.name())
else:
st = t.run()
if xmlr:
xmlr.report(t.name(), st)
if st.passed:
passed += 1
elif st.skipped:
skipped += 1
elif st.failed:
failed += 1
if opts.hoe:
break
end = time.time()
run = passed + failed
if not list_only:
if passed:
_pass = "pass"
else:
_pass = "fail"
if failed:
outcome = "fail"
else:
outcome = "pass"
if ignored:
ign = "ignored"
else:
ign = "pass"
if skipped:
skip = "skip"
else:
skip = "pass"
print(colorize("Totals:", 1), end=None)
totals = [colorize_word("total", "%s tests" % total),
colorize_word(_pass, "%s passed" % passed),
colorize_word(skip, "%s skipped" % skipped),
colorize_word(ign, "%s ignored" % len(ignored)),
colorize_word(outcome, "%s failed" % failed)]
print(", ".join(totals), end=None)
if opts.hoe and failed > 0:
print(" -- (halted after %s)" % run)
else:
print("")
if opts.time and run > 0:
print(colorize("Timing:", 1), end=None)
timing = [colorize_word("elapsed", "%.2fs elapsed" % (end - start)),
colorize_word("average", "%.2fs average" % ((end - start)/run))]
print(", ".join(timing))
if xmlr:
xmlr.end()
return failed
limit = opts.count
count = 0
failures = False
while limit == 0 or count < limit:
count += 1
if runthrough():
failures = True
if count > 1:
print(" -- (failures after %s runthroughs)" % count)
else:
continue
if failures:
sys.exit(1)
else:
sys.exit(0)

21
tests/python/unit_tests/__init__.py Normal file
View File

@ -0,0 +1,21 @@
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
import unit_tests.container

51
tests/python/unit_tests/common.py Normal file
View File

@ -0,0 +1,51 @@
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
class Test(object):
def __init__(self, name):
self.name = name
def configure(self, config):
self.config = config
def default(self, name, value, **profiles):
default = value
profile = self.config.defines.get("profile")
if profile:
default = profiles.get(profile, default)
return self.config.defines.get(name, default)
@property
def delay(self):
return float(self.default("delay", "1", fast="0.1"))
@property
def timeout(self):
return float(self.default("timeout", "60", fast="10"))
@property
def verbose(self):
return int(self.default("verbose", 0))
class Skipped(Exception):
"""Throw to skip a test without generating a test failure."""
skipped = True

35
tests/python/unit_tests/container.py Normal file
View File

@ -0,0 +1,35 @@
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
import fusion
import common
class APITest(common.Test):
def setup(self):
pass
def teardown(self):
pass
def test_create_destroy(self):
container = fusion.Container("My-Container")
assert container.name == "My-Container"
container.destroy()

16
tox.ini Normal file
View File

@ -0,0 +1,16 @@
[tox]
# Proton does not support Python3 yet: see Jira PROTON-490
#envlist = py27,py33,pep8
envlist = py27,pep8
[testenv]
deps = -r{toxinidir}/test-requirements.txt
commands = {toxinidir}/tests/python/test-runner
[testenv:pep8]
commands = flake8
[flake8]
show-source = True
exclude = .tox,dist,doc,*.egg,build,__init__.py
builtins = _