#
# 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.
#
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.datatypes import Message, RangedSet
from qpid.testlib import TestBase010

class TxTests(TestBase010):
    """
    Tests for 'methods' on the amqp tx 'class'
    """

    def test_commit(self):
        """
        Test that commited publishes are delivered and commited acks are not re-delivered
        """
        session = self.session

        #declare queues and create subscribers in the checking session
        #to ensure that the queues are not auto-deleted too early:
        self.declare_queues(["tx-commit-a", "tx-commit-b", "tx-commit-c"])
        session.message_subscribe(queue="tx-commit-a", destination="qa")
        session.message_subscribe(queue="tx-commit-b", destination="qb")
        session.message_subscribe(queue="tx-commit-c", destination="qc")

        #use a separate session for actual work
        session2 = self.conn.session("worker", 2)
        self.perform_txn_work(session2, "tx-commit-a", "tx-commit-b", "tx-commit-c")
        session2.tx_commit()
        session2.close()

        session.tx_select()

        self.enable_flow("qa")
        queue_a = session.incoming("qa")

        self.enable_flow("qb")
        queue_b = session.incoming("qb")

        self.enable_flow("qc")
        queue_c = session.incoming("qc")

        #check results
        for i in range(1, 5):
            msg = queue_c.get(timeout=1)
            self.assertEqual("TxMessage %d" % i, msg.body)
            session.message_accept(RangedSet(msg.id))

        msg = queue_b.get(timeout=1)
        self.assertEqual("TxMessage 6", msg.body)
        session.message_accept(RangedSet(msg.id))

        msg = queue_a.get(timeout=1)
        self.assertEqual("TxMessage 7", msg.body)
        session.message_accept(RangedSet(msg.id))

        for q in [queue_a, queue_b, queue_c]:
            try:
                extra = q.get(timeout=1)
                self.fail("Got unexpected message: " + extra.body)
            except Empty: None

        #cleanup
        session.tx_commit()

    def test_auto_rollback(self):
        """
        Test that a session closed with an open transaction is effectively rolled back
        """
        session = self.session
        self.declare_queues(["tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c"])
        session.message_subscribe(queue="tx-autorollback-a", destination="qa")
        session.message_subscribe(queue="tx-autorollback-b", destination="qb")
        session.message_subscribe(queue="tx-autorollback-c", destination="qc")

        session2 = self.conn.session("worker", 2)
        queue_a, queue_b, queue_c, ignore = self.perform_txn_work(session2, "tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c")

        for q in [queue_a, queue_b, queue_c]:
            try:
                extra = q.get(timeout=1)
                self.fail("Got unexpected message: " + extra.body)
            except Empty: None

        session2.close()

        session.tx_select()

        self.enable_flow("qa")
        queue_a = session.incoming("qa")

        self.enable_flow("qb")
        queue_b = session.incoming("qb")

        self.enable_flow("qc")
        queue_c = session.incoming("qc")

        #check results
        for i in range(1, 5):
            msg = queue_a.get(timeout=1)
            self.assertEqual("Message %d" % i, msg.body)
            session.message_accept(RangedSet(msg.id))

        msg = queue_b.get(timeout=1)
        self.assertEqual("Message 6", msg.body)
        session.message_accept(RangedSet(msg.id))

        msg = queue_c.get(timeout=1)
        self.assertEqual("Message 7", msg.body)
        session.message_accept(RangedSet(msg.id))

        for q in [queue_a, queue_b, queue_c]:
            try:
                extra = q.get(timeout=1)
                self.fail("Got unexpected message: " + extra.body)
            except Empty: None

        #cleanup
        session.tx_commit()

    def test_rollback(self):
        """
        Test that rolled back publishes are not delivered and rolled back acks are re-delivered
        """
        session = self.session
        queue_a, queue_b, queue_c, consumed = self.perform_txn_work(session, "tx-rollback-a", "tx-rollback-b", "tx-rollback-c")

        for q in [queue_a, queue_b, queue_c]:
            try:
                extra = q.get(timeout=1)
                self.fail("Got unexpected message: " + extra.body)
            except Empty: None

        session.tx_rollback()

        #need to release messages to get them redelivered now:
        session.message_release(consumed)

        #check results
        for i in range(1, 5):
            msg = queue_a.get(timeout=1)
            self.assertEqual("Message %d" % i, msg.body)
            session.message_accept(RangedSet(msg.id))

        msg = queue_b.get(timeout=1)
        self.assertEqual("Message 6", msg.body)
        session.message_accept(RangedSet(msg.id))

        msg = queue_c.get(timeout=1)
        self.assertEqual("Message 7", msg.body)
        session.message_accept(RangedSet(msg.id))

        for q in [queue_a, queue_b, queue_c]:
            try:
                extra = q.get(timeout=1)
                self.fail("Got unexpected message: " + extra.body)
            except Empty: None

        #cleanup
        session.tx_commit()

    def perform_txn_work(self, session, name_a, name_b, name_c):
        """
        Utility method that does some setup and some work under a transaction. Used for testing both
        commit and rollback
        """
        #setup:
        self.declare_queues([name_a, name_b, name_c])

        key = "my_key_" + name_b
        topic = "my_topic_" + name_c

        session.exchange_bind(queue=name_b, exchange="amq.direct", binding_key=key)
        session.exchange_bind(queue=name_c, exchange="amq.topic", binding_key=topic)

        dp = session.delivery_properties(routing_key=name_a)
        for i in range(1, 5):
            mp = session.message_properties(message_id="msg%d" % i)
            session.message_transfer(message=Message(dp, mp, "Message %d" % i))

        dp = session.delivery_properties(routing_key=key)
        mp = session.message_properties(message_id="msg6")
        session.message_transfer(destination="amq.direct", message=Message(dp, mp, "Message 6"))

        dp = session.delivery_properties(routing_key=topic)
        mp = session.message_properties(message_id="msg7")
        session.message_transfer(destination="amq.topic", message=Message(dp, mp, "Message 7"))

        session.tx_select()

        #consume and ack messages
        acked = RangedSet()
        self.subscribe(session, queue=name_a, destination="sub_a")
        queue_a = session.incoming("sub_a")
        for i in range(1, 5):
            msg = queue_a.get(timeout=1)
            acked.add(msg.id)
            self.assertEqual("Message %d" % i, msg.body)

        self.subscribe(session, queue=name_b, destination="sub_b")
        queue_b = session.incoming("sub_b")
        msg = queue_b.get(timeout=1)
        self.assertEqual("Message 6", msg.body)
        acked.add(msg.id)

        sub_c = self.subscribe(session, queue=name_c, destination="sub_c")
        queue_c = session.incoming("sub_c")
        msg = queue_c.get(timeout=1)
        self.assertEqual("Message 7", msg.body)
        acked.add(msg.id)

        session.message_accept(acked)

        dp = session.delivery_properties(routing_key=topic)
        #publish messages
        for i in range(1, 5):
            mp = session.message_properties(message_id="tx-msg%d" % i)
            session.message_transfer(destination="amq.topic", message=Message(dp, mp, "TxMessage %d" % i))

        dp = session.delivery_properties(routing_key=key)
        mp = session.message_properties(message_id="tx-msg6")
        session.message_transfer(destination="amq.direct", message=Message(dp, mp, "TxMessage 6"))

        dp = session.delivery_properties(routing_key=name_a)
        mp = session.message_properties(message_id="tx-msg7")
        session.message_transfer(message=Message(dp, mp, "TxMessage 7"))
        return queue_a, queue_b, queue_c, acked

    def declare_queues(self, names, session=None):
        session = session or self.session
        for n in names:
            session.queue_declare(queue=n, auto_delete=True)

    def subscribe(self, session=None, **keys):
        session = session or self.session
        consumer_tag = keys["destination"]
        session.message_subscribe(**keys)
        session.message_flow(destination=consumer_tag, unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination=consumer_tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL)

    def enable_flow(self, tag, session=None):
        session = session or self.session
        session.message_flow(destination=tag, unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination=tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL)

    def complete(self, session, msg):
        session.receiver._completed.add(msg.id)#TODO: this may be done automatically
        session.channel.session_completed(session.receiver._completed)