Update from trunk r1375509 through r1450773asyncstore

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/asyncstore@1451244 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 107 insertions, 3 deletions

diff --git a/tests/setup.py b/tests/setup.py
index 8d5345d56e..67d2c87ad9 100755
--- a/tests/setup.py
+++ b/tests/setup.py
@@ -20,7 +20,7 @@
 from distutils.core import setup
 
 setup(name="qpid-tests",
-      version="0.19",
+      version="0.21",
       author="Apache Qpid",
       author_email="dev@qpid.apache.org",
       packages=["qpid_tests", "qpid_tests.broker_0_10", "qpid_tests.broker_0_9",
diff --git a/tests/src/py/qpid_tests/broker_0_10/management.py b/tests/src/py/qpid_tests/broker_0_10/management.py
index 4ec3e0dd03..126b96853b 100644
--- a/tests/src/py/qpid_tests/broker_0_10/management.py
+++ b/tests/src/py/qpid_tests/broker_0_10/management.py
@@ -140,6 +140,30 @@ class ManagementTest (TestBase010):
                 return exchange
         return None
 
+    def test_move_queued_messages_empty(self):
+        """
+        Test that moving messages from an empty queue does not cause an error.
+        """
+        self.startQmf()
+        session = self.session
+        "Set up source queue"
+        session.queue_declare(queue="src-queue-empty", exclusive=True, auto_delete=True)
+
+        "Set up destination queue"
+        session.queue_declare(queue="dest-queue-empty", exclusive=True, auto_delete=True)
+
+        queues = self.qmf.getObjects(_class="queue")
+
+        "Move all messages from src-queue-empty to dest-queue-empty"
+        result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("src-queue-empty", "dest-queue-empty", 0, {})
+        self.assertEqual (result.status, 0) 
+
+        sq = self.qmf.getObjects(_class="queue", name="src-queue-empty")[0]
+        dq = self.qmf.getObjects(_class="queue", name="dest-queue-empty")[0]
+
+        self.assertEqual (sq.msgDepth,0)
+        self.assertEqual (dq.msgDepth,0)
+
     def test_move_queued_messages(self):
         """
         Test ability to move messages from the head of one queue to another.
diff --git a/tests/src/py/qpid_tests/broker_0_8/basic.py b/tests/src/py/qpid_tests/broker_0_8/basic.py
index d5837fc19c..606aad1293 100644
--- a/tests/src/py/qpid_tests/broker_0_8/basic.py
+++ b/tests/src/py/qpid_tests/broker_0_8/basic.py
@@ -79,6 +79,51 @@ class BasicTests(TestBase):
         except Closed, e:
             self.assertChannelException(403, e.args[0])
 
+    def test_reconnect_to_durable_subscription(self):
+      try:
+        publisherchannel = self.channel
+        my_id = "my_id"
+        consumer_connection_properties_with_instance = {"instance": my_id}
+        queue_for_subscription = "queue_for_subscription_%s" % my_id
+        topic_name = "my_topic_name"
+        test_message = self.uniqueString()
+
+        durable_subscription_client = self.connect(client_properties=consumer_connection_properties_with_instance)
+        consumerchannel = durable_subscription_client.channel(1)
+        consumerchannel.channel_open()
+
+        self._declare_and_bind_exclusive_queue_on_topic_exchange(consumerchannel, queue_for_subscription, topic_name)
+
+        # disconnect
+        durable_subscription_client.close()
+
+        # send message to topic
+        publisherchannel.basic_publish(routing_key=topic_name, exchange="amq.topic", content=Content(test_message))
+
+        # reconnect and consume message
+        durable_subscription_client = self.connect(client_properties=consumer_connection_properties_with_instance)
+        consumerchannel = durable_subscription_client.channel(1)
+        consumerchannel.channel_open()
+
+        self._declare_and_bind_exclusive_queue_on_topic_exchange(consumerchannel, queue_for_subscription, topic_name)
+
+        # Create consumer and consume the message that was sent whilst subscriber was disconnected.  By convention we
+        # declare the consumer as exclusive to forbid concurrent access.
+        subscription = consumerchannel.basic_consume(queue=queue_for_subscription, exclusive=True)
+        queue = durable_subscription_client.queue(subscription.consumer_tag)
+
+        # consume and verify message content
+        msg = queue.get(timeout=1)
+        self.assertEqual(test_message, msg.content.body)
+        consumerchannel.basic_ack(delivery_tag=msg.delivery_tag)
+      finally:
+        publisherchannel.queue_delete(queue=queue_for_subscription)
+        durable_subscription_client.close()
+
+    def _declare_and_bind_exclusive_queue_on_topic_exchange(self, channel, queue, topic_name):
+        channel.queue_declare(queue=queue, exclusive=True, auto_delete=False, durable=True)
+        channel.queue_bind(exchange="amq.topic", queue=queue, routing_key=topic_name)
+
     def test_consume_queue_errors(self):
         """
         Test error conditions associated with the queue field of the consume method:
@@ -129,7 +174,7 @@ class BasicTests(TestBase):
         myqueue = self.client.queue("my-consumer")
         msg = myqueue.get(timeout=1)
         self.assertEqual("One", msg.content.body)
-	
+
         #cancel should stop messages being delivered
         channel.basic_cancel(consumer_tag="my-consumer")
         channel.basic_publish(routing_key="test-queue-4", content=Content("Two"))
diff --git a/tests/src/py/qpid_tests/broker_0_9/__init__.py b/tests/src/py/qpid_tests/broker_0_9/__init__.py
index d9f2ed7dbb..6b46b96b1d 100644
--- a/tests/src/py/qpid_tests/broker_0_9/__init__.py
+++ b/tests/src/py/qpid_tests/broker_0_9/__init__.py
@@ -19,4 +19,4 @@
 # under the License.
 #
 
-import query, queue
+import query, queue, messageheader
diff --git a/tests/src/py/qpid_tests/broker_0_9/messageheader.py b/tests/src/py/qpid_tests/broker_0_9/messageheader.py
new file mode 100644
index 0000000000..3526cf37af
--- /dev/null
+++ b/tests/src/py/qpid_tests/broker_0_9/messageheader.py
@@ -0,0 +1,35 @@
+#
+# 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.testlib import TestBase
+
+class MessageHeaderTests(TestBase):
+    """Verify that messages with headers work as expected"""
+
+    def test_message_with_integer_header(self):
+        props={"headers":{"one":1, "zero":0}}
+        self.queue_declare(queue="q")
+        q = self.consume("q")
+        self.assertPublishGet(q, routing_key="q", properties=props)
+
+    def test_message_with_string_header(self):
+        props={"headers":{"mystr":"hello world", "myempty":""}}
+        self.queue_declare(queue="q")
+        q = self.consume("q")
+        self.assertPublishGet(q, routing_key="q", properties=props)