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

"""System test library, provides tools for tests that start multiple processes,
with special support for qdrouter processes.

Features:
- Create separate directories for each test.
- Save logs, sub-process output, core files etc.
- Automated clean-up after tests: kill sub-processes etc.
- Tools to manipulate qdrouter configuration files.
- Sundry other tools.
"""

import errno
import fcntl
import json
import logging
import os
import pathlib
import queue as Queue
import random
import re
import shlex
import shutil
import socket
import subprocess
import sys
import time
import unittest
import uuid
from copy import copy
from datetime import datetime
from subprocess import PIPE, STDOUT
from threading import Event
from threading import Thread
from typing import Callable, TextIO, List, Optional, Tuple

import __main__

import proton
import proton.utils
from proton import Delivery
from proton import Message
from proton.handlers import MessagingHandler
from proton.reactor import AtLeastOnce, Container
from proton.reactor import AtMostOnce
from qpid_dispatch.management.client import Node
from qpid_dispatch.management.error import NotFoundStatus

# Optional modules
MISSING_MODULES = []

try:
    import qpidtoollibs
except ImportError as err:
    qpidtoollibs = None         # pylint: disable=invalid-name
    MISSING_MODULES.append(str(err))

try:
    import qpid_messaging as qm
except ImportError as err:
    qm = None  # pylint: disable=invalid-name
    MISSING_MODULES.append(str(err))


def find_exe(program):
    """Find an executable in the system PATH"""
    def is_exe(fpath):
        """True if fpath is executable"""
        return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
    mydir = os.path.split(program)[0]
    if mydir:
        if is_exe(program):
            return program
    else:
        for path in os.environ["PATH"].split(os.pathsep):
            exe_file = os.path.join(path, program)
            if is_exe(exe_file):
                return exe_file
    return None


# The directory where this module lives. Used to locate static configuration files etc.
DIR = os.path.dirname(__file__)


def _check_requirements():
    """If requirements are missing, return a message, else return empty string."""
    missing = MISSING_MODULES
    required_exes = ['qdrouterd']
    missing += ["No exectuable %s" % e for e in required_exes if not find_exe(e)]

    if missing:
        return "%s: %s" % (__name__, ", ".join(missing))


MISSING_REQUIREMENTS = _check_requirements()


def retry_delay(deadline, delay, max_delay):
    """For internal use in retry. Sleep as required
    and return the new delay or None if retry should time out"""
    remaining = deadline - time.time()
    if remaining <= 0:
        return None
    time.sleep(min(delay, remaining))
    return min(delay * 2, max_delay)


# Valgrind significantly slows down the response time of the router, so use a
# long default timeout
TIMEOUT = float(os.environ.get("QPID_SYSTEM_TEST_TIMEOUT", 60))


def retry(function: Callable[[], bool], timeout: float = TIMEOUT, delay: float = .001, max_delay: float = 1):
    """Call function until it returns a true value or timeout expires.
    Double the delay for each retry up to max_delay.
    Returns what function returns or None if timeout expires.
    """
    deadline = time.time() + timeout
    while True:
        ret = function()
        if ret:
            return ret
        else:
            delay = retry_delay(deadline, delay, max_delay)
            if delay is None:
                return None


def retry_exception(function, timeout=TIMEOUT, delay=.001, max_delay=1, exception_test=None):
    """Call function until it returns without exception or timeout expires.
    Double the delay for each retry up to max_delay.
    Calls exception_test with any exception raised by function, exception_test
    may itself raise an exception to terminate the retry.
    Returns what function returns if it succeeds before timeout.
    Raises last exception raised by function on timeout.
    """
    deadline = time.time() + timeout
    while True:
        try:
            return function()
        except Exception as e:  # pylint: disable=broad-except
            if exception_test:
                exception_test(e)
            delay = retry_delay(deadline, delay, max_delay)
            if delay is None:
                raise


def get_local_host_socket(protocol_family='IPv4'):
    if protocol_family == 'IPv4':
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        host = '127.0.0.1'
    elif protocol_family == 'IPv6':
        s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
        host = '::1'

    return s, host


def port_refuses_connection(port, protocol_family='IPv4'):
    """Return true if connecting to host:port gives 'connection refused'."""
    s, host = get_local_host_socket(protocol_family)
    try:
        s.connect((host, port))
    except OSError as e:
        return e.errno == errno.ECONNREFUSED
    finally:
        s.close()

    return False


def port_permits_binding(port, protocol_family='IPv4'):
    """Return true if binding to the port succeeds."""
    s, _ = get_local_host_socket(protocol_family)
    host = ""
    try:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # so that followup binders are not blocked
        s.bind((host, port))
    except OSError:
        return False
    finally:
        s.close()

    return True


def port_available(port, protocol_family='IPv4'):
    """Return true if a new server will be able to bind to the port."""
    return port_refuses_connection(port, protocol_family) and port_permits_binding(port, protocol_family)


def wait_port(port, protocol_family='IPv4', **retry_kwargs):
    """Wait up to timeout for port (on host) to be connectable.
    Takes same keyword arguments as retry to control the timeout"""
    def check(e):
        """Only retry on connection refused"""
        if not isinstance(e, socket.error) or not e.errno == errno.ECONNREFUSED:
            raise

    host = None

    def connect():
        # macOS gives EINVAL for all connection attempts after a ECONNREFUSED
        # man 3 connect: "If connect() fails, the state of the socket is unspecified. [...]"
        s, host = get_local_host_socket(protocol_family)
        try:
            s.connect((host, port))
        finally:
            s.close()

    try:
        retry_exception(connect, exception_test=check, **retry_kwargs)
    except Exception as e:
        raise Exception("wait_port timeout on host %s port %s: %s" % (host, port, e))


def wait_ports(ports, **retry_kwargs):
    """Wait up to timeout for all ports (on host) to be connectable.
    Takes same keyword arguments as retry to control the timeout"""
    for port, protocol_family in ports.items():
        wait_port(port=port, protocol_family=protocol_family, **retry_kwargs)


