#
# 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.messaging import *
from qpid.tests.messaging import Base
import qmf.console

from time import sleep
#
# Tests the Broker's support for message groups
#

class MultiConsumerMsgGroupTests(Base):
    """
    Tests for the behavior of multi-consumer message groups.  These tests allow
    a messages from the same group be consumed by multiple different clients as
    long as each message is processed "in sequence".  See QPID-3346 for
    details.
    """

    def setup_connection(self):
        return Connection.establish(self.broker, **self.connection_options())

    def setup_session(self):
        return self.conn.session()

    def test_simple(self):
        """ Verify simple acquire/accept actions on a set of grouped
        messages shared between two receivers.
        """
        ## Create a msg group queue

        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","A","A","B","B","B","C","C","C"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        ## Queue = a-0, a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
        ## Owners= ---, ---, ---, ---, ---, ---, ---, ---, ---,

        # create consumers on separate sessions: C1,C2
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 should acquire A-0, then C2 should acquire B-3

        m1 = c1.fetch(0);
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        m2 = c2.fetch(0);
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 3

        # C1 Acknowledge A-0
        c1.session.acknowledge(m1);

        # C2 should next acquire A-1
        m3 = c2.fetch(0);
        assert m3.properties['THE-GROUP'] == 'A'
        assert m3.content['index'] == 1

        # C1 should next acquire C-6, since groups A&B are held by c2
        m4 = c1.fetch(0);
        assert m4.properties['THE-GROUP'] == 'C'
        assert m4.content['index'] == 6

        ## Queue = XXX, a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
        ## Owners= ---, ^C2, +C2, ^C2, +C2, +C2, ^C1, +C1, +C1,

        # C2 Acknowledge B-3, freeing up the rest of B group
        c2.session.acknowledge(m2);

        ## Queue = XXX, a-1, a-2, XXX, b-4, b-5, c-6, c-7, c-8...
        ## Owners= ---, ^C2, +C2, ---, ---, ---, ^C1, +C1, +C1,

        # C1 should now acquire B-4, since it is next "free"
        m5 = c1.fetch(0);
        assert m5.properties['THE-GROUP'] == 'B'
        assert m5.content['index'] == 4

        ## Queue = XXX, a-1, a-2, XXX, b-4, b-5, c-6, c-7, c-8...
        ## Owners= ---, ^C2, +C2, ---, ^C1, +C1, ^C1, +C1, +C1,

        # C1 acknowledges C-6, freeing the C group
        c1.session.acknowledge(m4)

        ## Queue = XXX, a-1, a-2, XXX, b-4, b-5, XXX, c-7, c-8...
        ## Owners= ---, ^C2, +C2, ---, ^C1, +C1, ---, ---, ---

        # C2 should next fetch A-2, followed by C-7
        m7 = c2.fetch(0);
        assert m7.properties['THE-GROUP'] == 'A'
        assert m7.content['index'] == 2

        m8 = c2.fetch(0);
        assert m8.properties['THE-GROUP'] == 'C'
        assert m8.content['index'] == 7

        ## Queue = XXX, a-1, a-2, XXX, b-4, b-5, XXX, c-7, c-8...
        ## Owners= ---, ^C2, ^C2, ---, ^C1, +C1, ---, ^C2, +C2

        # have C2 ack all fetched messages, freeing C-8
        c2.session.acknowledge()

        ## Queue = XXX, XXX, XXX, XXX, b-4, b-5, XXX, XXX, c-8...
        ## Owners= ---, ---, ---, ---, ^C1, +C1, ---, ---, ---

        # the next fetch of C2 would get C-8, since B-5 is "owned"
        m9 = c2.fetch(0);
        assert m9.properties['THE-GROUP'] == 'C'
        assert m9.content['index'] == 8

        ## Queue = XXX, XXX, XXX, XXX, b-4, b-5, XXX, XXX, c-8...
        ## Owners= ---, ---, ---, ---, ^C1, +C1, ---, ---, ^C2

        # C1 acks B-4, freeing B-5 for consumption
        c1.session.acknowledge(m5)

        ## Queue = XXX, XXX, XXX, XXX, XXX, b-5, XXX, XXX, c-8...
        ## Owners= ---, ---, ---, ---, ---, ^C2, ---, ---, ^C2

        # the next fetch of C2 would get B-5
        m10 = c2.fetch(0);
        assert m10.properties['THE-GROUP'] == 'B'
        assert m10.content['index'] == 5

        # there should be no more left for C1:
        try:
            mx = c1.fetch(0)
            assert False     # should never get here
        except Empty:
            pass

        c1.session.acknowledge()
        c2.session.acknowledge()
        c1.close()
        c2.close()
        snd.close()

    def test_simple_browse(self):
        """ Test the behavior of a browsing subscription on a message grouping
        queue.
        """

        ## Create a msg group queue

        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        ## Queue = A-0, B-1, A-2, b-3, C-4
        ## Owners= ---, ---, ---, ---, ---

        # create consumer and browser
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})

        m2 = b1.fetch(0);
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 0

        # C1 should acquire A-0

        m1 = c1.fetch(0);
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        ## Queue = A-0, B-1, A-2, b-3, C-4
        ## Owners= ^C1, ---, +C1, ---, ---

        m2 = b1.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 1

        # verify that the browser may see A-2, even though its group is owned
        # by C1
        m2 = b1.fetch(0)
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 2

        m2 = b1.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 3

        # verify the consumer can own groups currently seen by the browser
        m3 = c1.fetch(0);
        assert m3.properties['THE-GROUP'] == 'B'
        assert m3.content['index'] == 1

        m2 = b1.fetch(0)
        assert m2.properties['THE-GROUP'] == 'C'
        assert m2.content['index'] == 4

    def test_release(self):
        """ Verify releasing a message can free its assocated group
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","A","B","B"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 2

        # C1 release m1, and the first group

        s1.acknowledge(m1, Disposition(RELEASED, set_redelivered=True))

        # C2 should be able to get group 'A', msg 'A-0' now
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 0

    def test_reject(self):
        """ Verify rejecting a message can free its associated group
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","A","B","B"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 2

        # C1 rejects m1, and the first group is released
        s1.acknowledge(m1, Disposition(REJECTED))

        # C2 should be able to get group 'A', msg 'A-1' now
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 1

    def test_close(self):
        """ Verify behavior when a consumer that 'owns' a group closes.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","A","B","B"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 will own group A
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # C2 will own group B
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 2

        # C1 shuffles off the mortal coil...
        c1.close();

        # but the session (s1) remains active, so "A" remains blocked
        # from c2, c2 should fetch the next B-3

        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 3

        # and there should be no more messages available for C2
        try:
            m2 = c2.fetch(0)
            assert False     # should never get here
        except Empty:
            pass

        # close session s1, releasing the A group
        s1.close()

        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 0

        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'A'
        assert m2.content['index'] == 1

        # and there should be no more messages now
        try:
            m2 = c2.fetch(0)
            assert False     # should never get here
        except Empty:
            pass

    def test_transaction(self):
        """ Verify behavior when using transactions.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","A","B","B","A","B"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.conn.session(transactional=True)
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.conn.session(transactional=True)
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 gets group A
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # C2 gets group B
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 2

        s1.acknowledge(m1)  # A-0 consumed, A group freed
        s2.acknowledge(m2)  # B-2 consumed, B group freed

        s1.commit()    # A-0 consumption done, A group now free
        s2.rollback()  # releases B-2, and group B

        ## Q: ["A1","B2","B3","A4","B5"]

        # C2 should be able to get the next A
        m3 = c2.fetch(0)
        assert m3.properties['THE-GROUP'] == 'A'
        assert m3.content['index'] == 1

        # C1 should be able to get B-2
        m4 = c1.fetch(0)
        assert m4.properties['THE-GROUP'] == 'B'
        assert m4.content['index'] == 2

        s2.acknowledge(m3)  # C2 consumes A-1
        s1.acknowledge(m4)  # C1 consumes B-2
        s1.commit()    # C1 consume B-2 occurs, free group B

        ## Q: [["A1",]"B3","A4","B5"]

        # A-1 is still considered owned by C2, since the commit has yet to
        # occur, so the next available to C1 would be B-3
        m5 = c1.fetch(0)   # B-3
        assert m5.properties['THE-GROUP'] == 'B'
        assert m5.content['index'] == 3

        # and C2 should find A-4 available, since it owns the A group
        m6 = c2.fetch(0)  # A-4
        assert m6.properties['THE-GROUP'] == 'A'
        assert m6.content['index'] == 4

        s2.acknowledge(m6)  # C2 consumes A-4

        # uh-oh, A-1 and A-4 released, along with A group
        s2.rollback()

        ## Q: ["A1",["B3"],"A4","B5"]
        m7 = c1.fetch(0)   # A-1 is found
        assert m7.properties['THE-GROUP'] == 'A'
        assert m7.content['index'] == 1

        ## Q: [["A1"],["B3"],"A4","B5"]
        # since C1 "owns" both A and B group, C2 should find nothing available
        try:
            m8 = c2.fetch(0)
            assert False    # should not get here
        except Empty:
            pass

        # C1 next gets A4
        m9 = c1.fetch(0)
        assert m9.properties['THE-GROUP'] == 'A'
        assert m9.content['index'] == 4

        s1.acknowledge()

        ## Q: [["A1"],["B3"],["A4"],"B5"]
        # even though C1 acknowledges A1,B3, and A4, B5 is still considered
        # owned as the commit has yet to take place
        try:
            m10 = c2.fetch(0)
            assert False    # should not get here
        except Empty:
            pass

        # now A1,B3,A4 dequeued, B5 should be free
        s1.commit()

        ## Q: ["B5"]
        m11 = c2.fetch(0)
        assert m11.properties['THE-GROUP'] == 'B'
        assert m11.content['index'] == 5

        s2.acknowledge()
        s2.commit()

    def test_send_transaction(self):
        """ Verify behavior when sender is using transactions.
        """
        ssn = self.conn.session(transactional=True)
        snd = ssn.sender("msg-group-q; {create:always, delete:sender," +
                         " node: {x-declare: {arguments:" +
                         " {'qpid.group_header_key':'THE-GROUP'," +
                         "'qpid.shared_msg_group':1}}}}")

        msg = Message(content={'index':0}, properties={"THE-GROUP": "A"})
        snd.send(msg)
        msg = Message(content={'index':1}, properties={"THE-GROUP": "B"})
        snd.send(msg)
        snd.session.commit()
        msg = Message(content={'index':2}, properties={"THE-GROUP": "A"})
        snd.send(msg)

        # Queue: [A0,B1, (uncommitted: A2) ]

        s1 = self.conn.session(transactional=True)
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.conn.session(transactional=True)
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 gets A0, group A
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # C2 gets B2, group B
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 1

        # Since A2 uncommitted, there should be nothing left to fetch
        try:
            mX = c1.fetch(0)
            assert False    # should not get here
        except Empty:
            pass
        try:
            mX = c2.fetch(0)
            assert False    # should not get here
        except Empty:
            pass

        snd.session.commit()
        msg = Message(content={'index':3}, properties={"THE-GROUP": "B"})
        snd.send(msg)

        # Queue: [A2, (uncommitted: B3) ]

        # B3 has yet to be committed, so C2 should see nothing available:
        try:
            mX = c2.fetch(0)
            assert False    # should not get here
        except Empty:
            pass

        # but A2 should be available to C1
        m3 = c1.fetch(0)
        assert m3.properties['THE-GROUP'] == 'A'
        assert m3.content['index'] == 2

        # now make B3 available
        snd.session.commit()

        # C1 should still be done:
        try:
            mX = c1.fetch(0)
            assert False    # should not get here
        except Empty:
            pass

        # but C2 should find the new B
        m4 = c2.fetch(0)
        assert m4.properties['THE-GROUP'] == 'B'
        assert m4.content['index'] == 3

        # extra: have C1 rollback, verify C2 finds the released 'A' messages
        c1.session.rollback()

        ## Q: ["A0","A2"]

        # C2 should be able to get the next A
        m5 = c2.fetch(0)
        assert m5.properties['THE-GROUP'] == 'A'
        assert m5.content['index'] == 0

        m6 = c2.fetch(0)
        assert m6.properties['THE-GROUP'] == 'A'
        assert m6.content['index'] == 2

        c2.session.acknowledge()
        c2.session.commit()

    def test_query(self):
        """ Verify the queue query method against message groups
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","C","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        m1 = c1.fetch(0)
        m2 = c2.fetch(0)

        # at this point, group A should be owned by C1, group B by C2, and
        # group C should be available

        # now setup a QMF session, so we can call methods
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        brokers = self.qmf_session.getObjects(_class="broker")
        assert len(brokers) == 1
        broker = brokers[0]

        # verify the query method call's group information
        rc = broker.query("queue", "msg-group-q")
        assert rc.status == 0
        assert rc.text == "OK"
        results = rc.outArgs['results']
        assert 'qpid.message_group_queue' in results
        q_info = results['qpid.message_group_queue']
        assert 'group_header_key' in q_info and q_info['group_header_key'] == "THE-GROUP"
        assert 'group_state' in q_info and len(q_info['group_state']) == 3
        for g_info in q_info['group_state']:
            assert 'group_id' in g_info
            if g_info['group_id'] == "A":
                assert g_info['msg_count'] == 3
                assert g_info['consumer'] != ""
            elif g_info['group_id'] == "B":
                assert g_info['msg_count'] == 2
                assert g_info['consumer'] != ""
            elif g_info['group_id'] == "C":
                assert g_info['msg_count'] == 2
                assert g_info['consumer'] == ""
            else:
                assert(False)    # should never get here
        self.qmf_session.delBroker(self.qmf_broker)

    def test_purge_free(self):
        """ Verify we can purge a queue of all messages of a given "unowned"
        group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," + 
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # now setup a QMF session, so we can call methods
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0]
        assert queue
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "B" }}
        assert queue.msgDepth == 6
        rc = queue.purge(0, msg_filter)
        assert rc.status == 0
        queue.update()
        assert queue.msgDepth == 4

        # verify all B's removed....
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] != 'B'
                count += 1
        except Empty:
            pass
        assert count == 4

        self.qmf_session.delBroker(self.qmf_broker)

    def test_purge_acquired(self):
        """ Verify we can purge messages from an acquired group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # acquire group "A"
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # now setup a QMF session, so we can purge group A
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0]
        assert queue
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "A" }}
        assert queue.msgDepth == 6
        rc = queue.purge(0, msg_filter)
        assert rc.status == 0
        queue.update()
        queue.msgDepth == 4   # the pending acquired A still counts!
        s1.acknowledge()

        # verify all other A's removed....
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] != 'A'
                count += 1
        except Empty:
            pass
        assert count == 3   # only 3 really available
        s1.acknowledge()    # ack the consumed A-0
        self.qmf_session.delBroker(self.qmf_broker)

    def test_purge_count(self):
        """ Verify we can purge a fixed number of messages from an acquired
        group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # acquire group "A"
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # now setup a QMF session, so we can purge group A
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0]
        assert queue
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "A" }}
        assert queue.msgDepth == 6
        rc = queue.purge(1, msg_filter)
        assert rc.status == 0
        queue.update()
        queue.msgDepth == 5   # the pending acquired A still counts!

        # verify all other A's removed....
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
        count = 0
        a_count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                if m2.properties['THE-GROUP'] != 'A':
                    count += 1
                else:
                    a_count += 1
        except Empty:
            pass
        assert count == 3   # non-A's
        assert a_count == 1 # assumes the acquired message was not the one purged and regular browsers don't get acquired messages
        s1.acknowledge()    # ack the consumed A-0
        self.qmf_session.delBroker(self.qmf_broker)

    def test_move_all(self):
        """ Verify we can move messages from an acquired group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # set up destination queue
        rcvr = self.ssn.receiver("dest-q; {create:always, delete:receiver," +
                                 " node: {x-declare: {arguments:" +
                                 " {'qpid.group_header_key':'THE-GROUP'," +
                                 "'qpid.shared_msg_group':1}}}}")

        # acquire group "A"
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # now setup a QMF session, so we can move what's left of group A
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        brokers = self.qmf_session.getObjects(_class="broker")
        assert len(brokers) == 1
        broker = brokers[0]
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "A" }}
        rc = broker.queueMoveMessages("msg-group-q", "dest-q", 0, msg_filter)
        assert rc.status == 0

        # verify all other A's removed from msg-group-q
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] != 'A'
                count += 1
        except Empty:
            pass
        assert count == 3   # only 3 really available

        # verify the moved A's are at the dest-q
        s2 = self.setup_session()
        b1 = s2.receiver("dest-q; {mode: browse}", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] == 'A'
                assert m2.content['index'] == 2 or m2.content['index'] == 5
                count += 1
        except Empty:
            pass
        assert count == 2 # two A's moved

        s1.acknowledge()    # ack the consumed A-0
        self.qmf_session.delBroker(self.qmf_broker)

    def test_move_count(self):
        """ Verify we can move a fixed number of messages from an acquired group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # set up destination queue
        rcvr = self.ssn.receiver("dest-q; {create:always, delete:receiver," +
                                 " node: {x-declare: {arguments:" +
                                 " {'qpid.group_header_key':'THE-GROUP'," +
                                 "'qpid.shared_msg_group':1}}}}")

        # now setup a QMF session, so we can move group B
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        brokers = self.qmf_session.getObjects(_class="broker")
        assert len(brokers) == 1
        broker = brokers[0]
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "B" }}
        rc = broker.queueMoveMessages("msg-group-q", "dest-q", 3, msg_filter)
        assert rc.status == 0

        # verify all B's removed from msg-group-q
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] != 'B'
                count += 1
        except Empty:
            pass
        assert count == 4

        # verify the moved B's are at the dest-q
        s2 = self.setup_session()
        b1 = s2.receiver("dest-q; {mode: browse}", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] == 'B'
                assert m2.content['index'] == 1 or m2.content['index'] == 3
                count += 1
        except Empty:
            pass
        assert count == 2

        self.qmf_session.delBroker(self.qmf_broker)

    def test_reroute(self):
        """ Verify we can reroute messages from an acquired group.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        # create a topic exchange for the reroute
        rcvr = self.ssn.receiver("reroute-q; {create: always, delete:receiver," +
                                 " node: {type: topic}}")

        # acquire group "A"
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # now setup a QMF session, so we can reroute group A
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        queue = self.qmf_session.getObjects(_class="queue", name="msg-group-q")[0]
        assert queue
        msg_filter = { 'filter_type' : 'header_match_str',
                       'filter_params' : { 'header_key' : "THE-GROUP",
                                           'header_value' : "A" }}
        assert queue.msgDepth == 6
        rc = queue.reroute(0, False, "reroute-q", msg_filter)
        assert rc.status == 0
        queue.update()
        queue.msgDepth == 4   # the pending acquired A still counts!

        # verify all other A's removed....
        s2 = self.setup_session()
        b1 = s2.receiver("msg-group-q", options={"capacity":0})
        count = 0
        try:
            while True:
                m2 = b1.fetch(0)
                assert m2.properties['THE-GROUP'] != 'A'
                count += 1
        except Empty:
            pass
        assert count == 3   # only 3 really available

        # and what of reroute-q?
        count = 0
        try:
            while True:
                m2 = rcvr.fetch(0)
                assert m2.properties['THE-GROUP'] == 'A'
                assert m2.content['index'] == 2 or m2.content['index'] == 5
                count += 1
        except Empty:
            pass
        assert count == 2

        s1.acknowledge()    # ack the consumed A-0
        self.qmf_session.delBroker(self.qmf_broker)

    def test_queue_delete(self):
        """ Test deleting a queue while consumers are active.
        """

        ## Create a msg group queue

        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A","B","C"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        ## Queue = A-0, B-1, A-2, b-3, C-4
        ## Owners= ---, ---, ---, ---, ---

        # create consumers
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 should acquire A-0
        m1 = c1.fetch(0);
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        # c2 acquires B-1
        m2 = c2.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 1

        # with group A and B owned, and C free, delete the
        # queue
        snd.close()
        self.ssn.close()

    def test_default_group_id(self):
        """ Verify the queue assigns the default group id should a message
        arrive without a group identifier.
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        m = Message(content={}, properties={"NO-GROUP-HEADER":"HA-HA"})
        snd.send(m)

        # now setup a QMF session, so we can call methods
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        brokers = self.qmf_session.getObjects(_class="broker")
        assert len(brokers) == 1
        broker = brokers[0]

        # grab the group state off the queue, and verify the default group is
        # present ("qpid.no-group" is the broker default)
        rc = broker.query("queue", "msg-group-q")
        assert rc.status == 0
        assert rc.text == "OK"
        results = rc.outArgs['results']
        assert 'qpid.message_group_queue' in results
        q_info = results['qpid.message_group_queue']
        assert 'group_header_key' in q_info and q_info['group_header_key'] == "THE-GROUP"
        assert 'group_state' in q_info and len(q_info['group_state']) == 1
        g_info = q_info['group_state'][0]
        assert 'group_id' in g_info
        assert g_info['group_id'] == 'qpid.no-group'

        self.qmf_session.delBroker(self.qmf_broker)


    def test_transaction_order(self):
        """ Verify that rollback does not reorder the messages with respect to
        the consumer (QPID-3804)
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.conn.session(transactional=True)
        c1 = s1.receiver("msg-group-q", options={"capacity":0})

        # C1 gets group A
        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0
        s1.acknowledge(m1)

        s1.rollback()  # release A back to the queue

        # the order should be preserved as follows:

        m1 = c1.fetch(0)
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        m2 = c1.fetch(0)
        assert m2.properties['THE-GROUP'] == 'B'
        assert m2.content['index'] == 1

        m3 = c1.fetch(0)
        assert m3.properties['THE-GROUP'] == 'A'
        assert m3.content['index'] == 2

        s1.commit()

        c1.close()
        s1.close()
        snd.close()


    def test_ttl_expire(self):
        """ Verify that expired (TTL) group messages are skipped correctly
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","C","A","B","C"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            if m.properties['THE-GROUP'] == 'B':
                m.ttl = 1;
            snd.send(m)

        sleep(2)  # let all B's expire

        # create consumers on separate sessions: C1,C2
        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        # C1 should acquire A-0, then C2 should acquire C-2, Group B should
        # expire and never be fetched

        m1 = c1.fetch(0);
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 0

        m2 = c2.fetch(0);
        assert m2.properties['THE-GROUP'] == 'C'
        assert m2.content['index'] == 2

        m1 = c1.fetch(0);
        assert m1.properties['THE-GROUP'] == 'A'
        assert m1.content['index'] == 3

        m2 = c2.fetch(0);
        assert m2.properties['THE-GROUP'] == 'C'
        assert m2.content['index'] == 5

        # there should be no more left for either consumer
        try:
            mx = c1.fetch(0)
            assert False     # should never get here
        except Empty:
            pass
        try:
            mx = c2.fetch(0)
            assert False     # should never get here
        except Empty:
            pass

        c1.session.acknowledge()
        c2.session.acknowledge()
        c1.close()
        c2.close()
        snd.close()


class StickyConsumerMsgGroupTests(Base):
    """
    Tests for the behavior of sticky-consumer message groups.  These tests
    expect all messages from the same group be consumed by the same clients.
    See QPID-3347 for details.
    """
    pass  # TBD