QPIDIT-93: Added xUnit logging by using --xunit-log parameter; Refactoring of test classes, added qpid_interop_test.qit which runs all tests
diff --git a/src/python/qpid_interop_test/amqp_large_content_test.py b/src/python/qpid_interop_test/amqp_large_content_test.py
index 0b95b86..3e82f65 100755
--- a/src/python/qpid_interop_test/amqp_large_content_test.py
+++ b/src/python/qpid_interop_test/amqp_large_content_test.py
@@ -23,32 +23,23 @@
# under the License.
#
-import argparse
import sys
import unittest
from itertools import product
from json import dumps
-from os import getenv, path
-from proton import symbol
import qpid_interop_test.broker_properties
+import qpid_interop_test.qit_common
import qpid_interop_test.shims
-from qpid_interop_test.test_type_map import TestTypeMap
-# TODO: propose a sensible default when installation details are worked out
-QIT_INSTALL_PREFIX = getenv('QIT_INSTALL_PREFIX')
-if QIT_INSTALL_PREFIX is None:
- print 'ERROR: Environment variable QIT_INSTALL_PREFIX is not set'
- sys.exit(1)
-QIT_TEST_SHIM_HOME = path.join(QIT_INSTALL_PREFIX, 'libexec', 'qpid_interop_test', 'shims')
-class AmqpVariableSizeTypes(TestTypeMap):
+class AmqpVariableSizeTypes(qpid_interop_test.qit_common.QitTestTypeMap):
"""
Class which contains all the described AMQP variable-size types and the test values to be used in testing.
"""
- TYPE_MAP = {
+ type_map = {
# List of sizes in Mb (1024*1024 bytes)
# TODO: Until the issue of SLOW Proton performance for large messages is solved, the 100MB tests are
# disabled.
@@ -71,22 +62,22 @@
#'array': [[1, [1, 16, 256, 4096]], [10, [1, 16, 256, 4096]], [100, [1, 16, 256, 4096]]]
}
- # This section contains tests that should be skipped because of know issues that would cause the test to fail.
- # As the issues are resolved, these should be removed.
- BROKER_SKIP = {}
+ # This section contains tests that should be skipped because of known broker issues that would cause the
+ # test to fail. As the issues are resolved, these should be removed.
+ broker_skip = {}
+
+ client_skip = {}
-class AmqpLargeContentTestCase(unittest.TestCase):
- """
- Abstract base class for AMQP large content test cases
- """
+class AmqpLargeContentTestCase(qpid_interop_test.qit_common.QitTestCase):
+ """Abstract base class for AMQP large content tests"""
def run_test(self, sender_addr, receiver_addr, amqp_type, test_value_list, send_shim, receive_shim):
"""
Run this test by invoking the shim send method to send the test values, followed by the shim receive method
to receive the values. Finally, compare the sent values with the received values.
"""
- if len(test_value_list) > 0:
+ if test_value_list: # len > 0
# TODO: When Artemis can support it (in the next release), revert the queue name back to 'qpid-interop...'
# Currently, Artemis only supports auto-create queues for JMS, and the queue name must be prefixed by
# 'jms.queue.'
@@ -112,7 +103,7 @@
send_obj = sender.get_return_object()
if send_obj is not None:
if isinstance(send_obj, str):
- if len(send_obj) > 0:
+ if send_obj: # len > 0
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj))
else:
self.fail('Sender error: %s' % str(send_obj))
@@ -144,167 +135,87 @@
return tot_len
return None
-def create_testcase_class(amqp_type, shim_product):
- """
- Class factory function which creates new subclasses to AmqpTypeTestCase.
- """
-
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(amqp_type, BROKER),
- TYPES.skip_test_message(amqp_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- self.amqp_type,
- self.test_value_list,
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test_%s_%s->%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- class_name = amqp_type.title() + 'TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for AMQP 1.0 simple type \'%s\'' % amqp_type,
- 'amqp_type': amqp_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_value_list': TYPES.get_test_values(amqp_type)}
- new_class = type(class_name, (AmqpLargeContentTestCase,), class_dict)
- for send_shim, receive_shim in shim_product:
- add_test_method(new_class, send_shim, receive_shim)
- return new_class
+class TestOptions(qpid_interop_test.qit_common.QitCommonTestOptions):
+ """Command-line arguments used to control the test"""
-class TestOptions(object):
- """
- Class controlling command-line arguments used to control the test.
- """
- def __init__(self, shim_map):
- parser = argparse.ArgumentParser(description='Qpid-interop AMQP client interoparability test suite '
- 'for AMQP messages with large content')
- parser.add_argument('--sender', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node to which test suite will send messages.')
- parser.add_argument('--receiver', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node from which test suite will receive messages.')
- parser.add_argument('--no-skip', action='store_true',
- help='Do not skip tests that are excluded by default for reasons of a known bug')
- parser.add_argument('--broker-type', action='store', metavar='BROKER_NAME',
- help='Disable test of broker type (using connection properties) by specifying the broker' +
- ' name, or "None".')
- type_group = parser.add_mutually_exclusive_group()
+ def __init__(self, shim_map, default_xunit_dir=qpid_interop_test.qit_common.DEFUALT_XUNIT_LOG_DIR):
+ super(TestOptions, self).__init__('Qpid-interop AMQP client interoparability test suite for AMQP' +
+ ' messages with large content', shim_map, default_xunit_dir)
+ type_group = self._parser.add_mutually_exclusive_group()
type_group.add_argument('--include-type', action='append', metavar='AMQP-TYPE',
help='Name of AMQP type to include. Supported types:\n%s' %
- sorted(AmqpVariableSizeTypes.TYPE_MAP.keys()))
+ sorted(AmqpVariableSizeTypes.type_map.keys()))
type_group.add_argument('--exclude-type', action='append', metavar='AMQP-TYPE',
help='Name of AMQP type to exclude. Supported types: see "include-type" above')
- shim_group = parser.add_mutually_exclusive_group()
- shim_group.add_argument('--include-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to include. Supported shims:\n%s' % sorted(shim_map.keys()))
- shim_group.add_argument('--exclude-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to exclude. Supported shims: see "include-shim" above')
- self.args = parser.parse_args()
+
+
+class AmqpLargeContentTest(qpid_interop_test.qit_common.QitTest):
+ """Top-level test for AMQP large content (variable-size types)"""
+
+ TEST_NAME = 'amqp_large_content_test'
+
+ def __init__(self):
+ super(AmqpLargeContentTest, self).__init__(TestOptions, AmqpVariableSizeTypes)
+
+ def _generate_tests(self):
+ """Generate tests dynamically"""
+ self.test_suite = unittest.TestSuite()
+ # Create test classes dynamically
+ for amqp_type in sorted(self.types.get_type_list()):
+ if self.args.exclude_type is None or amqp_type not in self.args.exclude_type:
+ test_case_class = self.create_testcase_class(amqp_type, product(self.shim_map.values(), repeat=2))
+ self.test_suite.addTest(unittest.makeSuite(test_case_class))
+
+ def create_testcase_class(self, amqp_type, shim_product):
+ """
+ Class factory function which creates new subclasses to AmqpTypeTestCase.
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ @unittest.skipIf(self.types.skip_test(amqp_type, self.broker),
+ self.types.skip_test_message(amqp_type, self.broker))
+ @unittest.skipIf(self.types.skip_client_test(amqp_type, send_shim.NAME),
+ self.types.skip_client_test_message(amqp_type, send_shim.NAME, 'SENDER'))
+ @unittest.skipIf(self.types.skip_client_test(amqp_type, receive_shim.NAME),
+ self.types.skip_client_test_message(amqp_type, receive_shim.NAME, 'RECEIVER'))
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ self.amqp_type,
+ self.test_value_list,
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test_%s_%s->%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ class_name = amqp_type.title() + 'TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for AMQP 1.0 simple type \'%s\'' % amqp_type,
+ 'amqp_type': amqp_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_value_list': self.types.get_test_values(amqp_type)}
+ new_class = type(class_name, (AmqpLargeContentTestCase,), class_dict)
+ for send_shim, receive_shim in shim_product:
+ add_test_method(new_class, send_shim, receive_shim)
+ return new_class
#--- Main program start ---
if __name__ == '__main__':
-
- # SHIM_MAP contains an instance of each client language shim that is to be tested as a part of this test. For
- # every shim in this list, a test is dynamically constructed which tests it against itself as well as every
- # other shim in the list.
- #
- # As new shims are added, add them into this map to have them included in the test cases.
- PROTON_CPP_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'amqp_large_content_test', 'Receiver')
- PROTON_CPP_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'amqp_large_content_test', 'Sender')
- PROTON_PYTHON_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'amqp_large_content_test',
- 'Receiver.py')
- PROTON_PYTHON_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'amqp_large_content_test',
- 'Sender.py')
- AMQPNETLITE_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', 'amqp_large_content_test', 'Receiver.exe')
- AMQPNETLITE_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', 'amqp_large_content_test', 'Sender.exe')
-
- SHIM_MAP = {qpid_interop_test.shims.ProtonCppShim.NAME: \
- qpid_interop_test.shims.ProtonCppShim(PROTON_CPP_SENDER_SHIM, PROTON_CPP_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython2Shim.NAME: \
- qpid_interop_test.shims.ProtonPython2Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython3Shim.NAME: \
- qpid_interop_test.shims.ProtonPython3Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- }
- # Add shims that need detection during installation only if the necessary bits are present
- # AMQP DotNetLite client
- if path.isfile(AMQPNETLITE_RECEIVER_SHIM) and path.isfile(AMQPNETLITE_SENDER_SHIM):
- SHIM_MAP[qpid_interop_test.shims.AmqpNetLiteShim.NAME] = \
- qpid_interop_test.shims.AmqpNetLiteShim(AMQPNETLITE_SENDER_SHIM, AMQPNETLITE_RECEIVER_SHIM)
- else:
- print 'WARNING: AMQP DotNetLite shims not installed'
-
- ARGS = TestOptions(SHIM_MAP).args
- #print 'ARGS:', ARGS # debug
-
- # Add shims included from the command-line
- if ARGS.include_shim is not None:
- new_shim_map = {}
- for shim in ARGS.include_shim:
- try:
- new_shim_map[shim] = SHIM_MAP[shim]
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
- SHIM_MAP = new_shim_map
- # Remove shims excluded from the command-line
- elif ARGS.exclude_shim is not None:
- for shim in ARGS.exclude_shim:
- try:
- SHIM_MAP.pop(shim)
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
-
- # Connect to broker to find broker type, or use --broker-type param if present
- if ARGS.broker_type is not None:
- if ARGS.broker_type == 'None':
- BROKER = None
- else:
- BROKER = ARGS.broker_type
- else:
- CONNECTION_PROPS = qpid_interop_test.broker_properties.get_broker_properties(ARGS.sender)
- if CONNECTION_PROPS is None:
- print 'WARNING: Unable to get connection properties - unknown broker'
- BROKER = 'unknown'
- else:
- BROKER = CONNECTION_PROPS[symbol(u'product')] if symbol(u'product') in CONNECTION_PROPS \
- else '<product not found>'
- BROKER_VERSION = CONNECTION_PROPS[symbol(u'version')] if symbol(u'version') in CONNECTION_PROPS \
- else '<version not found>'
- BROKER_PLATFORM = CONNECTION_PROPS[symbol(u'platform')] if symbol(u'platform') in CONNECTION_PROPS \
- else '<platform not found>'
- print 'Test Broker: %s v.%s on %s' % (BROKER, BROKER_VERSION, BROKER_PLATFORM)
- print
- sys.stdout.flush()
- if ARGS.no_skip:
- BROKER = None # Will cause all tests to run
-
- TYPES = AmqpVariableSizeTypes().get_types(ARGS)
-
- # TEST_SUITE is the final suite of tests that will be run and which contains all the dynamically created
- # type classes, each of which contains a test for the combinations of client shims
- TEST_SUITE = unittest.TestSuite()
-
- # Create test classes dynamically
- for at in sorted(TYPES.get_type_list()):
- test_case_class = create_testcase_class(at, product(SHIM_MAP.values(), repeat=2))
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class))
-
- # Finally, run all the dynamically created tests
- RES = unittest.TextTestRunner(verbosity=2).run(TEST_SUITE)
- if not RES.wasSuccessful():
+ AMQP_LARGE_CONTENT_TEST = AmqpLargeContentTest()
+ AMQP_LARGE_CONTENT_TEST.run_test()
+ AMQP_LARGE_CONTENT_TEST.write_logs()
+ if not AMQP_LARGE_CONTENT_TEST.get_result():
sys.exit(1) # Errors or failures present
diff --git a/src/python/qpid_interop_test/amqp_types_test.py b/src/python/qpid_interop_test/amqp_types_test.py
index 17f3f1e..8f46f5e 100755
--- a/src/python/qpid_interop_test/amqp_types_test.py
+++ b/src/python/qpid_interop_test/amqp_types_test.py
@@ -23,35 +23,25 @@
# under the License.
#
-import argparse
import sys
import unittest
from itertools import product
from json import dumps
-from os import getenv, path
from time import mktime, time
from uuid import UUID, uuid4
-from proton import symbol
import qpid_interop_test.broker_properties
+import qpid_interop_test.qit_common
import qpid_interop_test.shims
-from qpid_interop_test.test_type_map import TestTypeMap
-
-# TODO: propose a sensible default when installation details are worked out
-QIT_INSTALL_PREFIX = getenv('QIT_INSTALL_PREFIX')
-if QIT_INSTALL_PREFIX is None:
- print 'ERROR: Environment variable QIT_INSTALL_PREFIX is not set'
- sys.exit(1)
-QIT_TEST_SHIM_HOME = path.join(QIT_INSTALL_PREFIX, 'libexec', 'qpid_interop_test', 'shims')
-class AmqpPrimitiveTypes(TestTypeMap):
+class AmqpPrimitiveTypes(qpid_interop_test.qit_common.QitTestTypeMap):
"""
Class which contains all the described AMQP primitive types and the test values to be used in testing.
"""
- TYPE_MAP = {
+ type_map = {
'null': ['None'],
'boolean': ['True',
'False'],
@@ -256,9 +246,9 @@
# ],
}
- # This section contains tests that should be skipped because of know issues that would cause the test to fail.
+ # This section contains tests that should be skipped because of known issues that would cause the test to fail.
# As the issues are resolved, these should be removed.
- BROKER_SKIP = {
+ broker_skip = {
'decimal32': {'ActiveMQ': 'decimal32 and decimal64 sent byte reversed: PROTON-1160',
'qpid-cpp': 'decimal32 not supported on qpid-cpp broker: QPIDIT-5, QPID-6328',
'apache-activemq-artemis': 'decimal32 and decimal64 sent byte reversed: PROTON-1160',
@@ -276,67 +266,62 @@
'double': {'apache-activemq-artemis': '-NaN is stripped of its sign: ENTMQ-1686',},
}
- CLIENT_SKIP = {
+ client_skip = {
'decimal32': {'AmqpNetLite': 'Decimal types not supported: https://github.com/Azure/amqpnetlite/issues/223', },
'decimal64': {'AmqpNetLite': 'Decimal types not supported: https://github.com/Azure/amqpnetlite/issues/223', },
'decimal128': {'AmqpNetLite': 'Decimal types not supported: https://github.com/Azure/amqpnetlite/issues/223', },
}
- def __init__(self):
- super(AmqpPrimitiveTypes, self).__init__()
-
- def create_array(self, amqp_type, repeat):
+ def create_array(self, array_amqp_type, repeat):
"""
Create a single test array for a given AMQP type from the test values for that type. It can be optionally
repeated for greater number of elements.
"""
- test_array = [amqp_type]
+ test_array = [array_amqp_type]
for _ in range(repeat):
- for val in self.TYPE_MAP[amqp_type]:
+ for val in self.type_map[array_amqp_type]:
test_array.append(val)
return test_array
def create_test_arrays(self):
""" Method to synthesize the test arrays from the values used in the previous type tests """
test_arrays = []
- for amqp_type in self.TYPE_MAP['array']:
- test_arrays.append(self.create_array(amqp_type, 1))
- print test_arrays
+ for array_amqp_type in self.type_map['array']:
+ test_arrays.append(self.create_array(array_amqp_type, 1))
+ print(test_arrays)
return test_arrays
- def get_test_values(self, amqp_type):
- """ Overload the parent method so that arrays can be synthesized rather than read directly """
- if amqp_type == 'array':
+ def get_test_values(self, test_type):
+ """
+ Overload the parent method so that arrays can be synthesized rather than read directly.
+ The test_type parameter is the AMQP type in this case
+ """
+ if test_type == 'array':
return self.create_test_arrays()
- return super(AmqpPrimitiveTypes, self).get_test_values(amqp_type)
+ return super(AmqpPrimitiveTypes, self).get_test_values(test_type)
-class AmqpTypeTestCase(unittest.TestCase):
- """
- Abstract base class for AMQP Type test cases
- """
+class AmqpTypeTestCase(qpid_interop_test.qit_common.QitTestCase):
+ """Abstract base class for AMQP Type test cases"""
def run_test(self, sender_addr, receiver_addr, amqp_type, test_value_list, send_shim, receive_shim):
"""
Run this test by invoking the shim send method to send the test values, followed by the shim receive method
to receive the values. Finally, compare the sent values with the received values.
"""
- if len(test_value_list) > 0:
+ if test_value_list: # len > 0
+ test_name = 'amqp_types_test.%s.%s.%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
# TODO: When Artemis can support it (in the next release), revert the queue name back to 'qpid-interop...'
# Currently, Artemis only supports auto-create queues for JMS, and the queue name must be prefixed by
# 'jms.queue.'
- #queue_name = 'qpid-interop.simple_type_tests.%s.%s.%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
- queue_name = 'jms.queue.qpid-interop.amqp_types_test.%s.%s.%s' % \
- (amqp_type, send_shim.NAME, receive_shim.NAME)
+ queue_name = 'jms.queue.qpid-interop.%s' % test_name
# Start the receive shim first (for queueless brokers/dispatch)
- receiver = receive_shim.create_receiver(receiver_addr, queue_name, amqp_type,
- str(len(test_value_list)))
+ receiver = receive_shim.create_receiver(receiver_addr, queue_name, amqp_type, str(len(test_value_list)))
receiver.start()
# Start the send shim
- sender = send_shim.create_sender(sender_addr, queue_name, amqp_type,
- dumps(test_value_list))
+ sender = send_shim.create_sender(sender_addr, queue_name, amqp_type, dumps(test_value_list))
sender.start()
# Wait for both shims to finish
@@ -347,7 +332,7 @@
send_obj = sender.get_return_object()
if send_obj is not None:
if isinstance(send_obj, str):
- if len(send_obj) > 0:
+ if send_obj: # len > 0
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj))
else:
self.fail('Sender error: %s' % str(send_obj))
@@ -367,177 +352,86 @@
else:
self.fail('Received non-tuple: %s' % str(receive_obj))
-def create_testcase_class(amqp_type, shim_product):
- """
- Class factory function which creates new subclasses to AmqpTypeTestCase.
- """
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
+class TestOptions(qpid_interop_test.qit_common.QitCommonTestOptions):
+ """Command-line arguments used to control the test"""
- def add_test_method(cls, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(amqp_type, BROKER),
- TYPES.skip_test_message(amqp_type, BROKER))
- @unittest.skipIf(TYPES.skip_client_test(amqp_type, send_shim.NAME),
- TYPES.skip_client_test_message(amqp_type, send_shim.NAME, "SENDER"))
- @unittest.skipIf(TYPES.skip_client_test(amqp_type, receive_shim.NAME),
- TYPES.skip_client_test_message(amqp_type, receive_shim.NAME, "RECEIVER"))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- self.amqp_type,
- self.test_value_list,
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test_%s_%s->%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- class_name = amqp_type.title() + 'TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for AMQP 1.0 simple type \'%s\'' % amqp_type,
- 'amqp_type': amqp_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_value_list': TYPES.get_test_values(amqp_type)}
- new_class = type(class_name, (AmqpTypeTestCase,), class_dict)
- for send_shim, receive_shim in shim_product:
- add_test_method(new_class, send_shim, receive_shim)
- return new_class
-
-
-class TestOptions(object):
- """
- Class controlling command-line arguments used to control the test.
- """
- def __init__(self, shim_map):
- parser = argparse.ArgumentParser(description='Qpid-interop AMQP client interoparability test suite '
- 'for AMQP simple types')
- parser.add_argument('--sender', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node to which test suite will send messages.')
- parser.add_argument('--receiver', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node from which test suite will receive messages.')
- parser.add_argument('--no-skip', action='store_true',
- help='Do not skip tests that are excluded by default for reasons of a known bug')
- parser.add_argument('--broker-type', action='store', metavar='BROKER_NAME',
- help='Disable test of broker type (using connection properties) by specifying the broker' +
- ' name, or "None".')
- type_group = parser.add_mutually_exclusive_group()
+ def __init__(self, shim_map, default_xunit_dir=qpid_interop_test.qit_common.DEFUALT_XUNIT_LOG_DIR):
+ super(TestOptions, self).__init__('Qpid-interop AMQP client interoparability test suite for AMQP simple types',
+ shim_map, default_xunit_dir)
+ type_group = self._parser.add_mutually_exclusive_group()
type_group.add_argument('--include-type', action='append', metavar='AMQP-TYPE',
help='Name of AMQP type to include. Supported types:\n%s' %
- sorted(AmqpPrimitiveTypes.TYPE_MAP.keys()))
+ sorted(AmqpPrimitiveTypes.type_map.keys()))
type_group.add_argument('--exclude-type', action='append', metavar='AMQP-TYPE',
help='Name of AMQP type to exclude. Supported types: see "include-type" above')
- shim_group = parser.add_mutually_exclusive_group()
- shim_group.add_argument('--include-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to include. Supported shims:\n%s' % sorted(shim_map.keys()))
- shim_group.add_argument('--exclude-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to exclude. Supported shims: see "include-shim" above')
- self.args = parser.parse_args()
+
+
+class AmqpTypesTest(qpid_interop_test.qit_common.QitTest):
+ """Top-level test for AMQP types"""
+
+ TEST_NAME = 'amqp_types_test'
+
+ def __init__(self):
+ super(AmqpTypesTest, self).__init__(TestOptions, AmqpPrimitiveTypes)
+
+ def _generate_tests(self):
+ """Generate tests dynamically"""
+ self.test_suite = unittest.TestSuite()
+ # Create test classes dynamically
+ for amqp_type in sorted(self.types.get_type_list()):
+ if self.args.exclude_type is None or amqp_type not in self.args.exclude_type:
+ test_case_class = self.create_testcase_class(amqp_type, product(self.shim_map.values(), repeat=2))
+ self.test_suite.addTest(unittest.makeSuite(test_case_class))
+
+ def create_testcase_class(self, amqp_type, shim_product):
+ """
+ Class factory function which creates new subclasses to AmqpTypeTestCase.
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ @unittest.skipIf(self.types.skip_test(amqp_type, self.broker),
+ self.types.skip_test_message(amqp_type, self.broker))
+ @unittest.skipIf(self.types.skip_client_test(amqp_type, send_shim.NAME),
+ self.types.skip_client_test_message(amqp_type, send_shim.NAME, 'SENDER'))
+ @unittest.skipIf(self.types.skip_client_test(amqp_type, receive_shim.NAME),
+ self.types.skip_client_test_message(amqp_type, receive_shim.NAME, 'RECEIVER'))
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ self.amqp_type,
+ self.test_value_list,
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test_%s_%s->%s' % (amqp_type, send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ class_name = amqp_type.title() + 'TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for AMQP 1.0 simple type \'%s\'' % amqp_type,
+ 'amqp_type': amqp_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_value_list': self.types.get_test_values(amqp_type)}
+ new_class = type(class_name, (AmqpTypeTestCase,), class_dict)
+ for send_shim, receive_shim in shim_product:
+ add_test_method(new_class, send_shim, receive_shim)
+ return new_class
#--- Main program start ---
if __name__ == '__main__':
-
- # SHIM_MAP contains an instance of each client language shim that is to be tested as a part of this test. For
- # every shim in this list, a test is dynamically constructed which tests it against itself as well as every
- # other shim in the list.
- #
- # As new shims are added, add them into this map to have them included in the test cases.
- PROTON_CPP_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'amqp_types_test', 'Receiver')
- PROTON_CPP_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'amqp_types_test', 'Sender')
- PROTON_PYTHON_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'amqp_types_test', 'Receiver.py')
- PROTON_PYTHON_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'amqp_types_test', 'Sender.py')
- PROTON_RHEAJS_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'rhea-js', 'amqp_types_test', 'Receiver.js')
- PROTON_RHEAJS_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'rhea-js', 'amqp_types_test', 'Sender.js')
- AMQPNETLITE_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', 'amqp_types_test', 'Receiver.exe')
- AMQPNETLITE_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', 'amqp_types_test', 'Sender.exe')
-
- SHIM_MAP = {qpid_interop_test.shims.ProtonCppShim.NAME: \
- qpid_interop_test.shims.ProtonCppShim(PROTON_CPP_SENDER_SHIM, PROTON_CPP_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython2Shim.NAME: \
- qpid_interop_test.shims.ProtonPython2Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython3Shim.NAME: \
- qpid_interop_test.shims.ProtonPython3Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- }
- # Add shims that need detection during installation only if the necessary bits are present
- # Rhea Javascript client
- if path.isfile(PROTON_RHEAJS_RECEIVER_SHIM) and path.isfile(PROTON_RHEAJS_SENDER_SHIM):
- SHIM_MAP[qpid_interop_test.shims.RheaJsShim.NAME] = \
- qpid_interop_test.shims.RheaJsShim(PROTON_RHEAJS_SENDER_SHIM, PROTON_RHEAJS_RECEIVER_SHIM)
- else:
- print 'WARNING: Rhea Javascript shims not installed'
- # AMQP DotNetLite client
- if path.isfile(AMQPNETLITE_RECEIVER_SHIM) and path.isfile(AMQPNETLITE_SENDER_SHIM):
- SHIM_MAP[qpid_interop_test.shims.AmqpNetLiteShim.NAME] = \
- qpid_interop_test.shims.AmqpNetLiteShim(AMQPNETLITE_SENDER_SHIM, AMQPNETLITE_RECEIVER_SHIM)
- else:
- print 'WARNING: AMQP DotNetLite shims not installed'
-
- ARGS = TestOptions(SHIM_MAP).args
- #print 'ARGS:', ARGS # debug
-
- # Add shims included from the command-line
- if ARGS.include_shim is not None:
- new_shim_map = {}
- for shim in ARGS.include_shim:
- try:
- new_shim_map[shim] = SHIM_MAP[shim]
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
- SHIM_MAP = new_shim_map
- # Remove shims excluded from the command-line
- elif ARGS.exclude_shim is not None:
- for shim in ARGS.exclude_shim:
- try:
- SHIM_MAP.pop(shim)
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
-
- # Connect to broker to find broker type, or use --broker-type param if present
- if ARGS.broker_type is not None:
- if ARGS.broker_type == 'None':
- BROKER = None
- else:
- BROKER = ARGS.broker_type
- else:
- CONNECTION_PROPS = qpid_interop_test.broker_properties.get_broker_properties(ARGS.sender)
- if CONNECTION_PROPS is None:
- print 'WARNING: Unable to get connection properties - unknown broker'
- BROKER = 'unknown'
- else:
- BROKER = CONNECTION_PROPS[symbol(u'product')] if symbol(u'product') in CONNECTION_PROPS \
- else '<product not found>'
- BROKER_VERSION = CONNECTION_PROPS[symbol(u'version')] if symbol(u'version') in CONNECTION_PROPS \
- else '<version not found>'
- BROKER_PLATFORM = CONNECTION_PROPS[symbol(u'platform')] if symbol(u'platform') in CONNECTION_PROPS \
- else '<platform not found>'
- print 'Test Broker: %s v.%s on %s' % (BROKER, BROKER_VERSION, BROKER_PLATFORM)
- print
- sys.stdout.flush()
- if ARGS.no_skip:
- BROKER = None # Will cause all tests to run
-
- TYPES = AmqpPrimitiveTypes().get_types(ARGS)
-
- # TEST_SUITE is the final suite of tests that will be run and which contains all the dynamically created
- # type classes, each of which contains a test for the combinations of client shims
- TEST_SUITE = unittest.TestSuite()
-
- # Create test classes dynamically
- for at in sorted(TYPES.get_type_list()):
- if ARGS.exclude_type is None or at not in ARGS.exclude_type:
- test_case_class = create_testcase_class(at, product(SHIM_MAP.values(), repeat=2))
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class))
-
- # Finally, run all the dynamically created tests
- RES = unittest.TextTestRunner(verbosity=2).run(TEST_SUITE)
- if not RES.wasSuccessful():
+ AMQP_TYPES_TEST = AmqpTypesTest()
+ AMQP_TYPES_TEST.run_test()
+ AMQP_TYPES_TEST.write_logs()
+ if not AMQP_TYPES_TEST.get_result():
sys.exit(1) # Errors or failures present
diff --git a/src/python/qpid_interop_test/interop_test_errors.py b/src/python/qpid_interop_test/interop_test_errors.py
index f300057..883d387 100644
--- a/src/python/qpid_interop_test/interop_test_errors.py
+++ b/src/python/qpid_interop_test/interop_test_errors.py
@@ -25,5 +25,4 @@
"""
Generic simple error class for use in interop tests
"""
- def __init__(self, error_message):
- super(InteropTestError, self).__init__(error_message)
+ pass
diff --git a/src/python/qpid_interop_test/jms_hdrs_props_test.py b/src/python/qpid_interop_test/jms_hdrs_props_test.py
index acefabf..9059e19 100755
--- a/src/python/qpid_interop_test/jms_hdrs_props_test.py
+++ b/src/python/qpid_interop_test/jms_hdrs_props_test.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python
"""
-Module to test JMS headers and properties
+Module to test JMS headers and properties on messages accross different clients
"""
#
@@ -23,34 +23,23 @@
# under the License.
#
-import argparse
import sys
import unittest
from itertools import combinations, product
from json import dumps
-from os import getenv, path
-from proton import symbol
import qpid_interop_test.broker_properties
+import qpid_interop_test.qit_common
import qpid_interop_test.shims
-from qpid_interop_test.test_type_map import TestTypeMap
-# TODO: propose a sensible default when installation details are worked out
-QIT_INSTALL_PREFIX = getenv('QIT_INSTALL_PREFIX')
-if QIT_INSTALL_PREFIX is None:
- print 'ERROR: Environment variable QIT_INSTALL_PREFIX is not set'
- sys.exit(1)
-QIT_TEST_SHIM_HOME = path.join(QIT_INSTALL_PREFIX, 'libexec', 'qpid_interop_test', 'shims')
-QPID_JMS_SHIM_VER = '0.1.0'
-
-class JmsMessageTypes(TestTypeMap):
+class JmsHdrPropTypes(qpid_interop_test.qit_common.QitTestTypeMap):
"""
Class which contains all the described JMS message types and the test values to be used in testing.
"""
- COMMON_SUBMAP = {
+ common_submap = {
'boolean': ['True',
'False'],
'byte': ['-0x80',
@@ -114,7 +103,7 @@
]
}
- TYPE_ADDITIONAL_SUBMAP = {
+ type_additional_submap = {
'bytes': [b'',
b'12345',
b'Hello, world',
@@ -127,13 +116,13 @@
b'\x7f'],
}
- # The TYPE_SUBMAP defines test values for JMS message types that allow typed message content. Note that the
+ # The type_submap defines test values for JMS message types that allow typed message content. Note that the
# types defined here are understood to be *Java* types and the stringified values are to be interpreted
# as the appropriate Java type by the send shim.
- TYPE_SUBMAP = TestTypeMap.merge_dicts(COMMON_SUBMAP, TYPE_ADDITIONAL_SUBMAP)
+ type_submap = qpid_interop_test.qit_common.QitTestTypeMap.merge_dicts(common_submap, type_additional_submap)
# JMS headers that can be set by the client prior to send / publish, and that should be preserved byt he broker
- HEADERS_MAP = {
+ headers_map = {
'JMS_CORRELATIONID_HEADER': {'string': ['Hello, world',
'"Hello, world"',
"Charlie's \"peach\"",
@@ -162,13 +151,13 @@
},
}
- PROPERTIES_MAP = COMMON_SUBMAP
+ properties_map = common_submap
- TYPE_MAP = {
+ type_map = {
'JMS_MESSAGE_TYPE': {'none': [None]},
- 'JMS_BYTESMESSAGE_TYPE': TYPE_SUBMAP,
- 'JMS_MAPMESSAGE_TYPE': TYPE_SUBMAP,
- 'JMS_STREAMMESSAGE_TYPE': TYPE_SUBMAP,
+ 'JMS_BYTESMESSAGE_TYPE': type_submap,
+ 'JMS_MAPMESSAGE_TYPE': type_submap,
+ 'JMS_STREAMMESSAGE_TYPE': type_submap,
'JMS_TEXTMESSAGE_TYPE': {'text': ['',
'Hello, world',
'"Hello, world"',
@@ -225,58 +214,60 @@
# },
}
- BROKER_SKIP = {}
+ # This section contains tests that should be skipped because of known broker issues that would cause the
+ # test to fail. As the issues are resolved, these should be removed.
+ broker_skip = {}
+
+ client_skip = {}
def get_types(self, args):
if 'include_hdr' in args and args.include_hdr is not None:
new_hdrs_map = {}
for hdr in args.include_hdr:
try:
- new_hdrs_map[hdr] = self.HEADERS_MAP[hdr]
+ new_hdrs_map[hdr] = self.headers_map[hdr]
except KeyError:
- print 'No such JMS header: "%s". Use --help for valid headers' % hdr
+ print('No such JMS header: "%s". Use --help for valid headers' % hdr)
sys.exit(1)
- self.HEADERS_MAP = new_hdrs_map
+ self.headers_map = new_hdrs_map
elif 'exclude_hdr' in args and args.exclude_hdr is not None:
if len(args.exclude_hdr) == 1 and args.exclude_hdr[0] == 'ALL':
- print 'NOTE: Command-line option has excluded all headers from test\n'
- self.HEADERS_MAP.clear()
+ print('NOTE: Command-line option has excluded all headers from test\n')
+ self.headers_map.clear()
else:
for hdr in args.exclude_hdr:
try:
- self.HEADERS_MAP.pop(hdr)
+ self.headers_map.pop(hdr)
except KeyError:
- print 'No such JMS header: "%s". Use --help for valid headers' % hdr
+ print('No such JMS header: "%s". Use --help for valid headers' % hdr)
sys.exit(1)
if 'include_prop' in args and args.include_prop is not None:
new_props_map = {}
for prop in args.include_prop:
try:
- new_props_map[prop] = self.PROPERTIES_MAP[prop]
+ new_props_map[prop] = self.properties_map[prop]
except KeyError:
- print 'No such JMS property: "%s". Use --help for valid property types' % prop
+ print('No such JMS property: "%s". Use --help for valid property types' % prop)
sys.exit(1)
- self.PROPERTIES_MAP = new_props_map
+ self.properties_map = new_props_map
elif 'exclude_prop' in args and args.exclude_prop is not None:
if len(args.exclude_prop) == 1 and args.exclude_prop[0] == 'ALL':
- print 'NOTE: Command-line option has excluded all properties from test\n'
- self.PROPERTIES_MAP.clear()
+ print('NOTE: Command-line option has excluded all properties from test\n')
+ self.properties_map.clear()
else:
for prop in args.exclude_prop:
try:
- self.PROPERTIES_MAP.pop(prop)
+ self.properties_map.pop(prop)
except KeyError:
- print 'No such JMS property: "%s". Use --help for valid property types' % prop
+ print('No such JMS property: "%s". Use --help for valid property types' % prop)
sys.exit(1)
return self
-class JmsMessageHdrsPropsTestCase(unittest.TestCase):
- """
- Abstract base class for JMS message headers and properties test cases
- """
+class JmsMessageHdrsPropsTestCase(qpid_interop_test.qit_common.QitTestCase):
+ """Abstract base class for JMS message headers and properties tests"""
def run_test(self, sender_addr, receiver_addr, queue_name_fragment, jms_message_type, test_values, msg_hdrs,
msg_props, send_shim, receive_shim):
@@ -288,7 +279,7 @@
# First create a map containing the numbers of expected mesasges for each JMS message type
num_test_values_map = {}
- if len(test_values) > 0:
+ if test_values: # len > 0
for index in test_values.keys():
num_test_values_map[index] = len(test_values[index])
# Create a map of flags which indicate to the receiver the details of some of the messages so that it can
@@ -320,7 +311,7 @@
send_obj = sender.get_return_object()
if send_obj is not None:
if isinstance(send_obj, str):
- if len(send_obj) > 0:
+ if send_obj: # len > 0
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj))
else:
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, str(send_obj)))
@@ -358,413 +349,305 @@
self.fail(str(receive_obj))
-def create_testcases():
- """Create all the test cases"""
- # --- Message headers on JMS Message ---
+class TestOptions(qpid_interop_test.qit_common.QitCommonTestOptions):
+ """Command-line arguments used to control the test"""
- # Part A: Single message header on each message
- test_case_class_a = create_part_a_testcase_class()
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class_a))
-
- # Part B: Combination of message headers, using first value in each value list
- test_case_class_b = create_part_b_testcase_class()
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class_b))
-
- # Part C: Single message property on each message
- test_case_class_c = create_part_c_testcase_class()
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class_c))
-
- # Part D: All headers and all properties on one of each type of JMS message
- test_case_class_d = create_part_d_testcase_class()
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class_d))
-
-
-def create_part_a_testcase_class():
- """
- Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
- a single JMS message type containing a single JMS header, one for each possible header
- """
-
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(jms_message_type, BROKER),
- TYPES.skip_test_message(jms_message_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- queue_name_fragment,
- self.jms_message_type,
- self.test_values,
- hdrs[1],
- props[1],
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test.A.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
- send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- jms_message_type = 'JMS_MESSAGE_TYPE'
- class_name = 'PartA_SingleJmsHeader_TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
- 'JMS header, one for each possible header',
- 'jms_message_type': jms_message_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_values': TYPES.get_test_values(jms_message_type)}
- new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
-
- for send_shim, receive_shim in product(SHIM_MAP.values(), repeat=2):
- for msg_header in TYPES.HEADERS_MAP.iterkeys():
- for header_type, header_val_list in TYPES.HEADERS_MAP[msg_header].iteritems():
- header_val_cnt = 0
- for header_val in header_val_list:
- header_val_cnt += 1
- method_subname = '%s.%s-%02d' % (msg_header, header_type, header_val_cnt)
- add_test_method(new_class,
- method_subname,
- (method_subname, {msg_header: {header_type: header_val}}),
- ('', {}),
- send_shim,
- receive_shim)
- return new_class
-
-
-def create_part_b_testcase_class():
- """
- Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
- a single JMS message type containing a combination of JMS headers
- """
-
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(jms_message_type, BROKER),
- TYPES.skip_test_message(jms_message_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- queue_name_fragment,
- self.jms_message_type,
- self.test_values,
- hdrs[1],
- props[1],
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test.B.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
- send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- jms_message_type = 'JMS_MESSAGE_TYPE'
- class_name = 'PartB_JmsHeaderCombination_TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for JMS message type \'%s\' containing a combination ' % jms_message_type +
- 'of possible JMS headers',
- 'jms_message_type': jms_message_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_values': TYPES.get_test_values(jms_message_type)}
- new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
-
- for send_shim, receive_shim in product(SHIM_MAP.values(), repeat=2):
- for jms_hdrs_combo_index in range(0, len(TYPES.HEADERS_MAP.keys())+1):
- for jms_hdrs_combo in combinations(TYPES.HEADERS_MAP.iterkeys(), jms_hdrs_combo_index):
- jms_hdr_list = []
- for jms_header in jms_hdrs_combo:
- data_type_list = []
- for data_type in TYPES.HEADERS_MAP[jms_header].keys():
- data_type_list.append((jms_header, data_type))
- jms_hdr_list.append(data_type_list)
- for combo in product(*jms_hdr_list):
- if len(combo) > 1: # ignore empty and single combos (already tested in Part A)
- method_subname = ''
- header_map = {}
- for combo_item in combo:
- if len(method_subname) > 0:
- method_subname += '+'
- method_subname += '%s:%s' % combo_item
- header_type_map = TYPES.HEADERS_MAP[combo_item[0]]
- header_val_list = header_type_map[combo_item[1]]
- header_map[combo_item[0]] = {combo_item[1]: header_val_list[0]}
- add_test_method(new_class,
- method_subname,
- (method_subname, header_map),
- ('', {}),
- send_shim,
- receive_shim)
- return new_class
-
-
-def create_part_c_testcase_class():
- """
- Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
- a single JMS message type containing a single JMS property
- """
-
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(jms_message_type, BROKER),
- TYPES.skip_test_message(jms_message_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- queue_name_fragment,
- self.jms_message_type,
- self.test_values,
- hdrs[1],
- props[1],
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test.C.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
- send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- jms_message_type = 'JMS_MESSAGE_TYPE'
- class_name = 'PartC_SingleJmsProperty_TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
- 'JMS property',
- 'jms_message_type': jms_message_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_values': TYPES.get_test_values(jms_message_type)}
- new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
-
- for send_shim, receive_shim in product(SHIM_MAP.values(), repeat=2):
- for prop_type, prop_val_list in TYPES.PROPERTIES_MAP.iteritems():
- prop_val_cnt = 0
- for prop_val in prop_val_list:
- prop_val_cnt += 1
- prop_name = 'prop_%s_%02d' % (prop_type, prop_val_cnt)
- add_test_method(new_class,
- prop_name,
- ('', {}),
- (prop_name, {prop_name: {prop_type: prop_val}}),
- send_shim,
- receive_shim)
- return new_class
-
-
-def create_part_d_testcase_class():
- """
- Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
- all message headers and properties on each type of JMS message
- """
-
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(jms_message_type, BROKER),
- TYPES.skip_test_message(jms_message_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- queue_name_fragment,
- self.jms_message_type,
- self.test_values,
- hdrs[1],
- props[1],
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test.D.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
- send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- jms_message_type = 'JMS_MESSAGE_TYPE'
- class_name = 'PartD_AllJmsHeaders_AllJmsProperties_TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
- 'JMS property',
- 'jms_message_type': jms_message_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_values': TYPES.get_test_values(jms_message_type)}
- new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
-
- all_hdrs = {}
- for msg_header in TYPES.HEADERS_MAP.iterkeys():
- header_type_dict = TYPES.HEADERS_MAP[msg_header]
- header_type, header_val_list = header_type_dict.iteritems().next()
- header_val = header_val_list[0]
- all_hdrs[msg_header] = {header_type: header_val}
-
- all_props = {}
- for prop_type, prop_val_list in TYPES.PROPERTIES_MAP.iteritems():
- prop_val_cnt = 0
- for prop_val in prop_val_list:
- prop_val_cnt += 1
- all_props['prop_%s_%02d' % (prop_type, prop_val_cnt)] = {prop_type: prop_val}
-
- for send_shim, receive_shim in product(SHIM_MAP.values(), repeat=2):
- add_test_method(new_class,
- 'HDRS+PROPS',
- ('hdrs', all_hdrs),
- ('props', all_props),
- send_shim,
- receive_shim)
- return new_class
-
-
-class TestOptions(object):
- """
- Class controlling command-line arguments used to control the test.
- """
- def __init__(self, shim_map):
- parser = argparse.ArgumentParser(description='Qpid-interop AMQP client interoparability test suite '
- 'for JMS headers and properties')
- parser.add_argument('--sender', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node to which test suite will send messages.')
- parser.add_argument('--receiver', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node from which test suite will receive messages.')
- parser.add_argument('--no-skip', action='store_true',
- help='Do not skip tests that are excluded by default for reasons of a known bug')
- parser.add_argument('--broker-type', action='store', metavar='BROKER_NAME',
- help='Disable test of broker type (using connection properties) by specifying the broker' +
- ' name, or "None".')
+ def __init__(self, shim_map, default_xunit_dir=qpid_interop_test.qit_common.DEFUALT_XUNIT_LOG_DIR):
+ super(TestOptions, self).__init__('Qpid-interop AMQP client interoparability test suite for JMS headers '
+ 'and properties', shim_map, default_xunit_dir)
# Control over JMS message headers
- hdrs_group = parser.add_mutually_exclusive_group()
+ hdrs_group = self._parser.add_mutually_exclusive_group()
hdrs_group.add_argument('--include-hdr', action='append', metavar='HDR-NAME',
help='Name of JMS header to include. Supported headers:\n%s' %
- sorted(JmsMessageTypes.HEADERS_MAP.keys()))
+ sorted(JmsHdrPropTypes.headers_map.keys()))
hdrs_group.add_argument('--exclude-hdr', action='append', metavar='HDR-NAME',
help='Name of JMS header to exclude. Supported types: see "include-hdr" above' +
' or "ALL" to exclude all header tests')
# Control over JMS message properties
- props_group = parser.add_mutually_exclusive_group()
+ props_group = self._parser.add_mutually_exclusive_group()
props_group.add_argument('--include-prop', action='append', metavar='PROP-TYPE',
help='Name of JMS property type to include. Supported property types:\n%s' %
- sorted(JmsMessageTypes.PROPERTIES_MAP.keys()))
+ sorted(JmsHdrPropTypes.properties_map.keys()))
props_group.add_argument('--exclude-prop', action='append', metavar='PROP-TYPE',
help='Name of JMS property type to exclude. Supported types: see "include-prop"' +
- ' above or "ALL" to exclude all properties tests')
+ ' above or "ALL" to exclude all properties tests')
- shim_group = parser.add_mutually_exclusive_group()
- shim_group.add_argument('--include-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to include. Supported shims:\n%s' % sorted(shim_map.keys()))
- shim_group.add_argument('--exclude-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to exclude. Supported shims: see "include-shim" above')
- self.args = parser.parse_args()
+class JmsHdrsPropsTest(qpid_interop_test.qit_common.QitJmsTest):
+ """Top-level test for JMS message types"""
+
+ TEST_NAME = 'jms_hdrs_props_test'
+
+ def __init__(self):
+ super(JmsHdrsPropsTest, self).__init__(TestOptions, JmsHdrPropTypes)
+
+ def _generate_tests(self):
+ """Generate tests dynamically"""
+ self.test_suite = unittest.TestSuite()
+ # Part A: Single message header on each message
+ test_case_class_a = self._generate_part_a()
+ self.test_suite.addTest(unittest.makeSuite(test_case_class_a))
+ # Part B: Combination of message headers, using first value in each value list
+ test_case_class_b = self._generate_part_b()
+ self.test_suite.addTest(unittest.makeSuite(test_case_class_b))
+ # Part C: Single message property on each message
+ test_case_class_c = self._generate_part_c()
+ self.test_suite.addTest(unittest.makeSuite(test_case_class_c))
+ # Part D: All headers and all properties on one of each type of JMS message
+ test_case_class_d = self._generate_part_d()
+ self.test_suite.addTest(unittest.makeSuite(test_case_class_d))
+
+ def _generate_part_a(self):
+ """
+ Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
+ a single JMS message type containing a single JMS header, one for each possible header
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ queue_name_fragment,
+ self.jms_message_type,
+ self.test_values,
+ hdrs[1],
+ props[1],
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test.A.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
+ send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ jms_message_type = 'JMS_MESSAGE_TYPE'
+ class_name = 'PartA_SingleJmsHeader_TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
+ 'JMS header, one for each possible header',
+ 'jms_message_type': jms_message_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_values': self.types.get_test_values(jms_message_type)}
+ new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
+
+ for send_shim, receive_shim in product(self.shim_map.values(), repeat=2):
+ for msg_header in self.types.headers_map.iterkeys():
+ for header_type, header_val_list in self.types.headers_map[msg_header].iteritems():
+ header_val_cnt = 0
+ for header_val in header_val_list:
+ header_val_cnt += 1
+ method_subname = '%s.%s-%02d' % (msg_header, header_type, header_val_cnt)
+ add_test_method(new_class,
+ method_subname,
+ (method_subname, {msg_header: {header_type: header_val}}),
+ ('', {}),
+ send_shim,
+ receive_shim)
+ return new_class
+
+
+ def _generate_part_b(self):
+ """
+ Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
+ a single JMS message type containing a combination of JMS headers
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ queue_name_fragment,
+ self.jms_message_type,
+ self.test_values,
+ hdrs[1],
+ props[1],
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test.B.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
+ send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ jms_message_type = 'JMS_MESSAGE_TYPE'
+ class_name = 'PartB_JmsHeaderCombination_TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for JMS message type \'%s\' containing a combination ' % jms_message_type +
+ 'of possible JMS headers',
+ 'jms_message_type': jms_message_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_values': self.types.get_test_values(jms_message_type)}
+ new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
+
+ for send_shim, receive_shim in product(self.shim_map.values(), repeat=2):
+ for jms_hdrs_combo_index in range(0, len(self.types.headers_map.keys())+1):
+ for jms_hdrs_combo in combinations(self.types.headers_map.iterkeys(), jms_hdrs_combo_index):
+ jms_hdr_list = []
+ for jms_header in jms_hdrs_combo:
+ data_type_list = []
+ for data_type in self.types.headers_map[jms_header].keys():
+ data_type_list.append((jms_header, data_type))
+ jms_hdr_list.append(data_type_list)
+ for combo in product(*jms_hdr_list):
+ if len(combo) > 1: # ignore empty and single combos (already tested in Part A)
+ method_subname = ''
+ header_map = {}
+ for combo_item in combo:
+ if method_subname: # len > 0
+ method_subname += '+'
+ method_subname += '%s:%s' % combo_item
+ header_type_map = self.types.headers_map[combo_item[0]]
+ header_val_list = header_type_map[combo_item[1]]
+ header_map[combo_item[0]] = {combo_item[1]: header_val_list[0]}
+ add_test_method(new_class,
+ method_subname,
+ (method_subname, header_map),
+ ('', {}),
+ send_shim,
+ receive_shim)
+ return new_class
+
+
+ def _generate_part_c(self):
+ """
+ Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
+ a single JMS message type containing a single JMS property
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ queue_name_fragment,
+ self.jms_message_type,
+ self.test_values,
+ hdrs[1],
+ props[1],
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test.C.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
+ send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ jms_message_type = 'JMS_MESSAGE_TYPE'
+ class_name = 'PartC_SingleJmsProperty_TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
+ 'JMS property',
+ 'jms_message_type': jms_message_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_values': self.types.get_test_values(jms_message_type)}
+ new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
+
+ for send_shim, receive_shim in product(self.shim_map.values(), repeat=2):
+ for prop_type, prop_val_list in self.types.properties_map.iteritems():
+ prop_val_cnt = 0
+ for prop_val in prop_val_list:
+ prop_val_cnt += 1
+ prop_name = 'prop_%s_%02d' % (prop_type, prop_val_cnt)
+ add_test_method(new_class,
+ prop_name,
+ ('', {}),
+ (prop_name, {prop_name: {prop_type: prop_val}}),
+ send_shim,
+ receive_shim)
+ return new_class
+
+
+ def _generate_part_d(self):
+ """
+ Class factory function which creates new subclasses to JmsMessageTypeTestCase. Creates a test case class for
+ all message headers and properties on each type of JMS message
+ """
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, queue_name_fragment, hdrs, props, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ queue_name_fragment,
+ self.jms_message_type,
+ self.test_values,
+ hdrs[1],
+ props[1],
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test.D.%s.%s%s.%s->%s' % (jms_message_type[4:-5], hdrs[0], props[0],
+ send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ jms_message_type = 'JMS_MESSAGE_TYPE'
+ class_name = 'PartD_AllJmsHeaders_AllJmsProperties_TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for JMS message type \'%s\' containing a single ' % jms_message_type +
+ 'JMS property',
+ 'jms_message_type': jms_message_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_values': self.types.get_test_values(jms_message_type)}
+ new_class = type(class_name, (JmsMessageHdrsPropsTestCase,), class_dict)
+
+ all_hdrs = {}
+ for msg_header in self.types.headers_map.iterkeys():
+ header_type_dict = self.types.headers_map[msg_header]
+ header_type, header_val_list = header_type_dict.iteritems().next()
+ header_val = header_val_list[0]
+ all_hdrs[msg_header] = {header_type: header_val}
+
+ all_props = {}
+ for prop_type, prop_val_list in self.types.properties_map.iteritems():
+ prop_val_cnt = 0
+ for prop_val in prop_val_list:
+ prop_val_cnt += 1
+ all_props['prop_%s_%02d' % (prop_type, prop_val_cnt)] = {prop_type: prop_val}
+
+ for send_shim, receive_shim in product(self.shim_map.values(), repeat=2):
+ add_test_method(new_class,
+ 'HDRS+PROPS',
+ ('hdrs', all_hdrs),
+ ('props', all_props),
+ send_shim,
+ receive_shim)
+ return new_class
#--- Main program start ---
if __name__ == '__main__':
-
- PROTON_CPP_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'jms_hdrs_props_test', 'Receiver')
- PROTON_CPP_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'jms_hdrs_props_test', 'Sender')
- PROTON_PYTHON_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'jms_hdrs_props_test',
- 'Receiver.py')
- PROTON_PYTHON_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'jms_hdrs_props_test', 'Sender.py')
- QIT_JMS_CLASSPATH_FILE = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms', 'cp.txt')
- if path.isfile(QIT_JMS_CLASSPATH_FILE):
- with open(QIT_JMS_CLASSPATH_FILE, 'r') as classpath_file:
- QIT_JMS_CLASSPATH = classpath_file.read()
- else:
- QIT_JMS_CLASSPATH = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms',
- 'qpid-interop-test-jms-shim-%s-jar-with-dependencies.jar' % QPID_JMS_SHIM_VER)
- QPID_JMS_RECEIVER_SHIM = 'org.apache.qpid.interop_test.jms_hdrs_props_test.Receiver'
- QPID_JMS_SENDER_SHIM = 'org.apache.qpid.interop_test.jms_hdrs_props_test.Sender'
-
- # SHIM_MAP contains an instance of each client language shim that is to be tested as a part of this test. For
- # every shim in this list, a test is dynamically constructed which tests it against itself as well as every
- # other shim in the list.
- #
- # As new shims are added, add them into this map to have them included in the test cases.
- SHIM_MAP = {qpid_interop_test.shims.ProtonCppShim.NAME: \
- qpid_interop_test.shims.ProtonCppShim(PROTON_CPP_SENDER_SHIM, PROTON_CPP_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython2Shim.NAME: \
- qpid_interop_test.shims.ProtonPython2Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython3Shim.NAME: \
- qpid_interop_test.shims.ProtonPython3Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.QpidJmsShim.NAME: \
- qpid_interop_test.shims.QpidJmsShim(QIT_JMS_CLASSPATH, QPID_JMS_SENDER_SHIM, QPID_JMS_RECEIVER_SHIM),
- }
-
- ARGS = TestOptions(SHIM_MAP).args
- #print 'ARGS:', ARGS # debug
-
- # Add shims included from the command-line
- if ARGS.include_shim is not None:
- new_shim_map = {}
- for shim in ARGS.include_shim:
- try:
- new_shim_map[shim] = SHIM_MAP[shim]
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
- SHIM_MAP = new_shim_map
- # Remove shims excluded from the command-line
- elif ARGS.exclude_shim is not None:
- for shim in ARGS.exclude_shim:
- try:
- SHIM_MAP.pop(shim)
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
-
- # Connect to broker to find broker type, or use --broker-type param if present
- if ARGS.broker_type is not None:
- if ARGS.broker_type == 'None':
- BROKER = None
- else:
- BROKER = ARGS.broker_type
- else:
- CONNECTION_PROPS = qpid_interop_test.broker_properties.get_broker_properties(ARGS.sender)
- if CONNECTION_PROPS is None:
- print 'WARNING: Unable to get connection properties - unknown broker'
- BROKER = 'unknown'
- else:
- BROKER = CONNECTION_PROPS[symbol(u'product')] if symbol(u'product') in CONNECTION_PROPS \
- else '<product not found>'
- BROKER_VERSION = CONNECTION_PROPS[symbol(u'version')] if symbol(u'version') in CONNECTION_PROPS \
- else '<version not found>'
- BROKER_PLATFORM = CONNECTION_PROPS[symbol(u'platform')] if symbol(u'platform') in CONNECTION_PROPS \
- else '<platform not found>'
- print 'Test Broker: %s v.%s on %s' % (BROKER, BROKER_VERSION, BROKER_PLATFORM)
- print
- sys.stdout.flush()
- if ARGS.no_skip:
- BROKER = None # Will cause all tests to run
-
- TYPES = JmsMessageTypes().get_types(ARGS)
-
- # TEST_SUITE is the final suite of tests that will be run and which contains all the dynamically created
- # type classes, each of which contains a test for the combinations of client shims
- TEST_SUITE = unittest.TestSuite()
-
- # Create test classes dynamically
- create_testcases()
-
- # Finally, run all the dynamically created tests
- RES = unittest.TextTestRunner(verbosity=2).run(TEST_SUITE)
- if not RES.wasSuccessful():
- sys.exit(1)
+ JMS_MESSAGES_TEST = JmsHdrsPropsTest()
+ JMS_MESSAGES_TEST.run_test()
+ JMS_MESSAGES_TEST.write_logs()
+ if not JMS_MESSAGES_TEST.get_result():
+ sys.exit(1) # Errors or failures present
diff --git a/src/python/qpid_interop_test/jms_messages_test.py b/src/python/qpid_interop_test/jms_messages_test.py
index 4798cf0..efd3885 100755
--- a/src/python/qpid_interop_test/jms_messages_test.py
+++ b/src/python/qpid_interop_test/jms_messages_test.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python
"""
-Module to test JMS message types across different APIs
+Module to test JMS message types across different clients
"""
#
@@ -23,35 +23,23 @@
# under the License.
#
-import argparse
import sys
import unittest
from itertools import product
from json import dumps
-from os import getenv, path
-from proton import symbol
import qpid_interop_test.broker_properties
+import qpid_interop_test.qit_common
import qpid_interop_test.shims
-from qpid_interop_test.test_type_map import TestTypeMap
-# TODO: propose a sensible default when installation details are worked out
-QIT_INSTALL_PREFIX = getenv('QIT_INSTALL_PREFIX')
-if QIT_INSTALL_PREFIX is None:
- print 'ERROR: Environment variable QIT_INSTALL_PREFIX is not set'
- sys.exit(1)
-QIT_TEST_SHIM_HOME = path.join(QIT_INSTALL_PREFIX, 'libexec', 'qpid_interop_test', 'shims')
-QPID_JMS_SHIM_VER = '0.1.0'
-
-
-class JmsMessageTypes(TestTypeMap):
+class JmsMessageTypes(qpid_interop_test.qit_common.QitTestTypeMap):
"""
Class which contains all the described JMS message types and the test values to be used in testing.
"""
- COMMON_SUBMAP = {
+ common_submap = {
'boolean': ['True',
'False'],
'byte': ['-0x80',
@@ -115,7 +103,7 @@
]
}
- TYPE_ADDITIONAL_SUBMAP = {
+ type_additional_submap = {
'bytes': [b'',
b'12345',
b'Hello, world',
@@ -128,16 +116,16 @@
b'\x7f'],
}
- # The TYPE_SUBMAP defines test values for JMS message types that allow typed message content. Note that the
+ # The type_submap defines test values for JMS message types that allow typed message content. Note that the
# types defined here are understood to be *Java* types and the stringified values are to be interpreted
# as the appropriate Java type by the send shim.
- TYPE_SUBMAP = TestTypeMap.merge_dicts(COMMON_SUBMAP, TYPE_ADDITIONAL_SUBMAP)
+ type_submap = qpid_interop_test.qit_common.QitTestTypeMap.merge_dicts(common_submap, type_additional_submap)
- TYPE_MAP = {
+ type_map = {
'JMS_MESSAGE_TYPE': {'none': [None]},
- 'JMS_BYTESMESSAGE_TYPE': TYPE_SUBMAP,
- 'JMS_MAPMESSAGE_TYPE': TYPE_SUBMAP,
- 'JMS_STREAMMESSAGE_TYPE': TYPE_SUBMAP,
+ 'JMS_BYTESMESSAGE_TYPE': type_submap,
+ 'JMS_MAPMESSAGE_TYPE': type_submap,
+ 'JMS_STREAMMESSAGE_TYPE': type_submap,
'JMS_TEXTMESSAGE_TYPE': {'text': ['',
'Hello, world',
'"Hello, world"',
@@ -194,13 +182,15 @@
# },
}
- BROKER_SKIP = {}
+ # This section contains tests that should be skipped because of known broker issues that would cause the
+ # test to fail. As the issues are resolved, these should be removed.
+ broker_skip = {}
+
+ client_skip = {}
-class JmsMessageTypeTestCase(unittest.TestCase):
- """
- Abstract base class for JMS message type test cases
- """
+class JmsMessageTypeTestCase(qpid_interop_test.qit_common.QitTestCase):
+ """Abstract base class for JMS message type tests"""
def run_test(self, sender_addr, receiver_addr, jms_message_type, test_values, send_shim, receive_shim):
"""
@@ -212,7 +202,7 @@
# First create a map containing the numbers of expected mesasges for each JMS message type
num_test_values_map = {}
- if len(test_values) > 0:
+ if test_values: # len > 0
for index in test_values.keys():
num_test_values_map[index] = len(test_values[index])
# Start the receiver shim
@@ -233,7 +223,7 @@
send_obj = sender.get_return_object()
if send_obj is not None:
if isinstance(send_obj, str):
- if len(send_obj) > 0:
+ if send_obj: # len > 0
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj))
else:
self.fail('Send shim \'%s\':\n%s' % (send_shim.NAME, str(send_obj)))
@@ -258,174 +248,89 @@
self.fail('Received non-tuple: %s' % str(receive_obj))
-def create_testcase_class(jms_message_type, shim_product):
- """
- Class factory function which creates new subclasses to JmsMessageTypeTestCase. Each call creates a single new
- test case named and based on the parameters supplied to the method
- """
+class TestOptions(qpid_interop_test.qit_common.QitCommonTestOptions):
+ """Command-line arguments used to control the test"""
- def __repr__(self):
- """Print the class name"""
- return self.__class__.__name__
-
- def add_test_method(cls, send_shim, receive_shim):
- """Function which creates a new test method in class cls"""
-
- @unittest.skipIf(TYPES.skip_test(jms_message_type, BROKER),
- TYPES.skip_test_message(jms_message_type, BROKER))
- def inner_test_method(self):
- self.run_test(self.sender_addr,
- self.receiver_addr,
- self.jms_message_type,
- self.test_values,
- send_shim,
- receive_shim)
-
- inner_test_method.__name__ = 'test_%s_%s->%s' % (jms_message_type[4:-5], send_shim.NAME, receive_shim.NAME)
- setattr(cls, inner_test_method.__name__, inner_test_method)
-
- class_name = jms_message_type[4:-5].title() + 'TestCase'
- class_dict = {'__name__': class_name,
- '__repr__': __repr__,
- '__doc__': 'Test case for JMS message type \'%s\'' % jms_message_type,
- 'jms_message_type': jms_message_type,
- 'sender_addr': ARGS.sender,
- 'receiver_addr': ARGS.receiver,
- 'test_values': TYPES.get_test_values(jms_message_type)} # tuple (tot_size, {...}
- new_class = type(class_name, (JmsMessageTypeTestCase,), class_dict)
- for send_shim, receive_shim in shim_product:
- add_test_method(new_class, send_shim, receive_shim)
-
- return new_class
-
-
-class TestOptions(object):
- """
- Class controlling command-line arguments used to control the test.
- """
- def __init__(self, shim_map):
- parser = argparse.ArgumentParser(description='Qpid-interop AMQP client interoparability test suite '
- 'for JMS message types')
- parser.add_argument('--sender', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node to which test suite will send messages.')
- parser.add_argument('--receiver', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
- help='Node from which test suite will receive messages.')
- parser.add_argument('--no-skip', action='store_true',
- help='Do not skip tests that are excluded by default for reasons of a known bug')
- parser.add_argument('--broker-type', action='store', metavar='BROKER_NAME',
- help='Disable test of broker type (using connection properties) by specifying the broker' +
- ' name, or "None".')
- type_group = parser.add_mutually_exclusive_group()
+ def __init__(self, shim_map, default_xunit_dir=qpid_interop_test.qit_common.DEFUALT_XUNIT_LOG_DIR):
+ super(TestOptions, self).__init__('Qpid-interop AMQP client interoparability test suite '
+ 'for JMS message types', shim_map, default_xunit_dir)
+ type_group = self._parser.add_mutually_exclusive_group()
type_group.add_argument('--include-type', action='append', metavar='JMS_MESSAGE-TYPE',
help='Name of JMS message type to include. Supported types:\n%s' %
- sorted(JmsMessageTypes.TYPE_MAP.keys()))
+ sorted(JmsMessageTypes.type_map.keys()))
type_group.add_argument('--exclude-type', action='append', metavar='JMS_MESSAGE-TYPE',
help='Name of JMS message type to exclude. Supported types: see "include-type" above')
- shim_group = parser.add_mutually_exclusive_group()
- shim_group.add_argument('--include-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to include. Supported shims:\n%s' % sorted(shim_map.keys()))
- shim_group.add_argument('--exclude-shim', action='append', metavar='SHIM-NAME',
- help='Name of shim to exclude. Supported shims: see "include-shim" above')
- self.args = parser.parse_args()
+
+
+class JmsMessagesTest(qpid_interop_test.qit_common.QitJmsTest):
+ """Top-level test for JMS message types"""
+
+ TEST_NAME = 'jms_messages_test'
+
+ def __init__(self):
+ super(JmsMessagesTest, self).__init__(TestOptions, JmsMessageTypes)
+
+ def _generate_tests(self):
+ """Generate tests dynamically"""
+ self.test_suite = unittest.TestSuite()
+ # Create test classes dynamically
+ for jmt in sorted(self.types.get_type_list()):
+ if self.args.exclude_type is None or jmt not in self.args.exclude_type:
+ test_case_class = self.create_testcase_class(jmt, product(self.shim_map.values(), repeat=2))
+ self.test_suite.addTest(unittest.makeSuite(test_case_class))
+
+
+ def create_testcase_class(self, jms_message_type, shim_product):
+ """
+ Class factory function which creates new subclasses to JmsMessageTypeTestCase. Each call creates a single new
+ test case named and based on the parameters supplied to the method
+ """
+
+
+ def __repr__(self):
+ """Print the class name"""
+ return self.__class__.__name__
+
+ def add_test_method(cls, send_shim, receive_shim):
+ """Function which creates a new test method in class cls"""
+
+ @unittest.skipIf(self.types.skip_test(jms_message_type, self.broker),
+ self.types.skip_test_message(jms_message_type, self.broker))
+ @unittest.skipIf(self.types.skip_client_test(jms_message_type, send_shim.NAME),
+ self.types.skip_client_test_message(jms_message_type, send_shim.NAME, 'SENDER'))
+ @unittest.skipIf(self.types.skip_client_test(jms_message_type, receive_shim.NAME),
+ self.types.skip_client_test_message(jms_message_type, receive_shim.NAME, 'RECEIVER'))
+ def inner_test_method(self):
+ self.run_test(self.sender_addr,
+ self.receiver_addr,
+ self.jms_message_type,
+ self.test_values,
+ send_shim,
+ receive_shim)
+
+ inner_test_method.__name__ = 'test_%s_%s->%s' % (jms_message_type[4:-5], send_shim.NAME, receive_shim.NAME)
+ setattr(cls, inner_test_method.__name__, inner_test_method)
+
+ class_name = jms_message_type[4:-5].title() + 'TestCase'
+ class_dict = {'__name__': class_name,
+ '__repr__': __repr__,
+ '__doc__': 'Test case for JMS message type \'%s\'' % jms_message_type,
+ 'jms_message_type': jms_message_type,
+ 'sender_addr': self.args.sender,
+ 'receiver_addr': self.args.receiver,
+ 'test_values': self.types.get_test_values(jms_message_type)} # tuple (tot_size, {...}
+ new_class = type(class_name, (JmsMessageTypeTestCase,), class_dict)
+ for send_shim, receive_shim in shim_product:
+ add_test_method(new_class, send_shim, receive_shim)
+
+ return new_class
#--- Main program start ---
if __name__ == '__main__':
-
- PROTON_CPP_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'jms_messages_test', 'Receiver')
- PROTON_CPP_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', 'jms_messages_test', 'Sender')
- PROTON_PYTHON_RECEIVER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'jms_messages_test', 'Receiver.py')
- PROTON_PYTHON_SENDER_SHIM = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', 'jms_messages_test', 'Sender.py')
- QIT_JMS_CLASSPATH_FILE = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms', 'cp.txt')
- if path.isfile(QIT_JMS_CLASSPATH_FILE):
- with open(QIT_JMS_CLASSPATH_FILE, 'r') as classpath_file:
- QIT_JMS_CLASSPATH = classpath_file.read()
- else:
- QIT_JMS_CLASSPATH = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms',
- 'qpid-interop-test-jms-shim-%s-jar-with-dependencies.jar' % QPID_JMS_SHIM_VER)
- QPID_JMS_RECEIVER_SHIM = 'org.apache.qpid.interop_test.jms_messages_test.Receiver'
- QPID_JMS_SENDER_SHIM = 'org.apache.qpid.interop_test.jms_messages_test.Sender'
-
- # SHIM_MAP contains an instance of each client language shim that is to be tested as a part of this test. For
- # every shim in this list, a test is dynamically constructed which tests it against itself as well as every
- # other shim in the list.
- #
- # As new shims are added, add them into this map to have them included in the test cases.
- SHIM_MAP = {qpid_interop_test.shims.ProtonCppShim.NAME: \
- qpid_interop_test.shims.ProtonCppShim(PROTON_CPP_SENDER_SHIM, PROTON_CPP_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython2Shim.NAME: \
- qpid_interop_test.shims.ProtonPython2Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.ProtonPython3Shim.NAME: \
- qpid_interop_test.shims.ProtonPython3Shim(PROTON_PYTHON_SENDER_SHIM, PROTON_PYTHON_RECEIVER_SHIM),
- qpid_interop_test.shims.QpidJmsShim.NAME: \
- qpid_interop_test.shims.QpidJmsShim(QIT_JMS_CLASSPATH, QPID_JMS_SENDER_SHIM, QPID_JMS_RECEIVER_SHIM),
- }
-
- ARGS = TestOptions(SHIM_MAP).args
- #print 'ARGS:', ARGS # debug
-
- # Add shims included from the command-line
- if ARGS.include_shim is not None:
- new_shim_map = {}
- for shim in ARGS.include_shim:
- try:
- new_shim_map[shim] = SHIM_MAP[shim]
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
- SHIM_MAP = new_shim_map
- # Remove shims excluded from the command-line
- elif ARGS.exclude_shim is not None:
- for shim in ARGS.exclude_shim:
- try:
- SHIM_MAP.pop(shim)
- except KeyError:
- print 'No such shim: "%s". Use --help for valid shims' % shim
- sys.exit(1) # Errors or failures present
-
- # Connect to broker to find broker type, or use --broker-type param if present
- if ARGS.broker_type is not None:
- if ARGS.broker_type == 'None':
- BROKER = None
- else:
- BROKER = ARGS.broker_type
- else:
- CONNECTION_PROPS = qpid_interop_test.broker_properties.get_broker_properties(ARGS.sender)
- if CONNECTION_PROPS is None:
- print 'WARNING: Unable to get connection properties - unknown broker'
- BROKER = 'unknown'
- else:
- BROKER = CONNECTION_PROPS[symbol(u'product')] if symbol(u'product') in CONNECTION_PROPS \
- else '<product not found>'
- BROKER_VERSION = CONNECTION_PROPS[symbol(u'version')] if symbol(u'version') in CONNECTION_PROPS \
- else '<version not found>'
- BROKER_PLATFORM = CONNECTION_PROPS[symbol(u'platform')] if symbol(u'platform') in CONNECTION_PROPS \
- else '<platform not found>'
- print 'Test Broker: %s v.%s on %s' % (BROKER, BROKER_VERSION, BROKER_PLATFORM)
- print
- sys.stdout.flush()
- if ARGS.no_skip:
- BROKER = None # Will cause all tests to run
-
- TYPES = JmsMessageTypes().get_types(ARGS)
-
- # TEST_CASE_CLASSES is a list that collects all the test classes that are constructed. One class is constructed
- # per AMQP type used as the key in map JmsMessageTypes.TYPE_MAP.
- TEST_CASE_CLASSES = []
-
- # TEST_SUITE is the final suite of tests that will be run and which contains all the dynamically created
- # type classes, each of which contains a test for the combinations of client shims
- TEST_SUITE = unittest.TestSuite()
-
- # Create test classes dynamically
- for jmt in sorted(TYPES.get_type_list()):
- if ARGS.exclude_type is None or jmt not in ARGS.exclude_type:
- test_case_class = create_testcase_class(jmt, product(SHIM_MAP.values(), repeat=2))
- TEST_CASE_CLASSES.append(test_case_class)
- TEST_SUITE.addTest(unittest.makeSuite(test_case_class))
-
- # Finally, run all the dynamically created tests
- RES = unittest.TextTestRunner(verbosity=2).run(TEST_SUITE)
- if not RES.wasSuccessful():
- sys.exit(1)
+ JMS_MESSAGES_TEST = JmsMessagesTest()
+ JMS_MESSAGES_TEST.run_test()
+ JMS_MESSAGES_TEST.write_logs()
+ if not JMS_MESSAGES_TEST.get_result():
+ sys.exit(1) # Errors or failures present
diff --git a/src/python/qpid_interop_test/qit.py b/src/python/qpid_interop_test/qit.py
new file mode 100644
index 0000000..ec805ae
--- /dev/null
+++ b/src/python/qpid_interop_test/qit.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+
+"""
+Script to run all available tests
+"""
+
+#
+# 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 sys
+
+import qpid_interop_test.amqp_large_content_test
+import qpid_interop_test.amqp_types_test
+import qpid_interop_test.jms_hdrs_props_test
+import qpid_interop_test.jms_messages_test
+
+if __name__ == '__main__':
+
+ print('\n==== Test %s ====' % qpid_interop_test.amqp_types_test.AmqpTypesTest.TEST_NAME)
+ AMQP_TYPES_TEST = qpid_interop_test.amqp_types_test.AmqpTypesTest()
+ AMQP_TYPES_TEST.run_test()
+ AMQP_TYPES_TEST.write_logs()
+
+ print('\n==== Test %s ====' % qpid_interop_test.amqp_large_content_test.AmqpLargeContentTest.TEST_NAME)
+ AMQP_LARGE_CONTENT_TEST = qpid_interop_test.amqp_large_content_test.AmqpLargeContentTest()
+ AMQP_LARGE_CONTENT_TEST.run_test()
+ AMQP_LARGE_CONTENT_TEST.write_logs()
+
+ print('\n==== Test %s ====' % qpid_interop_test.jms_messages_test.JmsMessagesTest.TEST_NAME)
+ JMS_MESSAGES_TEST = qpid_interop_test.jms_messages_test.JmsMessagesTest()
+ JMS_MESSAGES_TEST.run_test()
+ JMS_MESSAGES_TEST.write_logs()
+
+ print('\n==== Test %s ====' % qpid_interop_test.jms_hdrs_props_test.JmsHdrsPropsTest.TEST_NAME)
+ JMS_MESSAGES_TEST = qpid_interop_test.jms_hdrs_props_test.JmsHdrsPropsTest()
+ JMS_MESSAGES_TEST.run_test()
+ JMS_MESSAGES_TEST.write_logs()
+
+ if not AMQP_TYPES_TEST.get_result() or not AMQP_LARGE_CONTENT_TEST.get_result() or \
+ not JMS_MESSAGES_TEST.get_result() or not JMS_MESSAGES_TEST.get_result():
+ sys.exit(1)
diff --git a/src/python/qpid_interop_test/qit_common.py b/src/python/qpid_interop_test/qit_common.py
new file mode 100644
index 0000000..be0d15b
--- /dev/null
+++ b/src/python/qpid_interop_test/qit_common.py
@@ -0,0 +1,363 @@
+"""
+Module containing common classes
+"""
+
+#
+# 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 argparse
+from os import getcwd, getenv, path
+import sys
+from time import time
+import unittest
+
+from proton import symbol
+import qpid_interop_test.broker_properties
+import qpid_interop_test.shims
+import qpid_interop_test.xunit_log
+
+# TODO: propose a sensible default when installation details are worked out
+QIT_INSTALL_PREFIX = getenv('QIT_INSTALL_PREFIX')
+if QIT_INSTALL_PREFIX is None:
+ print('ERROR: Environment variable QIT_INSTALL_PREFIX is not set')
+ sys.exit(1)
+QIT_TEST_SHIM_HOME = path.join(QIT_INSTALL_PREFIX, 'libexec', 'qpid_interop_test', 'shims')
+
+QPID_JMS_SHIM_VER = '0.1.0'
+
+DEFUALT_XUNIT_LOG_DIR = path.join(getcwd(), 'xunit_logs')
+
+
+class QitTestTypeMap(object):
+ """
+ Class which contains all the described types and the test values to be used in testing against those types.
+ """
+
+ # type_map: Map containing all described types as the indecies, and a list of values to be used in testing
+ # that type as a list of values.
+ #
+ # Format: {'type_1' : [val_1_1, val_1_2, ...],
+ # 'type_2' : [val_2_1, val_2_2, ...],
+ # ...
+ # }
+ type_map = {}
+
+ # broker_skip: For known broker issues where a type would cause a test to fail or hang,
+ # entries in broker_skip will cause the test to be skipped with a message.
+ # This is a map containing AMQP types as a key, and a list of brokers for which this
+ # type should be skipped.
+ # Format: {'jms_msg_type_1' : {'broker_1' : 'skip msg for broker_1',
+ # 'broker_2' : 'skip msg for broker_2',
+ # ...
+ # },
+ # 'jms_msg_type_2' : {'broker_1' : 'skip msg for broker_1',
+ # 'broker_2' : 'skip msg for broker_2',
+ # ...
+ # },
+ # ...
+ # }
+ # where broker_1, broker_2, ... are broker product names as defined by the
+ # connection property string it returns.
+ broker_skip = {}
+
+ # client_skip: For known client issues where a type would cause a test to fail or hang,
+ # entries in client_skip will cause the test to be skipped with a message.
+ # This is a map containing AMQP types as a key, and a list of clients for which this
+ # type should be skipped.
+ # Format: {'jms_msg_type_1' : {'client_1' : 'skip msg for client_1',
+ # 'client_2' : 'skip msg for client_2',
+ # ...
+ # },
+ # 'jms_msg_type_2' : {'client_1' : 'skip msg for client_1',
+ # 'client_2' : 'skip msg for client_2',
+ # ...
+ # },
+ # ...
+ # }
+ # where client_1, client_2, ... are client product names as defined by the
+ # test shim NAME.
+ client_skip = {}
+
+ def get_type_list(self):
+ """Return a list of types which this test suite supports"""
+ return self.type_map.keys()
+
+ def get_types(self, args):
+ """Return the list of types"""
+ if "include_type" in args and args.include_type is not None:
+ new_type_map = {}
+ for this_type in args.include_type:
+ try:
+ new_type_map[this_type] = self.type_map[this_type]
+ except KeyError:
+ print('No such type: "%s". Use --help for valid types' % this_type)
+ sys.exit(1) # Errors or failures present
+ self.type_map = new_type_map
+ if "exclude_type" in args and args.exclude_type is not None:
+ for this_type in args.exclude_type:
+ try:
+ self.type_map.pop(this_type)
+ except KeyError:
+ print('No such type: "%s". Use --help for valid types' % this_type)
+ sys.exit(1) # Errors or failures present
+ return self
+
+ def get_test_values(self, test_type):
+ """Return test values to use when testing the supplied type."""
+ if test_type not in self.type_map.keys():
+ return None
+ return self.type_map[test_type]
+
+ def skip_test_message(self, test_type, broker_name):
+ """Return the message to use if a test is skipped"""
+ if test_type in self.broker_skip.keys():
+ if broker_name in self.broker_skip[test_type]:
+ return str("BROKER: " + self.broker_skip[test_type][broker_name])
+ return None
+
+ def skip_test(self, test_type, broker_name):
+ """Return boolean True if test should be skipped"""
+ return test_type in self.broker_skip.keys() and \
+ broker_name in self.broker_skip[test_type]
+
+ def skip_client_test_message(self, test_type, client_name, role):
+ """Return the message to use if a test is skipped"""
+ if test_type in self.client_skip.keys():
+ if client_name in self.client_skip[test_type]:
+ return str(role + ": " + self.client_skip[test_type][client_name])
+ return None
+
+ def skip_client_test(self, test_type, client_name):
+ """Return boolean True if test should be skipped"""
+ return test_type in self.client_skip.keys() and \
+ client_name in self.client_skip[test_type]
+
+ @staticmethod
+ def merge_dicts(*dict_args):
+ """Static method to merge two or more dictionaries"""
+ res = {}
+ for this_dict in dict_args:
+ res.update(this_dict)
+ return res
+
+
+class QitCommonTestOptions(object):
+ """
+ Class controlling common command-line arguments used to control tests.
+ """
+ def __init__(self, test_description, shim_map, default_xunit_dir=DEFUALT_XUNIT_LOG_DIR):
+ self._parser = argparse.ArgumentParser(description=test_description)
+ self._parser.add_argument('--sender', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
+ help='Node to which test suite will send messages.')
+ self._parser.add_argument('--receiver', action='store', default='localhost:5672', metavar='IP-ADDR:PORT',
+ help='Node from which test suite will receive messages.')
+ self._parser.add_argument('--no-skip', action='store_true',
+ help='Do not skip tests that are excluded by default for reasons of a known bug')
+ self._parser.add_argument('--broker-type', action='store', metavar='BROKER_NAME',
+ help='Disable test of broker type (using connection properties) by specifying' +
+ ' the broker name, or "None".')
+ self._parser.add_argument('--xunit-log', action='store_true',
+ help='Enable xUnit logging of test results')
+ self._parser.add_argument('--xunit-log-dir', action='store', default=default_xunit_dir,
+ metavar='LOG-DIR-PATH',
+ help='Default xUnit log directory where xUnit logs are written [xunit_logs dir' +
+ ' in current directory (%s)]' % default_xunit_dir)
+
+ shim_group = self._parser.add_mutually_exclusive_group()
+ shim_group.add_argument('--include-shim', action='append', metavar='SHIM-NAME',
+ help='Name of shim to include. Supported shims:\n%s' % sorted(shim_map.keys()))
+ shim_group.add_argument('--exclude-shim', action='append', metavar='SHIM-NAME',
+ help='Name of shim to exclude. Supported shims: see "include-shim" above')
+
+ def args(self):
+ """Return the parsed args"""
+ return self._parser.parse_args()
+
+ def print_help(self, file=None):
+ """Print help"""
+ self._parser.print_help(file)
+
+ def print_usage(self, file=None):
+ """Print usage"""
+ self._parser.print_usage(file)
+
+
+class QitTestCase(unittest.TestCase):
+ """
+ Abstract base class for QIT test cases
+ """
+
+ def __init__(self, methodName='runTest'):
+ super(QitTestCase, self).__init__(methodName)
+ self.duration = 0
+
+ def setUp(self):
+ """Called when test starts"""
+ self.start_time = time()
+
+ def tearDown(self):
+ """Called when test finishes"""
+ self.duration = time() - self.start_time
+
+ def name(self):
+ """Return test name"""
+ return self._testMethodName
+
+
+class QitTest(object):
+ """
+ Top-level test class with test entry-point
+ """
+
+ TEST_NAME = ''
+
+ def __init__(self, test_options_class, test_values_class):
+ self._create_shim_map()
+ self.args = test_options_class(self.shim_map).args()
+ self._modify_shim_map()
+ self._discover_broker()
+ self.types = test_values_class().get_types(self.args)
+ self._generate_tests()
+ self.test_result = None
+ self.duration = 0
+
+ def get_result(self):
+ """Get success of test run, True = success, False = failure/error"""
+ if self.test_result is None:
+ return None
+ return self.test_result.wasSuccessful()
+
+ def run_test(self):
+ """Run the test"""
+ start_time = time()
+ self.test_result = unittest.TextTestRunner(verbosity=2).run(self.test_suite)
+ self.duration = time() - start_time
+
+ def write_logs(self):
+ """Write the logs"""
+ if self.args.xunit_log_dir is not None:
+ xunit_log_dir = self.args.xunit_log_dir
+ else:
+ xunit_log_dir = DEFUALT_XUNIT_LOG_DIR
+ qpid_interop_test.xunit_log.Xunit(self.args.xunit_log, self.TEST_NAME, xunit_log_dir, self.test_suite,
+ self.test_result, self.duration)
+
+ def _create_shim_map(self):
+ """Create a shim map {'shim_name': <shim_instance>}"""
+ proton_cpp_rcv_shim = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', self.TEST_NAME, 'Receiver')
+ proton_cpp_snd_shim = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-cpp', self.TEST_NAME, 'Sender')
+ proton_python_rcv_shim = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', self.TEST_NAME, 'Receiver.py')
+ proton_python_snd_shim = path.join(QIT_TEST_SHIM_HOME, 'qpid-proton-python', self.TEST_NAME, 'Sender.py')
+
+ self.shim_map = {qpid_interop_test.shims.ProtonCppShim.NAME: \
+ qpid_interop_test.shims.ProtonCppShim(proton_cpp_snd_shim, proton_cpp_rcv_shim),
+ qpid_interop_test.shims.ProtonPython2Shim.NAME: \
+ qpid_interop_test.shims.ProtonPython2Shim(proton_python_snd_shim, proton_python_rcv_shim),
+ qpid_interop_test.shims.ProtonPython3Shim.NAME: \
+ qpid_interop_test.shims.ProtonPython3Shim(proton_python_snd_shim, proton_python_rcv_shim),
+ }
+
+ # Add shims that need detection during installation only if the necessary bits are present
+ # Rhea Javascript client
+ rhea_rcv_shim = path.join(QIT_TEST_SHIM_HOME, 'rhea-js', self.TEST_NAME, 'Receiver.js')
+ rhea_snd_shim = path.join(QIT_TEST_SHIM_HOME, 'rhea-js', self.TEST_NAME, 'Sender.js')
+ if path.isfile(rhea_rcv_shim) and path.isfile(rhea_snd_shim):
+ self.shim_map[qpid_interop_test.shims.RheaJsShim.NAME] = \
+ qpid_interop_test.shims.RheaJsShim(rhea_snd_shim, rhea_rcv_shim)
+ else:
+ print('WARNING: Rhea Javascript shims not found')
+
+ # AMQP DotNetLite client
+ amqpnetlite_rcv_shim = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', self.TEST_NAME, 'Receiver.exe')
+ amqpnetlite_snd_shim = path.join(QIT_TEST_SHIM_HOME, 'amqpnetlite', self.TEST_NAME, 'Sender.exe')
+ if path.isfile(amqpnetlite_rcv_shim) and path.isfile(amqpnetlite_snd_shim):
+ self.shim_map[qpid_interop_test.shims.AmqpNetLiteShim.NAME] = \
+ qpid_interop_test.shims.AmqpNetLiteShim(amqpnetlite_snd_shim, amqpnetlite_rcv_shim)
+ else:
+ print('WARNING: AMQP DotNetLite shims not found')
+
+ def _modify_shim_map(self):
+ """Modify shim_map based on args"""
+ # Use only shims included from the command-line
+ if self.args.include_shim is not None:
+ temp_shim_map = {}
+ for shim in self.args.include_shim:
+ try:
+ temp_shim_map[shim] = self.shim_map[shim]
+ except KeyError:
+ print('No such shim: "%s". Use --help for valid shims' % shim)
+ sys.exit(1) # Errors or failures present
+ self.shim_map = temp_shim_map
+ # Remove shims excluded from the command-line
+ elif self.args.exclude_shim is not None:
+ for shim in self.args.exclude_shim:
+ try:
+ self.shim_map.pop(shim)
+ except KeyError:
+ print('No such shim: "%s". Use --help for valid shims' % shim)
+ sys.exit(1) # Errors or failures present
+
+ def _discover_broker(self):
+ """Connect to broker and get connection properties to discover broker name and version"""
+ if self.args.broker_type is not None:
+ if self.args.broker_type == 'None':
+ self.broker = None
+ else:
+ self.broker = self.broker.broker_type
+ else:
+ connection_props = qpid_interop_test.broker_properties.get_broker_properties(self.args.sender)
+ if connection_props is None:
+ print('WARNING: Unable to get connection properties - unknown broker')
+ self.broker = 'unknown'
+ else:
+ self.broker = connection_props[symbol(u'product')] if symbol(u'product') in connection_props \
+ else '<product not found>'
+ self.broker_version = connection_props[symbol(u'version')] if symbol(u'version') in connection_props \
+ else '<version not found>'
+ self.broker_platform = connection_props[symbol(u'platform')] if symbol(u'platform') in \
+ connection_props else '<platform not found>'
+ print('Test Broker: %s v.%s on %s\n' % (self.broker, self.broker_version, self.broker_platform))
+ sys.stdout.flush()
+ if self.args.no_skip:
+ self.broker = None # Will cause all tests to run, no matter which broker
+
+ def _generate_tests(self):
+ """Generate tests dynamically - each subclass must override this function"""
+ self.test_suite = None
+
+
+class QitJmsTest(QitTest):
+ """Class with specialized Java and classpath functionality"""
+
+ def _create_shim_map(self):
+ """Create a shim map {'shim_name': <shim_instance>}"""
+ super(QitJmsTest, self)._create_shim_map()
+
+ qpid_jms_rcv_shim = 'org.apache.qpid.interop_test.%s.Receiver' % self.TEST_NAME
+ qpid_jms_snd_shim = 'org.apache.qpid.interop_test.%s.Sender' % self.TEST_NAME
+ classpath_file_name = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms', 'cp.txt')
+ if path.isfile(classpath_file_name):
+ with open(classpath_file_name, 'r') as classpath_file:
+ classpath = classpath_file.read()
+ else:
+ classpath = path.join(QIT_TEST_SHIM_HOME, 'qpid-jms',
+ 'qpid-interop-test-jms-shim-%s-jar-with-dependencies.jar' % QPID_JMS_SHIM_VER)
+
+ self.shim_map[qpid_interop_test.shims.QpidJmsShim.NAME] = \
+ qpid_interop_test.shims.QpidJmsShim(classpath, qpid_jms_snd_shim, qpid_jms_rcv_shim)
diff --git a/src/python/qpid_interop_test/shims.py b/src/python/qpid_interop_test/shims.py
index 6ba5da6..03b8dc0 100644
--- a/src/python/qpid_interop_test/shims.py
+++ b/src/python/qpid_interop_test/shims.py
@@ -53,27 +53,30 @@
self.join(timeout)
self.kill()
- def kill(self, num_attempts=2, wait_time=2):
+ def kill(self):
+ """
+ First try terminating, then killing this thread
+ """
if self.is_alive():
if self.proc is not None:
if self._terminate_pg_loop():
if self._kill_pg_loop():
- print '\n ERROR: Thread %s (pid=%d) alive after kill' % (self.name, self.proc.pid)
+ print('\n ERROR: Thread %s (pid=%d) alive after kill' % (self.name, self.proc.pid))
else:
- print 'Killed'
+ print('Killed')
stdout.flush()
else:
- print 'Terminated'
+ print('Terminated')
stdout.flush()
else:
- print 'ERROR: shims.join_or_kill(): Process joined and is alive, yet proc is None.'
+ print('ERROR: shims.join_or_kill(): Process joined and is alive, yet proc is None.')
def _terminate_pg_loop(self, num_attempts=2, wait_time=2):
cnt = 0
while cnt < num_attempts and self.is_alive():
cnt += 1
- print '\n Thread %s (pid=%d) alive after timeout, terminating (try #%d)...' % (self.name, self.proc.pid,
- cnt),
+ print('\n Thread %s (pid=%d) alive after timeout, terminating (try #%d)...' % (self.name, self.proc.pid,
+ cnt),)
stdout.flush()
killpg(getpgid(self.proc.pid), SIGTERM)
sleep(wait_time)
@@ -83,8 +86,8 @@
cnt = 0
while cnt < num_attempts and self.is_alive():
cnt += 1
- print '\n Thread %s (pid=%d) alive after terminate, killing (try #%d)...' % (self.name, self.proc.pid,
- cnt),
+ print('\n Thread %s (pid=%d) alive after terminate, killing (try #%d)...' % (self.name, self.proc.pid,
+ cnt),)
stdout.flush()
killpg(getpgid(self.proc.pid), SIGKILL)
sleep(wait_time)
@@ -96,21 +99,22 @@
def __init__(self, use_shell_flag, send_shim_args, broker_addr, queue_name, test_key, json_test_str, python3_flag):
super(Sender, self).__init__('sender_thread_%s' % queue_name)
if send_shim_args is None:
- print 'ERROR: Sender: send_shim_args == None'
+ print('ERROR: Sender: send_shim_args == None')
self.use_shell_flag = use_shell_flag
self.arg_list.extend(send_shim_args)
self.arg_list.extend([broker_addr, queue_name, test_key, json_test_str])
self.env = deepcopy(environ)
if python3_flag:
- self.env['PYTHONPATH']=self.env['PYTHON3PATH']
+ self.env['PYTHONPATH'] = self.env['PYTHON3PATH']
def run(self):
"""Thread starts here"""
try:
#print str('\n>>SNDR>>' + str(self.arg_list)) # DEBUG - useful to see command-line sent to shim
- self.proc = Popen(self.arg_list, stdout=PIPE, stderr=PIPE, shell=self.use_shell_flag, preexec_fn=setsid, env=self.env)
+ self.proc = Popen(self.arg_list, stdout=PIPE, stderr=PIPE, shell=self.use_shell_flag, preexec_fn=setsid,
+ env=self.env)
(stdoutdata, stderrdata) = self.proc.communicate()
- if len(stderrdata) > 0:
+ if stderrdata: # length > 0
#print '<<SNDR ERROR<<', stderrdata # DEBUG - useful to see shim's failure message
self.return_obj = (stdoutdata, stderrdata)
else:
@@ -124,7 +128,7 @@
else: # Make a single line of all the bits and return that
self.return_obj = stdoutdata
except OSError as exc:
- self.return_obj = str(exc) + ': shim=' + self.arg_list[0]
+ self.return_obj = str(exc) + ': shim=' + self.arg_list[0]
except CalledProcessError as exc:
self.return_obj = str(exc) + '\n\nOutput:\n' + exc.output
@@ -134,24 +138,24 @@
def __init__(self, receive_shim_args, broker_addr, queue_name, test_key, json_test_str, python3_flag):
super(Receiver, self).__init__('receiver_thread_%s' % queue_name)
if receive_shim_args is None:
- print 'ERROR: Receiver: receive_shim_args == None'
+ print('ERROR: Receiver: receive_shim_args == None')
self.arg_list.extend(receive_shim_args)
self.arg_list.extend([broker_addr, queue_name, test_key, json_test_str])
self.env = deepcopy(environ)
if python3_flag:
- self.env['PYTHONPATH']=self.env['PYTHON3PATH']
+ self.env['PYTHONPATH'] = self.env['PYTHON3PATH']
def run(self):
"""Thread starts here"""
try:
- #print str('\n>>RCVR>>' + str(self.arg_list)) # DEBUG - useful to see command-line sent to shim
+ #print(str('\n>>RCVR>>' + str(self.arg_list))) # DEBUG - useful to see command-line sent to shim
self.proc = Popen(self.arg_list, stdout=PIPE, stderr=PIPE, preexec_fn=setsid, env=self.env)
(stdoutdata, stderrdata) = self.proc.communicate()
- if len(stderrdata) > 0:
- #print '<<RCVR ERROR<<', stderrdata # DEBUG - useful to see shim's failure message
+ if stderrdata: # length > 0
+ #print('<<RCVR ERROR<<', stderrdata) # DEBUG - useful to see shim's failure message
self.return_obj = (stdoutdata, stderrdata)
else:
- #print '<<RCVR<<', stdoutdata # DEBUG - useful to see text received from shim
+ #print('<<RCVR<<', stdoutdata) # DEBUG - useful to see text received from shim
str_tvl = stdoutdata.split('\n')[0:-1] # remove trailing \n
if len(str_tvl) == 2:
try:
@@ -167,7 +171,7 @@
class Shim(object):
"""Abstract shim class, parent of all shims."""
- NAME = None
+ NAME = ''
JMS_CLIENT = False # Enables certain JMS-specific message checks
def __init__(self, sender_shim, receiver_shim):
self.sender_shim = sender_shim
@@ -243,12 +247,12 @@
def get_java_class_path(self):
"""Method to construct and return the Java class path necessary to run the shim"""
- return self.dependency_class_path;
+ return self.dependency_class_path
class AmqpNetLiteShim(Shim):
"""Shim for AMQP.Net Lite client"""
NAME = 'AmqpNetLite'
def __init__(self, sender_shim, receiver_shim):
super(AmqpNetLiteShim, self).__init__(sender_shim, receiver_shim)
- self.send_params = ['mono' ,self.sender_shim]
+ self.send_params = ['mono', self.sender_shim]
self.receive_params = ['mono', self.receiver_shim]
diff --git a/src/python/qpid_interop_test/test_type_map.py b/src/python/qpid_interop_test/test_type_map.py
deleted file mode 100644
index 59e5686..0000000
--- a/src/python/qpid_interop_test/test_type_map.py
+++ /dev/null
@@ -1,137 +0,0 @@
-"""
-Module containing Error classes for interop testing
-"""
-#
-# 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 sys
-
-class TestTypeMap(object):
- """
- Class which contains all the described types and the test values to be used in testing against those types.
- """
-
- # TYPE_MAP: Map containing all described types as the indecies, and a list of values to be used in testing
- # that type as a list of values.
- #
- # Format: {'type_1' : [val_1_1, val_1_2, ...],
- # 'type_2' : [val_2_1, val_2_2, ...],
- # ...
- # }
- TYPE_MAP = {}
-
- # BROKER_SKIP: For known broker issues where a type would cause a test to fail or hang,
- # entries in BROKER_SKIP will cause the test to be skipped with a message.
- # This is a map containing AMQP types as a key, and a list of brokers for which this
- # type should be skipped.
- # Format: {'jms_msg_type_1' : {'broker_1' : 'skip msg for broker_1',
- # 'broker_2' : 'skip msg for broker_2',
- # ...
- # },
- # 'jms_msg_type_2' : {'broker_1' : 'skip msg for broker_1',
- # 'broker_2' : 'skip msg for broker_2',
- # ...
- # },
- # ...
- # }
- # where broker_1, broker_2, ... are broker product names as defined by the
- # connection property string it returns.
- BROKER_SKIP = {}
-
- # CLIENT_SKIP: For known client issues where a type would cause a test to fail or hang,
- # entries in CLIENT_SKIP will cause the test to be skipped with a message.
- # This is a map containing AMQP types as a key, and a list of clients for which this
- # type should be skipped.
- # Format: {'jms_msg_type_1' : {'client_1' : 'skip msg for client_1',
- # 'client_2' : 'skip msg for client_2',
- # ...
- # },
- # 'jms_msg_type_2' : {'client_1' : 'skip msg for client_1',
- # 'client_2' : 'skip msg for client_2',
- # ...
- # },
- # ...
- # }
- # where client_1, client_2, ... are client product names as defined by the
- # test shim NAME.
- CLIENT_SKIP = {}
-
- def __init__(self):
- pass
-
- def get_type_list(self):
- """Return a list of types which this test suite supports"""
- return self.TYPE_MAP.keys()
-
- def get_types(self, args):
- if "include_type" in args and args.include_type is not None:
- new_type_map = {}
- for type in args.include_type:
- try:
- new_type_map[type] = self.TYPE_MAP[type]
- except KeyError:
- print('No such type: "%s". Use --help for valid types' % type)
- sys.exit(1) # Errors or failures present
- self.TYPE_MAP = new_type_map
- if "exclude_type" in args and args.exclude_type is not None:
- for type in args.exclude_type:
- try:
- self.TYPE_MAP.pop(type)
- except KeyError:
- print('No such type: "%s". Use --help for valid types' % type)
- sys.exit(1) # Errors or failures present
- return self
-
- def get_test_values(self, test_type):
- """Return test values to use when testing the supplied type."""
- if test_type not in self.TYPE_MAP.keys():
- return None
- return self.TYPE_MAP[test_type]
-
- def skip_test_message(self, test_type, broker_name):
- """Return the message to use if a test is skipped"""
- if test_type in self.BROKER_SKIP.keys():
- if broker_name in self.BROKER_SKIP[test_type]:
- return str("BROKER: " + self.BROKER_SKIP[test_type][broker_name])
- return None
-
- def skip_test(self, test_type, broker_name):
- """Return boolean True if test should be skipped"""
- return test_type in self.BROKER_SKIP.keys() and \
- broker_name in self.BROKER_SKIP[test_type]
-
- def skip_client_test_message(self, test_type, client_name, role):
- """Return the message to use if a test is skipped"""
- if test_type in self.CLIENT_SKIP.keys():
- if client_name in self.CLIENT_SKIP[test_type]:
- return str(role + ": " + self.CLIENT_SKIP[test_type][client_name])
- return None
-
- def skip_client_test(self, test_type, client_name):
- """Return boolean True if test should be skipped"""
- return test_type in self.CLIENT_SKIP.keys() and \
- client_name in self.CLIENT_SKIP[test_type]
-
- @staticmethod
- def merge_dicts(*dict_args):
- """Static method to merge two or more dictionaries"""
- res = {}
- for this_dict in dict_args:
- res.update(this_dict)
- return res
diff --git a/src/python/qpid_interop_test/xunit_log.py b/src/python/qpid_interop_test/xunit_log.py
new file mode 100644
index 0000000..0d54631
--- /dev/null
+++ b/src/python/qpid_interop_test/xunit_log.py
@@ -0,0 +1,116 @@
+"""
+Module providing xUnit logging functionality
+"""
+
+#
+# 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 datetime
+import os.path
+import xml.dom.minidom
+import xml.etree.cElementTree
+
+from qpid_interop_test.interop_test_errors import InteropTestError
+
+class Xunit(object):
+ """Class that provides test reporting in xUnit format"""
+ def __init__(self, enable_flag, test_name, xunit_log_dir, test_suite, test_result, duration):
+ self.root = None
+ if enable_flag:
+ self.date_time_str = datetime.datetime.now().strftime('%Y-%m-%dT%H-%M-%S')
+ self._check_make_dir(xunit_log_dir)
+ self.log_file = self._open(test_name, xunit_log_dir)
+ self.process_result(test_name, test_suite, test_result, duration)
+ self.write_log()
+
+ @staticmethod
+ def _check_make_dir(path):
+ """
+ Check if path exists as a directory. If not, create it (or raise exception if it exists as a non-directory)
+ """
+ if os.path.exists(path):
+ if not os.path.isdir(path):
+ raise InteropTestError('%s exists, but is not a directory' % path)
+ else:
+ os.makedirs(path)
+
+ def _open(self, test_name, path):
+ """Open file for writing"""
+ file_name = '%s.%s.xml' % (test_name, self.date_time_str)
+ try:
+ return open(os.path.join(path, file_name), 'w')
+ except IOError as err:
+ raise InteropTestError('Unable to open xUnit log file: %s' % err)
+
+ @staticmethod
+ def _prettify(element):
+ """Return a pretty-printed XML string for element"""
+ rough_string = xml.etree.ElementTree.tostring(element, 'utf-8')
+ reparsed = xml.dom.minidom.parseString(rough_string)
+ return reparsed.toprettyxml(indent=' ', encoding='utf-8')
+
+ def process_result(self, test_name, test_suite, test_result, duration):
+ """Create the xUnit XML tree"""
+ self.root = xml.etree.cElementTree.Element('testsuite')
+ self.root.set('timestamp', self.date_time_str)
+ self.root.set('hostname', 'localhost')
+ self.root.set('name', test_name)
+ self.root.set('tests', str(test_result.testsRun))
+ self.root.set('errors', str(len(test_result.errors)))
+ self.root.set('failures', str(len(test_result.failures)))
+ self.root.set('skipped', str(len(test_result.skipped)))
+ self.root.set('time', '%.3f' % duration)
+
+ errors = {}
+ for error_tup in test_result.errors:
+ errors[error_tup[0]] = error_tup[1]
+
+ failures = {}
+ for failure_tup in test_result.failures:
+ failures[failure_tup[0]] = failure_tup[1]
+
+ skips = {}
+ for skip_tup in test_result.skipped:
+ skips[skip_tup[0]] = skip_tup[1]
+
+ for type_test_suite in test_suite:
+ for test_case in type_test_suite:
+ test_case_child = xml.etree.ElementTree.SubElement(self.root, 'testcase')
+ test_case_child.set('class', test_case.id())
+ test_case_child.set('name', test_case.name())
+ test_case_child.set('time', '%.3f' % test_case.duration)
+
+ # Handle errors, failures and skipped tests
+ if test_case in errors:
+ error_child = xml.etree.ElementTree.SubElement(test_case_child, 'error')
+ error_child.set('type', '')
+ error_child.text = errors[test_case]
+ elif test_case in failures:
+ failure_child = xml.etree.ElementTree.SubElement(test_case_child, 'failure')
+ failure_child.set('type', '')
+ failure_child.text = failures[test_case]
+ elif test_case in skips:
+ skip_child = xml.etree.ElementTree.SubElement(test_case_child, 'skipped')
+ skip_child.set('type', '')
+ skip_child.text = skips[test_case]
+
+ def write_log(self):
+ """Write the xUnit log file"""
+ if self.log_file is not None and self.root is not None:
+ self.log_file.write(self._prettify(self.root))