def message(**properties):
    """Convenience to create a proton.Message with properties set"""
    m = Message()
    for name, value in properties.items():
        getattr(m, name)        # Raise exception if not a valid message attribute.
        setattr(m, name, value)
    return m


def skip_test_in_ci(environment_var):
    env_var = os.environ.get(environment_var)
    if env_var is not None:
        if env_var.lower() in ['true', '1', 't', 'y', 'yes']:
            return True
    return False


class Process(subprocess.Popen):
    """
    Popen that can be torn down at the end of a TestCase and stores its output.
    """

    # Expected states of a Process at teardown
    RUNNING = -1                # Still running
    EXIT_OK = 0                 # Exit status 0
    EXIT_FAIL = 1               # Exit status 1

    unique_id = 0

    @classmethod
    def unique(cls, name):
        cls.unique_id += 1
        return "%s-%s" % (name, cls.unique_id)

    def __init__(self, args, name=None, expect=EXIT_OK, **kwargs):
        """
        Takes same arguments as subprocess.Popen. Some additional/special args:
        @param expect: Raise error if process status not as expected at end of test:
            L{RUNNING} - expect still running.
            L{EXIT_OK} - expect process to have terminated with 0 exit status.
            L{EXIT_FAIL} - expect process to have terminated with exit status 1.
            integer    - expected return code
        @keyword stdout: Defaults to the file name+".out"
        @keyword stderr: Defaults to be the same as stdout
        """
        self.name = name or os.path.basename(args[0])
        self.args = args
        self.expect = expect
        self.outdir = os.getcwd()
        self.outfile = os.path.abspath(self.unique(self.name))
        self.torndown = False
        with open(self.outfile + '.out', 'w') as out:
            kwargs.setdefault('stdout', out)
            kwargs.setdefault('stderr', subprocess.STDOUT)
            try:
                super(Process, self).__init__(args, **kwargs)
                with open(self.outfile + '.cmd', 'w') as f:
                    f.write("%s\npid=%s\n" % (' '.join(args), self.pid))
            except Exception as e:
                raise Exception("subprocess.Popen(%s, %s) failed: %s: %s" %
                                (args, kwargs, type(e).__name__, e))

    def assert_running(self):
        """Assert that the process is still running"""
        assert self.poll() is None, "%s: exited" % ' '.join(self.args)

    def teardown(self):
        """Check process status and stop the process if necessary"""
        if self.torndown:
            return
        self.torndown = True

        def error(msg):
            with open(self.outfile + '.out') as f:
                raise RuntimeError("Process %s error: %s\n%s\n%s\n>>>>\n%s<<<<" % (
                    self.pid, msg, ' '.join(self.args),
                    self.outfile + '.cmd', f.read()))

        status = self.poll()
        if status is None:      # Still running
            self.terminate()
            if self.expect is not None and self.expect != Process.RUNNING:
                error("still running")
            self.expect = 0     # Expect clean exit after terminate
            status = self.wait()
        if self.expect is not None and self.expect != status:
            error("exit code %s, expected %s" % (status, self.expect))


class Config:
    """Base class for configuration objects that provide a convenient
    way to create content for configuration files."""

    def write(self, name, suffix=".conf"):
        """Write the config object to file name.suffix. Returns name.suffix."""
        name = name + suffix
        with open(name, 'w') as f:
            f.write(str(self))
        return name


