| #!/usr/bin/env python |
| |
| """ |
| Module to test AMQP complex types across different clients |
| """ |
| |
| # |
| # 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 copy |
| import signal |
| import sys |
| import unittest |
| |
| import itertools |
| |
| import qpid_interop_test.qit_common |
| from qpid_interop_test.qit_errors import InteropTestError, InteropTestTimeout |
| |
| DEFAULT_TEST_TIMEOUT = 10 # seconds |
| |
| class AmqpComplexTypes(qpid_interop_test.qit_common.QitTestTypeMap): |
| """ |
| Class which contains all the described AMQP complex types and the test values to be used in testing. |
| """ |
| |
| default_type_list = ['array', |
| 'list', |
| 'map', |
| ] |
| _type_list = [] |
| default_subtype_list = [#'None', Disabled until PROTON-1858 makes it into current proton releases (ie >= 0.23) |
| #'null', Disabled until PROTON-1858 makes it into current proton releases (ie >= 0.23) |
| 'boolean', |
| 'ubyte', |
| 'ushort', |
| 'uint', |
| 'ulong', |
| 'byte', |
| 'short', |
| 'int', |
| 'long', |
| 'float', |
| 'double', |
| 'decimal32', |
| 'decimal64', |
| 'decimal128', |
| 'char', |
| 'timestamp', |
| 'uuid', |
| 'binary', |
| 'string', |
| 'symbol', |
| 'array', |
| 'list', |
| 'map', |
| '*', # Reserved type string used for mixed subtypes |
| ] |
| _subtype_list = [] |
| |
| # 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 = { |
| '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', |
| 'qpid-dispatch-router': 'decimal32 and decimal64 sent byte reversed: PROTON-1160',}, |
| 'decimal64': {'ActiveMQ': 'decimal32 and decimal64 sent byte reversed: PROTON-1160', |
| 'qpid-cpp': 'decimal64 not supported on qpid-cpp broker: QPIDIT-6, QPID-6328', |
| 'apache-activemq-artemis': 'decimal32 and decimal64 sent byte reversed: PROTON-1160', |
| 'qpid-dispatch-router': 'decimal32 and decimal64 sent byte reversed: PROTON-1160',}, |
| } |
| |
| client_skip = {} |
| |
| def get_types(self, args): |
| """Return the list of types""" |
| if "include_type" in args and args.include_type is not None: |
| for this_type in args.include_type: |
| if this_type in self.default_type_list: |
| self._type_list.append(this_type) |
| else: |
| print('No such type: "%s". Use --help for valid types' % this_type) |
| sys.exit(1) # Errors or failures present |
| elif "exclude_type" in args and args.exclude_type is not None: |
| self._type_list = copy.deepcopy(self.default_type_list) |
| for this_type in args.exclude_type: |
| try: |
| self._type_list.remove(this_type) |
| except (KeyError, ValueError): |
| print('No such type: "%s". Use --help for valid types' % this_type) |
| sys.exit(1) # Errors or failures present |
| else: |
| self._type_list = self.default_type_list |
| if "include_subtype" in args and args.include_subtype is not None: |
| for this_subtype in args.include_subtype: |
| if this_subtype in self.default_subtype_list: |
| self._subtype_list.append(this_subtype) |
| else: |
| print('No such subtype: "%s". Use --help for valid types' % this_subtype) |
| sys.exit(1) # Errors or failures present |
| elif "exclude_subtype" in args and args.exclude_subtype is not None: |
| self._subtype_list = copy.deepcopy(self.default_subtype_list) |
| for this_subtype in args.exclude_subtype: |
| try: |
| self._subtype_list.remove(this_subtype) |
| except KeyError: |
| print('No such subtype "%s". Use --help for valid subtypes' % this_subtype) |
| sys.exit(1) |
| else: |
| self._subtype_list = self.default_subtype_list |
| return self |
| |
| def get_type_list(self): |
| """Return a list of types""" |
| return self._type_list |
| |
| def get_subtype_list(self): |
| """Return a list of subtypes which this test suite supports""" |
| return self._subtype_list |
| |
| |
| class AmqpComplexTypeTestCase(qpid_interop_test.qit_common.QitTestCase): |
| """Abstract base class for AMQP Complex Type test cases""" |
| |
| def run_test(self, sender_addr, receiver_addr, amqp_type, amqp_subtype, send_shim, receive_shim, timeout): |
| """ |
| Runs 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. |
| """ |
| test_name = 'amqp_complex_types_test.%s.%s.%s.%s' % (amqp_type, amqp_subtype, send_shim.NAME, receive_shim.NAME) |
| queue_name = 'qit.%s' % test_name |
| |
| # Start the receive shim first (for queueless brokers/dispatch) |
| receiver = receive_shim.create_receiver(receiver_addr, queue_name, amqp_type, amqp_subtype) |
| |
| # Start the send shim |
| sender = send_shim.create_sender(sender_addr, queue_name, amqp_type, amqp_subtype) |
| |
| # Wait for sender, process return string |
| try: |
| send_obj = sender.wait_for_completion(timeout) |
| except (KeyboardInterrupt, InteropTestTimeout): |
| receiver.send_signal(signal.SIGINT) |
| raise |
| if send_obj is not None: |
| if isinstance(send_obj, str): |
| if send_obj: # len > 0 |
| receiver.send_signal(signal.SIGINT) |
| raise InteropTestError('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj)) |
| else: |
| receiver.send_signal(signal.SIGINT) |
| raise InteropTestError('Send shim \'%s\':\n%s' % (send_shim.NAME, send_obj)) |
| |
| # Wait for receiver, process return string |
| receive_obj = receiver.wait_for_completion(timeout) |
| if isinstance(receive_obj, tuple): |
| if len(receive_obj) == 2: |
| return_amqp_type, pass_text = receive_obj |
| self.assertEqual(return_amqp_type, amqp_type, |
| msg='AMQP type error:\n\n sent:%s\n\n received:%s' % \ |
| (amqp_type, return_amqp_type)) |
| self.assertEqual(pass_text, ['pass'], msg='\n %s' % pass_text) |
| else: |
| raise InteropTestError('Receive shim \'%s\':\n%s' % (receive_shim.NAME, receive_obj)) |
| else: |
| raise InteropTestError('Receive shim \'%s\':\n%s' % (receive_shim.NAME, receive_obj)) |
| |
| class TestOptions(qpid_interop_test.qit_common.QitCommonTestOptions): |
| """Command-line arguments used to control the test""" |
| |
| def __init__(self, shim_map, default_timeout=DEFAULT_TEST_TIMEOUT, |
| default_xunit_dir=qpid_interop_test.qit_xunit_log.DEFUALT_XUNIT_LOG_DIR): |
| super(TestOptions, self).__init__('Qpid-interop AMQP client interoparability test suite for AMQP complex types', |
| shim_map, default_timeout, 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(AmqpComplexTypes.default_type_list)) |
| type_group.add_argument('--exclude-type', action='append', metavar='AMQP-TYPE', |
| help='Name of AMQP type to exclude. Supported types: see "include-type" above') |
| subtype_group = self._parser.add_mutually_exclusive_group() |
| subtype_group.add_argument('--include-subtype', action='append', metavar='AMQP-SUBTYPE', |
| help='Name of AMQP subtype to include. Supported types:\n%s' % |
| sorted(AmqpComplexTypes.default_subtype_list)) |
| subtype_group.add_argument('--exclude-subtype', action='append', metavar='AMQP-SUBTYPE', |
| help='Name of AMQP subtype to exclude. Supported types: see "include-subtype" above') |
| |
| |
| class AmqpComplexTypesTest(qpid_interop_test.qit_common.QitTest): |
| """Top-level test for AMQP complex types""" |
| |
| TEST_NAME = 'amqp_complex_types_test' |
| |
| def __init__(self): |
| super(AmqpComplexTypesTest, self).__init__(TestOptions, AmqpComplexTypes) |
| |
| def _generate_tests(self): |
| """Generate tests dynamically""" |
| self.test_suite = unittest.TestSuite() |
| # Create test classes dynamically |
| for amqp_type in self.types.get_type_list(): |
| if self.args.exclude_type is None or amqp_type not in self.args.exclude_type: |
| for amqp_subtype in self.types.get_subtype_list(): |
| if amqp_type != 'array' or amqp_subtype != '*': # array type does not support mixed types (*) |
| test_case_class = self.create_testcase_class(amqp_type, amqp_subtype, |
| itertools.product(self.shim_map.values(), repeat=2), |
| int(self.args.timeout)) |
| self.test_suite.addTest(unittest.makeSuite(test_case_class)) |
| |
| def create_testcase_class(self, amqp_type, amqp_subtype, shim_product, timeout): |
| """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, timeout): |
| """Function which creates a new test method in class cls""" |
| |
| @unittest.skipIf(self.types.skip_test(amqp_subtype, self.broker), |
| self.types.skip_test_message(amqp_subtype, self.broker)) |
| @unittest.skipIf(self.types.skip_client_test(amqp_subtype, send_shim.NAME), |
| self.types.skip_client_test_message(amqp_subtype, send_shim.NAME, u'SENDER')) |
| @unittest.skipIf(self.types.skip_client_test(amqp_subtype, receive_shim.NAME), |
| self.types.skip_client_test_message(amqp_subtype, receive_shim.NAME, u'RECEIVER')) |
| def inner_test_method(self): |
| self.run_test(self.sender_addr, |
| self.receiver_addr, |
| self.amqp_type, |
| amqp_subtype, |
| send_shim, |
| receive_shim, |
| timeout) |
| |
| inner_test_method.__name__ = 'test_%s_%s_%s->%s' % (amqp_type, amqp_subtype, 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, (AmqpComplexTypeTestCase,), class_dict) |
| for send_shim, receive_shim in shim_product: |
| add_test_method(new_class, send_shim, receive_shim, timeout) |
| return new_class |
| |
| |
| #--- Main program start --- |
| |
| if __name__ == '__main__': |
| try: |
| AMQP_COMPLEX_TYPES_TEST = AmqpComplexTypesTest() |
| AMQP_COMPLEX_TYPES_TEST.run_test() |
| AMQP_COMPLEX_TYPES_TEST.write_logs() |
| if not AMQP_COMPLEX_TYPES_TEST.get_result(): |
| sys.exit(1) # Errors or failures present |
| except InteropTestError as err: |
| print(err) |
| sys.exit(1) |