bin/qpid-interop-test.in

#! /usr/bin/env python3

#  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.

# NOTE: THIS FILE IS GENERATED, DO NOT EDIT DIRECTLY.

""" Qpid Interop Test entry point """

import os
import subprocess
import sys

TEST_NAME_LIST = ['amqp-types-test',
                  'amqp-complex-types-test',
                  'amqp-large-content-test',
                  'jms-messages-test',
                  'jms-hdrs-props-test']

class TestOptions:
    """ Test options for QIT entry point """
    def __init__(self):
        if len(sys.argv) == 1:
            print('ERROR: No test name.\n')
            TestOptions._usage(sys.argv[0])
            sys.exit(1)
        test_name = sys.argv[1]
        if test_name in ['-h', '--help', 'help']:
            self._usage(sys.argv[0], True)
            sys.exit()
        if test_name not in TEST_NAME_LIST and test_name != 'all':
            print('ERROR: Invalid test name.\n')
            TestOptions._usage(sys.argv[0])
            sys.exit(1)
        self.test_list = TEST_NAME_LIST if test_name == 'all' else [test_name]
        self.test_opts = sys.argv[2:] if len(sys.argv) > 2 else None

    @staticmethod
    def _usage(argv0, help_flag=False):
        prog_name = argv0.split('/')[-1]
        if help_flag:
            print('A test suite which sends AMQP messages between different clients to test interoperability.')
        print('Usage:')
        print(f'  {prog_name} --help:  Print help and exit')
        print(f'  {prog_name} test-name [test-options...]:  Run test <test-name>')
        print(f'      test-name:             Required, one of {TEST_NAME_LIST},')
        print( '                             or "all" to run all tests sequentially')
        print( '      test-options:          Optional arguments for test, or --help to see test-specific options')

if __name__ == "__main__":
    # Set the environment so proton and QIT can be found
    os.environ['PYTHONPATH'] = ('@PROTON_INSTALL_DIR@/lib64/proton/bindings/python:'
                                '@CMAKE_INSTALL_PREFIX@/lib:'
                                '@CMAKE_INSTALL_PREFIX@/libexec/qpid_interop_test/shims/qpid-proton-python')
    os.environ['LD_LIBRARY_PATH'] = '@PROTON_INSTALL_DIR@/lib64:%s'%(os.getenv('LD_LIBRARY_PATH'))
    os.environ['CMAKE_INSTALL_PREFIX'] = '@CMAKE_INSTALL_PREFIX@'

    # Process test options
    TEST_OPTIONS = TestOptions()

    # Run the test(s)
    for this_test_name in TEST_OPTIONS.test_list:
        if len(TEST_OPTIONS.test_list) > 1:
            print(f'=== {this_test_name} ===')
        py_module_name = this_test_name.replace('-', '_')
        subprocess_args = ['python3', '-m', f'qpid_interop_test.{py_module_name}']
        if TEST_OPTIONS.test_opts is not None:
            subprocess_args.extend(TEST_OPTIONS.test_opts)
        try:
            subprocess.run(subprocess_args, check=True)
        except subprocess.CalledProcessError as err:
            cmd = ' '.join(err.cmd)
            print(f'"{cmd}" returned exit-code {err.returncode}')
        if len(TEST_OPTIONS.test_list) > 1:
            print()