class Qdrouterd(Process):
    """Run a Qpid Dispatch Router Daemon"""

    class Config(list, Config):  # type: ignore[misc]  # Cannot resolve name "Config" (possible cyclic definition)  # mypy#10958
        """
        A router configuration.

        The Config class is a list of tuples in the following format:

        [ ('section-name', {attribute-map}), ...]

        where attribute-map is a dictionary of key+value pairs.  Key is an
        attribute name (string), value can be any of [scalar | string | dict]

        When written to a configuration file to be loaded by the router:
        o) there is no ":' between the section-name and the opening brace
        o) attribute keys are separated by a ":" from their values
        o) attribute values that are scalar or string follow the ":" on the
        same line.
        o) attribute values do not have trailing commas
        o) The section-name and attribute keywords are written
        without enclosing quotes
        o) string type attribute values are not enclosed in quotes
        o) attribute values of type dict are written in their JSON representation.

        Fills in some default values automatically, see Qdrouterd.DEFAULTS
        """

        DEFAULTS = {
            'listener': {'host': '0.0.0.0', 'saslMechanisms': 'ANONYMOUS', 'idleTimeoutSeconds': '120',
                         'authenticatePeer': 'no', 'role': 'normal'},
            'connector': {'host': '127.0.0.1', 'saslMechanisms': 'ANONYMOUS', 'idleTimeoutSeconds': '120'},
            'router': {'mode': 'standalone', 'id': 'QDR'}
        }

        def sections(self, name):
            """Return list of sections named name"""
            return [p for n, p in self if n == name]

        @property
        def router_id(self): return self.sections("router")[0]["id"]

        def defaults(self):
            """Fill in default values in gconfiguration"""
            for name, props in self:
                if name in Qdrouterd.Config.DEFAULTS:
                    for n, p in Qdrouterd.Config.DEFAULTS[name].items():
                        props.setdefault(n, p)

        def __str__(self):
            """Generate config file content. Calls default() first."""
            def tabs(level):
                if level:
                    return "    " * level
                return ""

            def value(item, level):
                if isinstance(item, dict):
                    result = "{\n"
                    result += "".join(["%s%s: %s,\n" % (tabs(level + 1),
                                                        json.dumps(k),
                                                        json.dumps(v))
                                       for k, v in item.items()])
                    result += "%s}" % tabs(level)
                    return result
                return "%s" %  item

            def attributes(e, level):
                assert(isinstance(e, dict))
                # k = attribute name
                # v = string | scalar | dict
                return "".join(["%s%s: %s\n" % (tabs(level),
                                                k,
                                                value(v, level + 1))
                                for k, v in e.items()])

            self.defaults()
            # top level list of tuples ('section-name', dict)
            return "".join(["%s {\n%s}\n" % (n, attributes(p, 1)) for n, p in self])

    def __init__(self, name=None, config=Config(), pyinclude=None, wait=True,
                 perform_teardown=True, cl_args=None, expect=Process.RUNNING):
        """
        @param name: name used for for output files, default to id from config.
        @param config: router configuration
        @keyword wait: wait for router to be ready (call self.wait_ready())
        """
        cl_args = cl_args or []
        self.config = copy(config)
        self.perform_teardown = perform_teardown
        if not name:
            name = self.config.router_id
        assert name
        # setup log and debug dump files
        self.dumpfile = os.path.abspath('%s-qddebug.txt' % name)
        self.config.sections('router')[0]['debugDumpFile'] = self.dumpfile
        default_log = [l for l in config if (l[0] == 'log' and l[1]['module'] == 'DEFAULT')]
        if not default_log:
            self.logfile = "%s.log" % name
            config.append(
                ('log', {'module': 'DEFAULT', 'enable': 'trace+',
                         'includeSource': 'true', 'outputFile': self.logfile}))
        else:
            self.logfile = default_log[0][1].get('outputFile')
        args = ['qdrouterd', '-c', config.write(name)] + cl_args
        env_home = os.environ.get('QPID_DISPATCH_HOME')
        if pyinclude:
            args += ['-I', pyinclude]
        elif env_home:
            args += ['-I', os.path.join(env_home, 'python')]

        # shlex.split parses -ex 'thread apply all' into two parts, not in 4 words as string split does
        args = shlex.split(os.environ.get('QPID_DISPATCH_RUNNER', '')) + args
        super(Qdrouterd, self).__init__(args, name=name, expect=expect)
        self._management = None
        self._wait_ready = False
        if wait:
            self.wait_ready()

    @property
    def management(self):
        """Return a management agent proxy for this router"""
        if not self._management:
            self._management = Node.connect(self.addresses[0], timeout=TIMEOUT)
        return self._management

    def teardown(self):
        if self._management:
            try:
                self._management.close()
            except:
                pass
            self._management = None

        if not self.perform_teardown:
            return

        teardown_exc = None
        try:
            super(Qdrouterd, self).teardown()
        except Exception as exc:
            # re-raise _after_ dumping all the state we can
            teardown_exc = exc

        def check_output_file(filename, description):
            """check router's debug dump file for anything interesting (should be
            empty) and dump it to stderr for perusal by organic lifeforms"""
            try:
                if os.stat(filename).st_size > 0:
                    with open(filename) as f:
                        sys.stderr.write("\nRouter %s %s:\n>>>>\n" %
                                         (self.config.router_id, description))
                        sys.stderr.write(f.read())
                        sys.stderr.write("\n<<<<\n")
                        sys.stderr.flush()
            except OSError:
                # failed to open file.  This can happen when an individual test
                # spawns a temporary router (i.e. not created as part of the
                # TestCase setUpClass method) that gets cleaned up by the test.
                pass

        check_output_file(filename=self.outfile + '.out', description="output file")
        check_output_file(filename=self.dumpfile, description="debug dump file")

        if teardown_exc:
            # teardown failed - possible router crash?
            # dump extra stuff (command line, output, log)

            def tail_file(fname, line_count=50):
                """Tail a file to a list"""
                out = []
                with open(fname) as f:
                    line = f.readline()
                    while line:
                        out.append(line)
                        if len(out) > line_count:
                            out.pop(0)
                        line = f.readline()
                return out

            try:
                for fname in [("output", self.outfile + '.out'),
                              ("command", self.outfile + '.cmd')]:
                    with open(fname[1]) as f:
                        sys.stderr.write("\nRouter %s %s file:\n>>>>\n" %
                                         (self.config.router_id, fname[0]))
                        sys.stderr.write(f.read())
                        sys.stderr.write("\n<<<<\n")

                if self.logfile:
                    sys.stderr.write("\nRouter %s log file tail:\n>>>>\n" %
                                     self.config.router_id)
                    tail = tail_file(os.path.join(self.outdir, self.logfile))
                    for ln in tail:
                        sys.stderr.write("%s" % ln)
                    sys.stderr.write("\n<<<<\n")
                sys.stderr.flush()
            except OSError:
                # ignore file not found in case test never opens these
                pass

            raise teardown_exc

    @property
    def ports_family(self):
        """
        Return a dict of listener ports and the respective port family
        Example -
        { 23456: 'IPv4', 243455: 'IPv6' }
        """
        ports_fam = {}
        for l in self.config.sections('listener'):
            if l.get('protocolFamily'):
                ports_fam[l['port']] = l['protocolFamily']
            else:
                ports_fam[l['port']] = 'IPv4'

        return ports_fam

    @property
    def ports(self):
        """Return list of configured ports for all listeners"""
        return [l['port'] for l in self.config.sections('listener')]

    def _cfg_2_host_port(self, c):
        host = c['host']
        port = c['port']
        protocol_family = c.get('protocolFamily', 'IPv4')
        if protocol_family == 'IPv6':
            return "[%s]:%s" % (host, port)
        elif protocol_family == 'IPv4':
            return "%s:%s" % (host, port)
        raise Exception("Unknown protocol family: %s" % protocol_family)

    @property
    def addresses(self):
        """Return amqp://host:port addresses for all listeners"""
        cfg = self.config.sections('listener')
        return ["amqp://%s" % self._cfg_2_host_port(l) for l in cfg]

    @property
    def connector_addresses(self):
        """Return list of amqp://host:port for all connectors"""
        cfg = self.config.sections('connector')
        return ["amqp://%s" % self._cfg_2_host_port(c) for c in cfg]

    @property
    def hostports(self):
        """Return host:port for all listeners"""
        return [self._cfg_2_host_port(l) for l in self.config.sections('listener')]

    def is_connected(self, port, host='127.0.0.1'):
        """If router has a connection to host:port:identity return the management info.
        Otherwise return None"""
        try:
            ret_val = False
            response = self.management.query(type="org.apache.qpid.dispatch.connection")
            index_host = response.attribute_names.index('host')
            for result in response.results:
                outs = '%s:%s' % (host, port)
                if result[index_host] == outs:
                    ret_val = True
            return ret_val
        except:
            return False

    def wait_address(self, address, subscribers=0, remotes=0, containers=0,
                     count=1, **retry_kwargs):
        """
        Wait for an address to be visible on the router.
        @keyword subscribers: Wait till subscriberCount >= subscribers
        @keyword remotes: Wait till remoteCount >= remotes
        @keyword containers: Wait till containerCount >= remotes
        @keyword count: Wait until >= count matching addresses are found
        @param retry_kwargs: keyword args for L{retry}
        """
        def check():
            # TODO aconway 2014-06-12: this should be a request by name, not a query.
            # Need to rationalize addresses in management attributes.
            # endswith check is because of M0/L/R prefixes
            addrs = self.management.query(
                type='org.apache.qpid.dispatch.router.address',
                attribute_names=['name', 'subscriberCount', 'remoteCount', 'containerCount']).get_entities()

            addrs = [a for a in addrs if a['name'].endswith(address)]

            return (len(addrs) >= count
                    and addrs[0]['subscriberCount'] >= subscribers
                    and addrs[0]['remoteCount'] >= remotes
                    and addrs[0]['containerCount'] >= containers)
        assert retry(check, **retry_kwargs)

    def wait_address_unsubscribed(self, address, **retry_kwargs):
        """
        Block until address has no subscribers
        """
        a_type = 'org.apache.qpid.dispatch.router.address'

        def check():
            addrs = self.management.query(a_type).get_dicts()
            rc = [a for a in addrs if a['name'].endswith(address)]
            count = 0
            for a in rc:
                count += a['subscriberCount']
                count += a['remoteCount']
                count += a['containerCount']

            return count == 0
        assert retry(check, **retry_kwargs)

    def get_host(self, protocol_family):
        if protocol_family == 'IPv4':
            return '127.0.0.1'
        elif protocol_family == 'IPv6':
            return '::1'
        else:
            return '127.0.0.1'

    def wait_ports(self, **retry_kwargs):
        wait_ports(self.ports_family, **retry_kwargs)

    def wait_connectors(self, **retry_kwargs):
        """
        Wait for all connectors to be connected
        @param retry_kwargs: keyword args for L{retry}
        """
        for c in self.config.sections('connector'):
            assert retry(lambda: self.is_connected(port=c['port'], host=self.get_host(c.get('protocolFamily'))),
                         **retry_kwargs), "Port not connected %s" % c['port']

    def wait_startup_message(self, **retry_kwargs):
        """Wait for router startup message to be printed into logfile

        This ensures that the router installs its signal handlers, avoiding
        a router failure with return code -15 upon premature SIGTERM (DISPATCH-1689)

        e.g. 2022-03-03 19:08:13.608655 +0100 SERVER (notice) Operational, 4 Threads Running (process ID 2190110)
        """
        def _is_startup_line_present(f: TextIO) -> bool:
            for line in f:
                m = re.search(r'SERVER \(notice\) Operational, (\d+) Threads Running \(process ID (\d+)\)', line)
                if m:
                    return True
            return False

        logfile_path = self.logfile_path
        # system_tests_log_level_update filters SERVER module logs to a separate file
        server_log = [l for l in self.config if (l[0] == 'log' and l[1]['module'] == 'SERVER')]
        if server_log:
            logfile_path = os.path.join(self.outdir, server_log[0][1].get('outputFile'))

        assert retry(lambda: pathlib.Path(logfile_path).is_file(), **retry_kwargs), \
            f"Router logfile {logfile_path} does not exist or is not a file"
        with open(logfile_path, 'rt') as router_log:
            assert retry(lambda: _is_startup_line_present(router_log), **retry_kwargs),\
                "Router startup line not present in router log"

    def wait_ready(self, **retry_kwargs):
        """Wait for ports and connectors to be ready"""
        if not self._wait_ready:
            self._wait_ready = True
            self.wait_ports(**retry_kwargs)
            self.wait_connectors(**retry_kwargs)
            self.wait_startup_message(**retry_kwargs)
        return self

    def is_router_connected(self, router_id, **retry_kwargs):
        node = None
        try:
            self.management.read(identity="router.node/%s" % router_id)
            # TODO aconway 2015-01-29: The above check should be enough, we
            # should not advertise a remote router in management till it is fully
            # connected. However we still get a race where the router is not
            # actually ready for traffic. Investigate.
            # Meantime the following actually tests send-thru to the router.
            node = Node.connect(self.addresses[0], router_id, timeout=1)
            return retry_exception(lambda: node.query('org.apache.qpid.dispatch.router'))
        except (proton.ConnectionException, NotFoundStatus, proton.utils.LinkDetached):
            # proton.ConnectionException: the router is not yet accepting connections
            # NotFoundStatus: the queried router is not yet connected
            # TODO(DISPATCH-2119) proton.utils.LinkDetached: should be removed, currently needed for DISPATCH-2033
            return False
        finally:
            if node:
                node.close()

    def wait_router_connected(self, router_id, **retry_kwargs):
        retry(lambda: self.is_router_connected(router_id), **retry_kwargs)

    @property
    def logfile_path(self):
        """Path to a DEFAULT logfile"""
        return os.path.join(self.outdir, self.logfile)


