class TestConfig:
    def __init__(self):
        self._default_broker_uri = "amqp://guest:guest@localhost:5672//"
        self._environ_broker_uri = os.environ.get("BROKER")

    @property
    def broker_uri(self) -> str:
        if self._environ_broker_uri is not None:
            return self._environ_broker_uri
        else:
            return self._default_broker_uri class TestMessage(Command):
    def __init__(self):
        self._integer_value = 99
        self._float_value = 3.14
        self._string_value = "fubar"
        self._bool_value = True
        super().__init__()

    @property
    def integer_value(self):
        return self._integer_value

    @property
    def float_value(self):
        return self._float_value

    @property
    def string_value(self):
        return self._string_value

    @property
    def bool_value(self):
        return self._bool_value class ArameConsumer(BrightsideConsumer):
    """ Implements reading a message from an RMQ broker. It uses a queue, created by subscribing to a message topic

    """
    RETRY_OPTIONS = {
        'interval_start': 1,
        'interval_step': 1,
        'interval_max': 1,
        'max_retries': 3,
    }

    def __init__(self, connection: Connection, configuration: BrightsideConsumerConfiguration, logger: logging.Logger=None) -> None:
        self._exchange = Exchange(connection.exchange, type=connection.exchange_type, durable=connection.is_durable)
        self._routing_key = configuration.routing_key
        self._amqp_uri = connection.amqp_uri
        self._queue_name = configuration.queue_name
        self._routing_key = configuration.routing_key
        self._prefetch_count = configuration.prefetch_count
        self._is_durable = configuration.is_durable
        self._heartbeat = connection.heartbeat
        self._connect_timeout = connection.connect_timeout
        self._message_factory = ArameMessageFactory()
        self._logger = logger or logging.getLogger(__name__)
        self._conn = None
        consumer_arguments = {}
        if configuration.is_ha is True:
            consumer_arguments = {"x-ha-policy": "all"}

        self._is_long_running_handler = configuration.is_long_runing_handler

        self._queue = Queue(self._queue_name, exchange=self._exchange, routing_key=self._routing_key,
                            durable=self._is_durable, consumer_arguments=consumer_arguments)

        self._msg = None  # Kombu Message
        self._message = None  # Brightside Message

        self._establish_connection(BrokerConnection(hostname=self._amqp_uri, connect_timeout=self._connect_timeout, heartbeat=self._heartbeat))
        self._establish_channel()
        self._establish_consumer()

    def acknowledge(self, message: BrightsideMessage):
        if (self._message is not None) and self._message.id == message.id:
            self._msg.ack()
            self._msg = None

    def _ensure_connection(self):
        # We can get connection aborted before we try to read, so despite ensure()
        # we check the connection here
        if self._conn.connected is not True:
            self._conn = self._conn.clone()
            self._conn.ensure_connection(max_retries=3)
            self._channel = self._conn.channel()
            self._consumer.revive(self._channel)
            self._consumer.consume()

    def _establish_channel(self):
        self._channel = self._conn.channel()

    def _establish_consumer(self):
        self._consumer = Consumer(channel=self._channel, queues=[self._queue], callbacks=[self._read_message])
        self._consumer.qos(prefetch_count=1)
        self._consumer.consume()

    def _establish_connection(self, conn: BrokerConnection) -> None:
        """
        We don't use a pool here. We only have one consumer connection per process, so
        we get no value from a pool, and we want to use a heartbeat to keep the consumer
        collection alive, which does not work with a pool
        :return: the connection to the transport
        """
        try:
            self._logger.debug("Establishing connection.")
            self._conn = conn.ensure_connection(max_retries=3)
            self._logger.debug('Got connection: %s', conn.as_uri())
        except kombu_exceptions.OperationalError as oe:
            self._logger.error("Error connecting to RMQ, could not retry %s", oe)
            # Try to clean up the mess
            if self._conn is not None:
                self._conn.close()
            else:
                conn.close()

    def has_acknowledged(self, message):
        if (self._message is not None) and self._message.id == message.id:
            if self._msg is None:
                return True
            else:
                return False

    def purge(self, timeout: int = 5) -> None:

        def _purge_errors(exc, interval):
            self._logger.error('Purging error: %s, will retry triggering in %s seconds', exc, interval, exc_info=True)

        def _purge_messages(cnsmr: BrightsideConsumer):
            cnsmr.purge()
            self._message = None

        self._ensure_connection()

        ensure_kwargs = self.RETRY_OPTIONS.copy()
        ensure_kwargs['errback'] = _purge_errors
        safe_purge = self._conn.ensure(self._consumer, _purge_messages, **ensure_kwargs)
        safe_purge(self._consumer)

    def _read_message(self, body: str, msg: KombuMessage) -> None:
        self._logger.debug("Monitoring event received at: %s headers: %s payload: %s", datetime.utcnow().isoformat(), msg.headers, body)
        self._msg = msg
        self._message = self._message_factory.create_message(msg)

    def receive(self, timeout: int) -> BrightsideMessage:

        self._message = BrightsideMessage(BrightsideMessageHeader(uuid4(), "", BrightsideMessageType.MT_NONE), BrightsideMessageBody(""))

        def _consume(cnx: BrokerConnection, timesup: int) -> None:
            try:
                cnx.drain_events(timeout=timesup)
            except kombu_exceptions.TimeoutError:
                self._logger.debug("Time out reading from queue %s", self._queue_name)
                cnx.heartbeat_check()
            except(kombu_exceptions.ChannelLimitExceeded,
                   kombu_exceptions.ConnectionLimitExceeded,
                   kombu_exceptions.OperationalError,
                   kombu_exceptions.NotBoundError,
                   kombu_exceptions.MessageStateError,
                   kombu_exceptions.LimitExceeded) as err:
                raise ChannelFailureException("Error connecting to RabbitMQ, see inner exception for details", err)
            except (OSError, IOError, ConnectionError) as socket_err:
                self._reset_connection()
                raise ChannelFailureException("Error connecting to RabbitMQ, see inner exception for details", socket_err)

        def _consume_errors(exc, interval: int)-> None:
            self._logger.error('Draining error: %s, will retry triggering in %s seconds', exc, interval, exc_info=True)

        self._ensure_connection()

        ensure_kwargs = self.RETRY_OPTIONS.copy()
        ensure_kwargs['errback'] = _consume_errors
        safe_drain = self._conn.ensure(self._consumer, _consume, **ensure_kwargs)
        safe_drain(self._conn, timeout)

        return self._message

    def _reset_connection(self) -> None:
        self._logger.debug('Reset connection to RabbitMQ following socket error')
        self._conn.close()
        self._establish_connection(BrokerConnection(hostname=self._amqp_uri, connect_timeout=self._connect_timeout, heartbeat=self._heartbeat))
        self._establish_channel()
        self._establish_consumer()

    def requeue(self, message: BrightsideMessage) -> None:
        """
            TODO: has does a consumer resend
        """
        self._msg.requeue()

    def run_heartbeat_continuously(self) -> threading.Event:
        """
        For a long runing handler, there is a danger that we do not send a heartbeat message or activity on the
        connection whilst we are running the handler. With a default heartbeat of 30s, for example, there is a risk
        that a handler which takes more than 15s will fail to send the heartbeat in time and then the broker will
        reset the connection. So we spin up another thread, where the user has marked the thread as having a
        long-running thread
        :return: an event to cancel the thread
        """

        cancellation_event = threading.Event()

        # Effectively a no-op if we are not actually a long-running thread
        if not self._is_long_running_handler:
            return cancellation_event

        self._logger.debug("Running long running handler on %s", self._conn)

        def _send_heartbeat(cnx: BrokerConnection, period: int, logger: logging.Logger) -> None:
                while not cancellation_event.is_set():
                    cnx.heartbeat_check()
                    cancellation_event.wait(timeout= period)
                logger.debug("Signalled to exit long-running handler heartbeat")

        heartbeat_thread = threading.Thread(target=_send_heartbeat, args=(self._conn, 1, self._logger), daemon=True)
        self._logger.debug("Begin heartbeat thread for  %s", self._conn)
        heartbeat_thread.start()
        self._logger.debug("Heartbeat running on thread for  %s", self._conn)
        return cancellation_event

    def stop(self):
        if self._conn is not None:
            self._logger.debug("Closing connection: %s", self._conn)
            self._conn.close()
            self._conn = None class ArameProducer(BrightsideProducer):
    """Implements sending a message to a RMQ broker. It does not use a queue, just a connection to the broker
    """
    RETRY_OPTIONS = {
        'interval_start': 1,
        'interval_step': 1,
        'interval_max': 1,
        'max_retries': 3,
    }

    def __init__(self, connection: Connection, logger: logging.Logger=None) -> None:
        self._amqp_uri = connection.amqp_uri
        self._cnx = BrokerConnection(hostname=connection.amqp_uri)
        self._exchange = Exchange(connection.exchange, type=connection.exchange_type, durable=connection.is_durable)
        self._logger = logger or logging.getLogger(__name__)

    def send(self, message: BrightsideMessage):
        # we want to expose our logger to the functions defined in inner scope, so put it in their outer scope

        logger = self._logger

        def _build_message_header(msg: BrightsideMessage) -> Dict:
            return KombuMessageFactory(msg).create_message_header()

        def _publish(sender: Producer) -> None:
            logger.debug("Send message {body} to broker {amqpuri} with routing key {routing_key}"
                         .format(body=message, amqpuri=self._amqp_uri, routing_key=message.header.topic))
            sender.publish(message.body.bytes,
                           headers=_build_message_header(message),
                           exchange=self._exchange,
                           content_type="text/plain",
                           routing_key=message.header.topic,
                           declare=[self._exchange])

        def _error_callback(e, interval) -> None:
            logger.debug("Publishing error: {e}. Will retry in {interval} seconds".format(e=e, interval=interval))

        self._logger.debug("Connect to broker {amqpuri}".format(amqpuri=self._amqp_uri))

        # Producer uses a pool, because you may have many instances in your code, but no heartbeat as a result
        with connections[self._cnx].acquire(block=True) as conn:
            with Producer(conn) as producer:
                ensure_kwargs = self.RETRY_OPTIONS.copy()
                ensure_kwargs['errback'] = _error_callback
                safe_publish = conn.ensure(producer, _publish, **ensure_kwargs)
                safe_publish(producer) class JsonRequestSerializer:
    def __init__(self, request: Request=None, serialized_request: str=None):
        if request is None and serialized_request is None:
            raise MessagingException("You must provide either an object to serialize, or a dictionary of object properties and an object to hydrate")

        self._request = request
        self._serialized_request = serialized_request

    def serialize_to_json(self):
        def _serialize_instance(obj: object) -> Dict:
            d = {}
            d.update(vars(obj))
            for key, value in d.items():
                if isinstance(value, UUID):  # json does not know how to serliaze a UUID, so convince it is a string instead
                    d[key] = str(value)
            return d

        if self._request is None:
            raise MessagingException("You must provide a request to serialize")
        return json.dumps(self._request, default=_serialize_instance)

    def deserialize_from_json(self):
        def _unserialize_instance(d: Dict) -> object:
            for key, value in d.items():
                if isinstance(value, str):  # We need to check if the string on the wire is actually a UUID, by conversion
                    try:
                        guid = UUID(value)
                        value = guid
                    except ValueError:
                        pass
                setattr(self._request, key, value)
            return self._request

        return json.loads(self._serialized_request, object_hook=_unserialize_instance) class Connection:
    """Contains the details required to connect to a RMQ broker: the amqp uri and the exchange"""
    def __init__(self, amqp_uri: str, exchange: str, exchange_type: str = "direct", is_durable: bool = False, connect_timeout: int = 30, heartbeat: int = 30) -> None:
        self._amqp_uri = amqp_uri
        self._exchange = exchange
        self._exchange_type = exchange_type
        self._is_durable = is_durable
        self._connect_timeout = connect_timeout
        self._heartbeat = heartbeat

    @property
    def amqp_uri(self) -> str:
        return self._amqp_uri

    @amqp_uri.setter
    def amqp_uri(self, value: str):
        self._amqp_uri = value

    @property
    def exchange(self) -> str:
        return self._exchange

    @exchange.setter
    def exchange(self, value: str):
        self._exchange = value

    @property
    def exchange_type(self) -> str:
        return self._exchange_type

    @exchange_type.setter
    def exchange_type(self, value: str):
        self._exchange_type = value

    @property
    def is_durable(self):
        return self._is_durable

    @is_durable.setter
    def is_durable(self, value):
        self._is_durable = value

    @property
    def connect_timeout(self) -> int:
        return self._connect_timeout
    
    @connect_timeout.setter
    def connect_timeout(self, value:int):
        self._connect_timeout = value

    @property
    def heartbeat(self) -> int:
        return self._heartbeat
    
    @heartbeat.setter
    def heartbeat(self, value:int):
        self._heartbeat = value class BrightsideMessage:
    """The representation of an on the wire message in Brighter. It abstracts the message typeof the underlying
    implementation and thus acts as a anti-corruption layer between us and an implementation specific message
    type
    """
    def __init__(self, message_header: BrightsideMessageHeader, message_body: BrightsideMessageBody) -> None:
        self._message_header = message_header
        self._message_body = message_body

    @property
    def header(self) -> BrightsideMessageHeader:
        return self._message_header

    @property
    def body(self) -> BrightsideMessageBody:
        return self._message_body

    def handled_count_reached(self, requeue_count: int) -> bool:
        return self._message_header.handled_count >= requeue_count

    @property
    def id(self) -> UUID:
        return self._message_header.id

    def increment_handled_count(self):
        self._message_header.increment_handled_count() class BrightsideConsumerConfiguration:
    """
    A class to encapsulate the configuration required to create a Brightside Consumer
    Required because we need to pass the parameters to a performer that runs the message pump on another thread
    Use is_long_running_handler when you expect a handler to take more than about 10s to execute, this will instruct
    the consumer to spin up a separate thread to send a heartbeat over the connection. Due to the GIL this won't help
    you that much if your handler is CPU bound though.
    """
    def __init__(self, pipeline: Queue, queue_name: str, routing_key: str, prefetch_count: int=1,
                 is_durable: bool=False, is_ha: bool=False, is_long_running_handler: bool=False):
        self._pipeline = pipeline
        self._queue_name = queue_name
        self._routing_key = routing_key
        self._prefetch_count = prefetch_count
        self._is_durable = is_durable
        self._is_ha = is_ha
        self._is_long_running = is_long_running_handler

    @property
    def pipeline(self):
        return self._pipeline

    @property
    def queue_name(self) -> str:
        return self._queue_name

    @property
    def routing_key(self) -> str:
        return self._routing_key

    @property
    def prefetch_count(self) -> int:
        return self._prefetch_count

    @property
    def is_durable(self) -> bool:
        return self._is_durable

    @property
    def is_ha(self) -> bool:
        return self._is_ha

    @property
    def is_long_runing_handler(self) -> bool:
        return self._is_long_running class BrightsideMessageBody:
    """The body of our message. Note that this must use the same binary payload approach as Paramore Brighter to
        ensure that payload is binary compatible. plain/text should be encoded as a UTF8 byte array for example
    """
    def __init__(self, body: str, body_type: str = BrightsideMessageBodyType.text_plain) -> None:
        if body is not None:
            self._encoded_body = body.encode()
        else:
            self._encoded_body = "".encode()
        self._body_type = body_type

    @property
    def value(self) -> str:
        """ Assumes that the body is text/plain i.e. json or xml and so returns the content as a string"""
        return self._encoded_body.decode()

    @property
    def bytes(self) -> bytes:
        """ returns the """
        return self._encoded_body class BrightsideMessageBodyType:
    application_json = "application/json"
    application_xml = "application/xml"
    text_plain = "text/plain"
    text_xml = "text/xml" class BrightsideMessageHeader:
    """The header for our message. Note that this should agree with the Paramore.Brighter definition to ensure that
        different language implementations are compatible
    """
    def __init__(self, identity: UUID, topic: str, message_type: BrightsideMessageType, correlation_id: UUID = None,
                 reply_to: str = None, content_type: str = "text/plain", header_bag: dict = None, handled_count: int = None) -> None:
        self._id = identity
        self._topic = topic
        self._message_type = message_type
        self._correlation_id = correlation_id
        self._reply_to = reply_to
        self._content_type = content_type
        self._msg = None
        self._header_bag = header_bag
        self._handled_count = handled_count if handled_count is not None else 0

    @property
    def bag(self) -> dict:
        return self._header_bag

    @property
    def handled_count(self) -> int:
        return self._handled_count

    @property
    def id(self) -> UUID:
        return self._id

    def increment_handled_count(self):
        self._handled_count += 1

    @property
    def topic(self) -> str:
        return self._topic

    @topic.setter
    def topic(self, value: str):
        self._topic = value

    @property
    def message_type(self) -> BrightsideMessageType:
        return self._message_type

    @property
    def correlation_id(self) -> UUID:
        return self._correlation_id

    @property
    def reply_to(self) -> str:
        return self._reply_to

    @reply_to.setter
    def reply_to(self, value: str):
        self._reply_to = value

    @property
    def content_type(self) -> str:
        return self._content_type

    @content_type.setter
    def content_type(self, value: str):
        self._content_type = value class BrightsideMessageType(Enum):
    """
    MT_UNACCEPTABLE = The message could not be read
    MT_NONE =  The message type has not been configured
    MT_COMMAND = The message was sent as a command, and the producer intended it to be handled by a consumer
    MT_EVENT = The message was raised as an event and the producer does not care if anyone listens to it
        It only contains a simple notification, not the data of what changed
    MT_DOCUMENT = The message was raised as an event and the producer does not care if anyone listens to it
        It contains a notification of what changed
    MT_QUIT = The message was raised as an event and the producer does not care if anyone listens to it
        It contains a notification of what changed
    MT_CALLBACK = Posted back onto the message pump
    """
    MT_UNACCEPTABLE = -1
    MT_NONE = 0
    MT_COMMAND = 1
    MT_EVENT = 2
    MT_DOCUMENT = 3
    MT_QUIT = 5
    MT_CALLBACK = 6
from multiprocessing import Queue
from uuid import uuid4
from tests.config import TestConfig
from tests.messaging_testdoubles import TestMessage
from arame.gateway import ArameConsumer, ArameProducer
from arame.messaging import JsonRequestSerializer
from brightside.connection import Connection
from brightside.messaging import BrightsideMessage, BrightsideConsumerConfiguration, BrightsideMessageBody, BrightsideMessageBodyType, BrightsideMessageHeader, BrightsideMessageType
import unittest
#!/usr/bin/env python
""""
File             : tests_arame_gateway.py
Author           : ian
Created          : 09-01-2016

Last Modified By : ian
Last Modified On : 09-01-2016
***********************************************************************
The MIT License (MIT)
Copyright © 2015 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the “Software”), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
**********************************************************************i*s
"""




class ArameGatewayTests(unittest.TestCase):

    def setUp(self):
        config = TestConfig()
        self._connection = Connection(config.broker_uri, "paramore.brightside.exchange", is_durable=True)

        self._producer = ArameProducer(self._connection)