class Tester:
    """Tools for use by TestCase
- Create a directory for the test.
- Utilities to create processes and servers, manage ports etc.
- Clean up processes on teardown"""

    # Top level directory above any Tester directories.
    # CMake-generated configuration may be found here.
    top_dir = os.getcwd()

    # The root directory for Tester directories, under top_dir
    root_dir = os.path.abspath(__name__ + '.dir')

    # Minimum and maximum port number for free port searches
    port_range = (20000, 30000)

    def __init__(self, id):
        """
        @param id: module.class.method or False if no directory should be created
        """
        self.directory = os.path.join(self.root_dir, *id.split('.')) if id else None
        self.cleanup_list = []

        self.port_file = pathlib.Path(self.top_dir, "next_port.lock").open("a+t")
        self.cleanup(self.port_file)

    def rmtree(self):
        """Remove old test class results directory"""
        if self.directory:
            shutil.rmtree(os.path.dirname(self.directory), ignore_errors=True)

    def setup(self):
        """Called from test setup and class setup."""
        if self.directory:
            os.makedirs(self.directory)
            os.chdir(self.directory)

    def _next_port(self) -> int:
        """Reads and increments value stored in self.port_file, under an exclusive file lock.

        When a lock cannot be acquired immediately, fcntl.lockf blocks.

        Failure possibilities:
            File locks may not work correctly on network filesystems. We still should be no worse off than we were.

            This method always unlocks the lock file, so it should not ever deadlock other tests running in parallel.
            Even if that happened, the lock is unlocked by the OS when the file is closed, which happens automatically
            when the process that opened and locked it ends.

            Invalid content in the self.port_file will break this method. Manual intervention is then required.
        """
        try:
            fcntl.flock(self.port_file, fcntl.LOCK_EX)

            # read old value
            self.port_file.seek(0, os.SEEK_END)
            if self.port_file.tell() != 0:
                self.port_file.seek(0)
                port = int(self.port_file.read())
            else:
                # file is empty
                port = random.randint(self.port_range[0], self.port_range[1])

            next_port = port + 1
            if next_port >= self.port_range[1]:
                next_port = self.port_range[0]

            # write new value
            self.port_file.seek(0)
            self.port_file.truncate(0)
            self.port_file.write(str(next_port))
            self.port_file.flush()

            return port
        finally:
            fcntl.flock(self.port_file, fcntl.LOCK_UN)

    def teardown(self):
        """Clean up (tear-down, stop or close) objects recorded via cleanup()"""
        self.cleanup_list.reverse()
        errors = []
        for obj in self.cleanup_list:
            try:
                for method in ["teardown", "tearDown", "stop", "close"]:
                    cleanup = getattr(obj, method, None)
                    if cleanup:
                        cleanup()
                        break
            except Exception as exc:
                errors.append(exc)
        if errors:
            raise RuntimeError("Errors during teardown: \n\n%s" % "\n\n".join([str(e) for e in errors]))

    def cleanup(self, x):
        """Record object x for clean-up during tear-down.
        x should have on of the methods teardown, tearDown, stop or close"""
        self.cleanup_list.append(x)
        return x

    def popen(self, *args, **kwargs):
        """Start a Process that will be cleaned up on teardown"""
        return self.cleanup(Process(*args, **kwargs))

    def qdrouterd(self, *args, **kwargs):
        """Return a Qdrouterd that will be cleaned up on teardown"""
        return self.cleanup(Qdrouterd(*args, **kwargs))

    port_range = (20000, 30000)
    next_port = random.randint(port_range[0], port_range[1])

    def get_port(self, protocol_family='IPv4'):
        """Get an unused port"""
        p = self._next_port()
        start = p
        while not port_available(p, protocol_family):
            p = self._next_port()
            if p == start:
                raise Exception("No available ports in range %s", self.port_range)
        return p


class TestCase(unittest.TestCase, Tester):  # pylint: disable=too-many-public-methods
    """A TestCase that sets up its own working directory and is also a Tester."""

    tester: Tester

    def __init__(self, test_method):
        unittest.TestCase.__init__(self, test_method)
        Tester.__init__(self, self.id())

    @classmethod
    def setUpClass(cls):
        cls.maxDiff = None
        cls.tester = Tester('.'.join([cls.__module__, cls.__name__, 'setUpClass']))
        cls.tester.rmtree()
        cls.tester.setup()

    @classmethod
    def tearDownClass(cls):
        if hasattr(cls, 'tester'):
            cls.tester.teardown()
            del cls.tester

    def setUp(self):
        Tester.setup(self)

    def tearDown(self):
        Tester.teardown(self)

    def assert_fair(self, seq):
        avg = sum(seq) / len(seq)
        for i in seq:
            assert i > avg / 2, "Work not fairly distributed: %s" % seq

    if not hasattr(unittest.TestCase, 'assertRegex'):
        def assertRegex(self, text, regexp, msg=None):
            assert re.search(regexp, text), msg or "Can't find %r in '%s'" % (regexp, text)

    if not hasattr(unittest.TestCase, 'assertNotRegex'):
        def assertNotRegex(self, text, regexp, msg=None):
            assert not re.search(regexp, text), msg or "Found %r in '%s'" % (regexp, text)


def main_module():
    """
    Return the module name of the __main__ module - i.e. the filename with the
    path and .py extension stripped. Useful to run the tests in the current file but
    using the proper module prefix instead of '__main__', as follows:
        if __name__ == '__main__':
            unittest.main(module=main_module())
    """
    return os.path.splitext(os.path.basename(__main__.__file__))[0]


class AsyncTestReceiver(MessagingHandler):
    """
    A simple receiver that runs in the background and queues any received
    messages.  Messages can be retrieved from this thread via the queue member.
    :param wait: block the constructor until the link has been fully
                 established.
    :param recover_link: restart on remote link detach
    """
    Empty = Queue.Empty

    class MyQueue(Queue.Queue):
        def __init__(self, receiver):
            self._async_receiver = receiver
            super(AsyncTestReceiver.MyQueue, self).__init__()

        def get(self, timeout=TIMEOUT):
            self._async_receiver.num_queue_gets += 1
            msg = super(AsyncTestReceiver.MyQueue, self).get(timeout=timeout)
            self._async_receiver._logger.log("message %d get"
                                             % self._async_receiver.num_queue_gets)
            return msg

        def put(self, msg):
            self._async_receiver.num_queue_puts += 1
            super(AsyncTestReceiver.MyQueue, self).put(msg)
            self._async_receiver._logger.log("message %d put"
                                             % self._async_receiver.num_queue_puts)

    def __init__(self, address, source, conn_args=None, container_id=None,
                 wait=True, recover_link=False, msg_args=None, print_to_console=False):
        if msg_args is None:
            msg_args = {}
        super(AsyncTestReceiver, self).__init__(**msg_args)
        self.address = address
        self.source = source
        self.conn_args = conn_args
        self.queue = AsyncTestReceiver.MyQueue(self)
        self._conn = None
        self._container = Container(self)
        cid = container_id or "ATR-%s:%s" % (source, uuid.uuid4())
        self._container.container_id = cid
        self._ready = Event()
        self._recover_link = recover_link
        self._recover_count = 0
        self._stop_thread = False
        self._thread = Thread(target=self._main)
        self._logger = Logger(title="AsyncTestReceiver %s" % cid, print_to_console=print_to_console)
        self._thread.daemon = True
        self._thread.start()
        self.num_queue_puts = 0
        self.num_queue_gets = 0
        if wait and self._ready.wait(timeout=TIMEOUT) is False:
            raise Exception("Timed out waiting for receiver start")
        self.queue_stats = "self.num_queue_puts=%d, self.num_queue_gets=%d"

    def get_queue_stats(self):
        return self.queue_stats % (self.num_queue_puts, self.num_queue_gets)

    def _main(self):
        self._container.timeout = 0.5
        self._container.start()
        self._logger.log("AsyncTestReceiver Starting reactor")
        while self._container.process():
            if self._stop_thread:
                if self._conn:
                    self._conn.close()
                    self._conn = None
        self._logger.log("AsyncTestReceiver reactor thread done")

    def on_connection_error(self, event):
        self._logger.log("AsyncTestReceiver on_connection_error=%s" % event.connection.remote_condition.description)

    def on_link_error(self, event):
        self._logger.log("AsyncTestReceiver on_link_error=%s" % event.link.remote_condition.description)

    def stop(self, timeout=TIMEOUT):
        self._stop_thread = True
        self._container.wakeup()
        self._thread.join(timeout=TIMEOUT)
        self._logger.log("thread done")
        if self._thread.is_alive():
            raise Exception("AsyncTestReceiver did not exit")
        del self._conn
        del self._container

    def on_start(self, event):
        kwargs = {'url': self.address}
        if self.conn_args:
            kwargs.update(self.conn_args)
        self._conn = event.container.connect(**kwargs)

    def on_connection_opened(self, event):
        self._logger.log("Connection opened")
        kwargs = {'source': self.source}
        event.container.create_receiver(event.connection, **kwargs)

    def on_link_opened(self, event):
        self._logger.log("link opened")
        self._ready.set()

    def on_link_closing(self, event):
        self._logger.log("link closing")
        event.link.close()
        if self._recover_link and not self._stop_thread:
            # lesson learned: the generated link name will be the same as the
            # old link (which is bad) so we specify a new one
            self._recover_count += 1
            kwargs = {'source': self.source,
                      'name': "%s:%s" % (event.link.name, self._recover_count)}
            rcv = event.container.create_receiver(event.connection,
                                                  **kwargs)

    def on_message(self, event):
        self.queue.put(event.message)

    def on_disconnected(self, event):
        # if remote terminates the connection kill the thread else it will spin
        # on the cpu
        self._logger.log("Disconnected")
        if self._conn:
            self._conn.close()
            self._conn = None

    def dump_log(self):
        self._logger.dump()


class AsyncTestSender(MessagingHandler):
    """
    A simple sender that runs in the background and sends 'count' messages to a
    given target.
    """
    class TestSenderException(Exception):
        def __init__(self, error=None):
            super(AsyncTestSender.TestSenderException, self).__init__(error)

    def __init__(self, address, target, count=1, message=None,
                 container_id=None, presettle=False, print_to_console=False):
        super(AsyncTestSender, self).__init__(auto_accept=False,
                                              auto_settle=False)
        self.address = address
        self.target = target
        self.total = count
        self.presettle = presettle
        self.accepted = 0
        self.released = 0
        self.modified = 0
        self.rejected = 0
        self.sent = 0
        self.error = None
        self.link_stats = None
        self._conn = None
        self._sender = None
        self._message = message or Message(body="test")
        self._container = Container(self)
        cid = container_id or "ATS-%s:%s" % (target, uuid.uuid4())
        self._container.container_id = cid
        self._link_name = "%s-%s" % (cid, "tx")
        self._thread = Thread(target=self._main)
        self._thread.daemon = True
        self._logger = Logger(title="AsyncTestSender %s" % cid, print_to_console=print_to_console)
        self._thread.start()
        self.msg_stats = "self.sent=%d, self.accepted=%d, self.released=%d, self.modified=%d, self.rejected=%d"

    def _main(self):
        self._container.timeout = 0.5
        self._container.start()
        self._logger.log("AsyncTestSender Starting reactor")
        while self._container.process():
            self._check_if_done()
        self._logger.log("AsyncTestSender reactor thread done")

    def get_msg_stats(self):
        return self.msg_stats % (self.sent, self.accepted, self.released, self.modified, self.rejected)

    def wait(self):
        # don't stop it - wait until everything is sent
        self._logger.log("AsyncTestSender wait: about to join thread")
        self._thread.join(timeout=TIMEOUT)
        self._logger.log("AsyncTestSender wait: thread done")
        assert not self._thread.is_alive(), "sender did not complete"
        if self.error:
            raise AsyncTestSender.TestSenderException(self.error)
        del self._sender
        del self._conn
        del self._container
        self._logger.log("AsyncTestSender wait: no errors in wait")

    def on_start(self, event):
        self._conn = self._container.connect(self.address)

    def on_connection_opened(self, event):
        self._logger.log("Connection opened")
        option = AtMostOnce if self.presettle else AtLeastOnce
        self._sender = self._container.create_sender(self._conn,
                                                     target=self.target,
                                                     options=option(),
                                                     name=self._link_name)

    def on_sendable(self, event):
        if self.sent < self.total:
            self._sender.send(self._message)
            self.sent += 1
            self._logger.log("message %d sent" % self.sent)

    def _check_if_done(self):
        done = (self.sent == self.total
                and (self.presettle
                     or (self.accepted + self.released + self.modified
                         + self.rejected == self.sent)))
        if done and self._conn:
            self.link_stats = get_link_info(self._link_name,
                                            self.address)
            self._conn.close()
            self._conn = None
            self._logger.log("Connection closed")

    def on_accepted(self, event):
        self.accepted += 1
        event.delivery.settle()
        self._logger.log("message %d accepted" % self.accepted)

    def on_released(self, event):
        # for some reason Proton 'helpfully' calls on_released even though the
        # delivery state is actually MODIFIED
        if event.delivery.remote_state == Delivery.MODIFIED:
            return self.on_modified(event)
        self.released += 1
        event.delivery.settle()
        self._logger.log("message %d released" % self.released)

    def on_modified(self, event):
        self.modified += 1
        event.delivery.settle()
        self._logger.log("message %d modified" % self.modified)

    def on_rejected(self, event):
        self.rejected += 1
        event.delivery.settle()
        self._logger.log("message %d rejected" % self.rejected)

    def on_link_error(self, event):
        self.error = "link error:%s" % str(event.link.remote_condition)
        self._logger.log(self.error)
        if self._conn:
            self._conn.close()
            self._conn = None

    def on_disconnected(self, event):
        # if remote terminates the connection kill the thread else it will spin
        # on the cpu
        self.error = "connection to remote dropped"
        self._logger.log(self.error)
        if self._conn:
            self._conn.close()
            self._conn = None

    def dump_log(self):
        self._logger.dump()


class QdManager:
    """
    A means to invoke qdmanage during a testcase
    """

    def __init__(self, address: Optional[str] = None,
                 timeout: Optional[float] = TIMEOUT,
                 router_id: Optional[str] = None,
                 edge_router_id: Optional[str] = None) -> None:
        # 'tester' - can be 'self' when called in a test,
        # or an instance any class derived from Process (like Qdrouterd)
        self._timeout = timeout
        self._address = address
        self.router_id = router_id
        self.edge_router_id = edge_router_id
        self.router: List[str] = []
        if self.router_id:
            self.router = self.router + ['--router', self.router_id]
        elif self.edge_router_id:
            self.router = self.router + ['--edge-router', self.edge_router_id]

    def __call__(self, cmd: str,
                 address: Optional[str] = None,
                 input: Optional[str] = None,
                 timeout: Optional[float] = None) -> str:
        addr = address or self._address
        assert addr, "address missing"
        with subprocess.Popen(['qdmanage'] + cmd.split(' ') + self.router
                              + ['--bus', addr, '--indent=-1', '--timeout',
                                 str(timeout or self._timeout)], stdin=PIPE,
                              stdout=PIPE, stderr=STDOUT,
                              universal_newlines=True) as p:
            rc = p.communicate(input)
            if p.returncode != 0:
                raise Exception("%s %s" % rc)
            return rc[0]

    def create(self, long_type, kwargs):
        cmd = "CREATE --type=%s" % long_type
        for k, v in kwargs.items():
            cmd += " %s=%s" % (k, v)
        return json.loads(self(cmd))

    def update(self, long_type, kwargs, name=None, identity=None):
        cmd = 'UPDATE --type=%s' % long_type
        if identity is not None:
            cmd += " --identity=%s" % identity
        elif name is not None:
            cmd += " --name=%s" % name
        for k, v in kwargs.items():
            cmd += " %s=%s" % (k, v)
        return json.loads(self(cmd))

    def delete(self, long_type, name=None, identity=None):
        cmd = 'DELETE --type=%s' %  long_type
        if identity is not None:
            cmd += " --identity=%s" % identity
        elif name is not None:
            cmd += " --name=%s" % name
        else:
            assert False, "name or identity not supplied!"
        self(cmd)

    def query(self, long_type):
        return json.loads(self('QUERY --type=%s' % long_type))

    def get_log(self, limit=None):
        cmd = 'GET-LOG'
        if (limit):
            cmd += " limit=%s" % limit
        return json.loads(self(cmd))


class MgmtMsgProxy:
    """
    Utility for creating and inspecting management messages
    """
    class _Response:
        def __init__(self, status_code, status_description, body):
            self.status_code        = status_code
            self.status_description = status_description
            if body.__class__ == dict and len(body.keys()) == 2 and 'attributeNames' in body.keys() and 'results' in body.keys():
                results = []
                names   = body['attributeNames']
                for result in body['results']:
                    result_map = {}
                    for i in range(len(names)):
                        result_map[names[i]] = result[i]
                    results.append(MgmtMsgProxy._Response(status_code, status_description, result_map))
                self.attrs = {'results': results}
            else:
                self.attrs = body

        def __getattr__(self, key):
            return self.attrs[key]

    def __init__(self, reply_addr):
        self.reply_addr = reply_addr

    def response(self, msg):
        ap = msg.properties
        return self._Response(ap['statusCode'], ap['statusDescription'], msg.body)

    def query_router(self):
        ap = {'operation': 'QUERY', 'type': 'org.apache.qpid.dispatch.router'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def query_connections(self):
        ap = {'operation': 'QUERY', 'type': 'org.apache.qpid.dispatch.connection'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def query_links(self):
        ap = {'operation': 'QUERY', 'type': 'org.apache.qpid.dispatch.router.link'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def query_link_routes(self):
        ap = {'operation': 'QUERY',
              'type': 'org.apache.qpid.dispatch.router.config.linkRoute'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def query_addresses(self):
        ap = {'operation': 'QUERY',
              'type': 'org.apache.qpid.dispatch.router.address'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def create_link_route(self, name, kwargs):
        ap = {'operation': 'CREATE',
              'type': 'org.apache.qpid.dispatch.router.config.linkRoute',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr,
                       body=kwargs)

    def delete_link_route(self, name):
        ap = {'operation': 'DELETE',
              'type': 'org.apache.qpid.dispatch.router.config.linkRoute',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr)

    def create_connector(self, name, **kwargs):
        ap = {'operation': 'CREATE',
              'type': 'org.apache.qpid.dispatch.connector',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr,
                       body=kwargs)

    def delete_connector(self, name):
        ap = {'operation': 'DELETE',
              'type': 'org.apache.qpid.dispatch.connector',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr)

    def query_conn_link_routes(self):
        ap = {'operation': 'QUERY',
              'type': 'org.apache.qpid.dispatch.router.connection.linkRoute'}
        return Message(properties=ap, reply_to=self.reply_addr)

    def create_conn_link_route(self, name, kwargs):
        ap = {'operation': 'CREATE',
              'type': 'org.apache.qpid.dispatch.router.connection.linkRoute',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr,
                       body=kwargs)

    def delete_conn_link_route(self, name):
        ap = {'operation': 'DELETE',
              'type': 'org.apache.qpid.dispatch.router.connection.linkRoute',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr)

    def read_conn_link_route(self, name):
        ap = {'operation': 'READ',
              'type': 'org.apache.qpid.dispatch.router.connection.linkRoute',
              'name': name}
        return Message(properties=ap, reply_to=self.reply_addr)


class TestTimeout:
    """
    A callback object for MessagingHandler class
    parent: A MessagingHandler with a timeout() method
    """
    __test__ = False

    def __init__(self, parent):
        self.parent = parent

    def on_timer_task(self, event):
        self.parent.timeout()


class PollTimeout:
    """
    A callback object for MessagingHandler scheduled timers
    parent: A MessagingHandler with a poll_timeout() method
    """

    def __init__(self, parent):
        self.parent = parent

    def on_timer_task(self, event):
        self.parent.poll_timeout()


def get_link_info(name, address):
    """
    Query the router at address for the status and statistics of the named link
    """
    qdm = QdManager(address=address)
    rc = qdm.query('org.apache.qpid.dispatch.router.link')
    for item in rc:
        if item.get('name') == name:
            return item
    return None


def has_mobile_dest_in_address_table(address, dest):
    qdm = QdManager(address=address)
    rc = qdm.query('org.apache.qpid.dispatch.router.address')
    has_dest = False
    for item in rc:
        if dest in item.get("name"):
            has_dest = True
            break
    return has_dest


def get_inter_router_links(address):
    """
    Return a list of all links with type="inter-router
    :param address:
    """
    inter_router_links = []
    qdm = QdManager(address=address)
    rc = qdm.query('org.apache.qpid.dispatch.router.link')
    for item in rc:
        if item.get("linkType") == "inter-router":
            inter_router_links.append(item)

    return inter_router_links


class Timestamp:
    """
    Time stamps for logging.
    """

    def __init__(self):
        self.ts = datetime.now()

    def __str__(self):
        return self.ts.strftime("%Y-%m-%d %H:%M:%S.%f")


class Logger:
    """
    Record an event log for a self test.
    May print per-event or save events to be printed later.
    Pytest will automatically collect the logs and will dump them for a failed test
    Optional file opened in 'append' mode to which each log line is written.
    """

    def __init__(self,
                 title: str = "Logger",
                 print_to_console: bool = False,
                 save_for_dump: bool = True,
                 python_log_level: Optional[int] = logging.DEBUG,
                 ofilename: Optional[str] = None) -> None:
        self.title = title
        self.print_to_console = print_to_console
        self.save_for_dump = save_for_dump
        self.python_log_level = python_log_level
        self.ofilename = ofilename
        self.logs: List[Tuple[Timestamp, str]] = []

    def log(self, msg):
        ts = Timestamp()
        if self.save_for_dump:
            self.logs.append((ts, msg))
        if self.print_to_console:
            print("%s %s" % (ts, msg))
            sys.stdout.flush()
        if self.python_log_level is not None:
            logging.log(self.python_log_level, f"{ts} {self.title}: {msg}")
        if self.ofilename is not None:
            with open(self.ofilename, 'a') as f_out:
                f_out.write("%s %s\n" % (ts, msg))
                f_out.flush()

    def dump(self):
        print(self)
        sys.stdout.flush()

    def __str__(self):
        lines = [self.title]
        for ts, msg in self.logs:
            lines.append("%s %s" % (ts, msg))
        res = str('\n'.join(lines))
        return res


def curl_available():
    """
    Check if the curl command line tool is present on the system.
    Return a tuple containing the version if found, otherwise
    return false.
    """
    popen_args = ['curl', '--version']
    try:
        process = Process(popen_args,
                          name='curl_check',
                          stdout=PIPE,
                          expect=None,
                          universal_newlines=True)
        out = process.communicate()[0]
        if process.returncode == 0:
            # return curl version as a tuple (major, minor[,fix])
            # expects --version outputs "curl X.Y.Z ..."
            return tuple([int(x) for x in out.split()[1].split('.')])
    except:
        pass
    return False


def run_curl(args, input=None, timeout=TIMEOUT):
    """
    Run the curl command with the given argument list.
    Pass optional input to curls stdin.
    Return tuple of (return code, stdout, stderr)
    """
    popen_args = ['curl'] + args
    if timeout is not None:
        popen_args = popen_args + ["--max-time", str(timeout)]
    stdin_value = PIPE if input is not None else None
    with subprocess.Popen(popen_args, stdin=stdin_value, stdout=PIPE,
                          stderr=PIPE, universal_newlines=True) as p:
        out = p.communicate(input, timeout)
        return p.returncode, out[0], out[1]