added listener support to queues, also added support for non version specific tests

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@542955 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 742 insertions, 13 deletions

diff --git a/qpid/python/qpid/delegate.py b/qpid/python/qpid/delegate.py
index 90e5c1edc8..8f5033e485 100644
--- a/qpid/python/qpid/delegate.py
+++ b/qpid/python/qpid/delegate.py
@@ -21,9 +21,12 @@
 Delegate implementation intended for use with the peer module.
 """
 
-import threading, inspect
+import threading, inspect, traceback, sys
 from connection import Method, Request, Response
 
+def _handler_name(method):
+  return "%s_%s" % (method.klass.name, method.name)
+
 class Delegate:
 
   def __init__(self):
@@ -36,11 +39,15 @@ class Delegate:
     try:
       handler = self.handlers[method]
     except KeyError:
-      name = "%s_%s" % (method.klass.name, method.name)
+      name = _handler_name(method)
       handler = getattr(self, name)
       self.handlers[method] = handler
 
-    return handler(channel, frame)
+    try:
+      return handler(channel, frame)
+    except:
+      print >> sys.stderr, "Error in handler: %s\n\n%s" % \
+            (_handler_name(method), traceback.format_exc())
 
   def close(self, reason):
     print "Connection closed: %s" % reason
diff --git a/qpid/python/qpid/queue.py b/qpid/python/qpid/queue.py
index 5438b328ab..af0565b6cc 100644
--- a/qpid/python/qpid/queue.py
+++ b/qpid/python/qpid/queue.py
@@ -31,15 +31,29 @@ class Queue(BaseQueue):
 
   END = object()
 
+  def __init__(self, *args, **kwargs):
+    BaseQueue.__init__(self, *args, **kwargs)
+    self._real_put = self.put
+    self.listener = self._real_put
+
   def close(self):
     self.put(Queue.END)
 
   def get(self, block = True, timeout = None):
-    result = BaseQueue.get(self, block, timeout)
-    if result == Queue.END:
-      # this guarantees that any other waiting threads or any future
-      # calls to get will also result in a Closed exception
-      self.put(Queue.END)
-      raise Closed()
-    else:
-      return result
+    self.put = self._real_put
+    try:
+      result = BaseQueue.get(self, block, timeout)
+      if result == Queue.END:
+        # this guarantees that any other waiting threads or any future
+        # calls to get will also result in a Closed exception
+        self.put(Queue.END)
+        raise Closed()
+      else:
+        return result
+    finally:
+      self.put = self.listener
+      pass
+
+  def listen(self, listener):
+    self.listener = listener
+    self.put = self.listener
diff --git a/qpid/python/qpid/testlib.py b/qpid/python/qpid/testlib.py
index ba2dbe9fc3..2a5e4dcfcd 100644
--- a/qpid/python/qpid/testlib.py
+++ b/qpid/python/qpid/testlib.py
@@ -122,10 +122,11 @@ Options:
         print "Using specification from:", self.specfile
         self.spec = qpid.spec.load(self.specfile, *self.errata)
         if len(self.tests) == 0:
+            self.tests=findmodules("tests")
             if self.use08spec():
-                self.tests=findmodules("tests_0-8")
+                self.tests+=findmodules("tests_0-8")
             else:
-                self.tests=findmodules("tests_0-9")
+                self.tests+=findmodules("tests_0-9")
 
     def testSuite(self):
         class IgnoringTestSuite(unittest.TestSuite):
diff --git a/qpid/python/tests/__init__.py b/qpid/python/tests/__init__.py
new file mode 100644
index 0000000000..9a09d2d04f
--- /dev/null
+++ b/qpid/python/tests/__init__.py
@@ -0,0 +1,20 @@
+# Do not delete - marks this directory as a python package.
+
+#
+# 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.
+#
diff --git a/qpid/python/tests/codec.py b/qpid/python/tests/codec.py
new file mode 100644
index 0000000000..12dd5c85fc
--- /dev/null
+++ b/qpid/python/tests/codec.py
@@ -0,0 +1,611 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+import unittest
+from qpid.codec import Codec
+from cStringIO import StringIO
+from qpid.reference import ReferenceId
+
+__doc__ = """
+
+    This is a unit test script for qpid/codec.py
+
+    It can be run standalone or as part of the existing test framework.
+
+    To run standalone:
+    -------------------
+
+        Place in the qpid/python/tests/ directory and type...
+
+        python codec.py
+
+        A brief output will be printed on screen. The verbose output will be placed inn a file called
+        codec_unit_test_output.txt. [TODO: make this filename configurable]
+
+    To run as part of the existing test framework:
+    -----------------------------------------------
+
+        python run-tests tests.codec
+
+    Change History:
+    -----------------
+        Jimmy John  05/19/2007  Initial draft
+        Jimmy John  05/22/2007  Implemented comments by Rafael Schloming
+
+
+"""
+
+
+# --------------------------------------
+# --------------------------------------
+class BaseDataTypes(unittest.TestCase):
+
+    """
+    Base class containing common functions
+    """
+
+    # ---------------
+    def setUp(self):
+        """
+        standard setUp for unitetest (refer unittest documentation for details)
+        """
+        self.codec = Codec(StringIO())
+
+    # ------------------
+    def tearDown(self):
+        """
+        standard tearDown for unitetest (refer unittest documentation for details)
+        """
+        self.codec.stream.flush()
+        self.codec.stream.close()
+
+    # ----------------------------------------
+    def callFunc(self, functionName, *args):
+        """
+        helper function - given a function name and arguments, calls the function with the args and
+        returns the contents of the stream
+        """
+        getattr(self.codec, functionName)(args[0])
+        return self.codec.stream.getvalue()
+
+    # ----------------------------------------
+    def readFunc(self, functionName, *args):
+        """
+        helper function - creates a input stream and then calls the function with arguments as have been
+        supplied
+        """
+        self.codec.stream = StringIO(args[0])
+        return getattr(self.codec, functionName)()
+
+
+# ----------------------------------------
+# ----------------------------------------
+class IntegerTestCase(BaseDataTypes):
+
+    """
+    Handles octet, short, long, long long
+
+    """
+
+    # -------------------------
+    def __init__(self, *args):
+        """
+        sets constants for use in tests
+        """
+
+        BaseDataTypes.__init__(self, *args)
+        self.const_integer = 2
+        self.const_integer_octet_encoded = '\x02'
+        self.const_integer_short_encoded = '\x00\x02'
+        self.const_integer_long_encoded = '\x00\x00\x00\x02'
+        self.const_integer_long_long_encoded = '\x00\x00\x00\x00\x00\x00\x00\x02'
+
+    # -------------------------- #
+    # Unsigned Octect - 8 bits   #
+    # -------------------------- #
+
+    # --------------------------
+    def test_unsigned_octet(self):
+        """
+        ubyte format requires 0<=number<=255
+        """
+        self.failUnlessEqual(self.callFunc('encode_octet', self.const_integer), self.const_integer_octet_encoded, 'octect encoding FAILED...')
+
+    # -------------------------------------------
+    def test_octet_out_of_upper_range(self):
+        """
+        testing for input above acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_octet, 256)
+
+    # -------------------------------------------
+    def test_uoctet_out_of_lower_range(self):
+        """
+        testing for input below acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_octet, -1)
+
+    # ---------------------------------
+    def test_uoctet_with_fraction(self):
+        """
+        the fractional part should be ignored...
+        """
+        self.failUnlessEqual(self.callFunc('encode_octet', 2.5), self.const_integer_octet_encoded, 'octect encoding FAILED with fractions...')
+
+    # ------------------------------------
+    def test_unsigned_octet_decode(self):
+        """
+        octet decoding
+        """
+        self.failUnlessEqual(self.readFunc('decode_octet', self.const_integer_octet_encoded), self.const_integer, 'octect decoding FAILED...')
+
+    # ----------------------------------- #
+    # Unsigned Short Integers - 16 bits   #
+    # ----------------------------------- #
+
+    # -----------------------
+    def test_ushort_int(self):
+        """
+        testing unsigned short integer
+        """
+        self.failUnlessEqual(self.callFunc('encode_short', self.const_integer), self.const_integer_short_encoded, 'short encoding FAILED...')
+
+    # -------------------------------------------
+    def test_ushort_int_out_of_upper_range(self):
+        """
+        testing for input above acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_short, 65536)
+
+    # -------------------------------------------
+    def test_ushort_int_out_of_lower_range(self):
+        """
+        testing for input below acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_short, -1)
+
+    # ---------------------------------
+    def test_ushort_int_with_fraction(self):
+        """
+        the fractional part should be ignored...
+        """
+        self.failUnlessEqual(self.callFunc('encode_short', 2.5), self.const_integer_short_encoded, 'short encoding FAILED with fractions...')
+
+    # ------------------------------------
+    def test_ushort_int_decode(self):
+        """
+        unsigned short decoding
+        """
+        self.failUnlessEqual(self.readFunc('decode_short', self.const_integer_short_encoded), self.const_integer, 'unsigned short decoding FAILED...')
+
+
+    # ---------------------------------- #
+    # Unsigned Long Integers - 32 bits   #
+    # ---------------------------------- #
+
+    # -----------------------
+    def test_ulong_int(self):
+        """
+        testing unsigned long iteger
+        """
+        self.failUnlessEqual(self.callFunc('encode_long', self.const_integer), self.const_integer_long_encoded, 'long encoding FAILED...')
+
+    # -------------------------------------------
+    def test_ulong_int_out_of_upper_range(self):
+        """
+        testing for input above acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_long, 4294967296)
+
+    # -------------------------------------------
+    def test_ulong_int_out_of_lower_range(self):
+        """
+        testing for input below acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_long, -1)
+
+    # ---------------------------------
+    def test_ulong_int_with_fraction(self):
+        """
+        the fractional part should be ignored...
+        """
+        self.failUnlessEqual(self.callFunc('encode_long', 2.5), self.const_integer_long_encoded, 'long encoding FAILED with fractions...')
+
+    # -------------------------------
+    def test_ulong_int_decode(self):
+        """
+        unsigned long decoding
+        """
+        self.failUnlessEqual(self.readFunc('decode_long', self.const_integer_long_encoded), self.const_integer, 'unsigned long decoding FAILED...')
+
+
+    # --------------------------------------- #
+    # Unsigned Long Long Integers - 64 bits   #
+    # --------------------------------------- #
+
+    # -----------------------
+    def test_ulong_long_int(self):
+        """
+        testing unsinged long long integer
+        """
+        self.failUnlessEqual(self.callFunc('encode_longlong', self.const_integer), self.const_integer_long_long_encoded, 'long long encoding FAILED...')
+
+    # -------------------------------------------
+    def test_ulong_long_int_out_of_upper_range(self):
+        """
+        testing for input above acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_longlong, 18446744073709551616)
+
+    # -------------------------------------------
+    def test_ulong_long_int_out_of_lower_range(self):
+        """
+        testing for input below acceptable range
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_longlong, -1)
+
+    # ---------------------------------
+    def test_ulong_long_int_with_fraction(self):
+        """
+        the fractional part should be ignored...
+        """
+        self.failUnlessEqual(self.callFunc('encode_longlong', 2.5), self.const_integer_long_long_encoded, 'long long encoding FAILED with fractions...')
+
+    # ------------------------------------
+    def test_ulong_long_int_decode(self):
+        """
+        unsigned long long decoding
+        """
+        self.failUnlessEqual(self.readFunc('decode_longlong', self.const_integer_long_long_encoded), self.const_integer, 'unsigned long long decoding FAILED...')
+
+# -----------------------------------
+# -----------------------------------
+class BitTestCase(BaseDataTypes):
+
+    """
+    Handles bits
+    """
+
+    # ----------------------------------------------
+    def callFunc(self, functionName, *args):
+        """
+        helper function
+        """
+        for ele in args:
+            getattr(self.codec, functionName)(ele)
+
+        self.codec.flush()
+        return self.codec.stream.getvalue()
+
+    # -------------------
+    def test_bit1(self):
+        """
+        sends in 11
+        """
+        self.failUnlessEqual(self.callFunc('encode_bit', 1, 1), '\x03', '11 bit encoding FAILED...')
+
+    # -------------------
+    def test_bit2(self):
+        """
+        sends in 10011
+        """
+        self.failUnlessEqual(self.callFunc('encode_bit', 1, 1, 0, 0, 1), '\x13', '10011 bit encoding FAILED...')
+
+    # -------------------
+    def test_bit3(self):
+        """
+        sends in 1110100111 [10 bits(right to left), should be compressed into two octets]
+        """
+        self.failUnlessEqual(self.callFunc('encode_bit', 1,1,1,0,0,1,0,1,1,1), '\xa7\x03', '1110100111(right to left) bit encoding FAILED...')
+
+    # ------------------------------------
+    def test_bit_decode_1(self):
+        """
+        decode bit 1
+        """
+        self.failUnlessEqual(self.readFunc('decode_bit', '\x01'), 1, 'decode bit 1 FAILED...')
+
+    # ------------------------------------
+    def test_bit_decode_0(self):
+        """
+        decode bit 0
+        """
+        self.failUnlessEqual(self.readFunc('decode_bit', '\x00'), 0, 'decode bit 0 FAILED...')
+
+# -----------------------------------
+# -----------------------------------
+class StringTestCase(BaseDataTypes):
+
+    """
+    Handles short strings, long strings
+    """
+
+    # ------------------------------------------------------------- #
+    # Short Strings - 8 bit length followed by zero or more octets  #
+    # ------------------------------------------------------------- #
+
+    # ---------------------------------------
+    def test_short_string_zero_length(self):
+        """
+        0 length short string
+        """
+        self.failUnlessEqual(self.callFunc('encode_shortstr', ''), '\x00', '0 length short string encoding FAILED...')
+
+    # -------------------------------------------
+    def test_short_string_positive_length(self):
+        """
+        positive length short string
+        """
+        self.failUnlessEqual(self.callFunc('encode_shortstr', 'hello world'), '\x0bhello world', 'positive length short string encoding FAILED...')
+
+    # -------------------------------------------
+    def test_short_string_out_of_upper_range(self):
+        """
+        string length > 255
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_shortstr, 'x'*256)
+
+    # ------------------------------------
+    def test_short_string_decode(self):
+        """
+        short string decode
+        """
+        self.failUnlessEqual(self.readFunc('decode_shortstr', '\x0bhello world'), 'hello world', 'short string decode FAILED...')
+
+
+    # ------------------------------------------------------------- #
+    # Long Strings - 32 bit length followed by zero or more octets  #
+    # ------------------------------------------------------------- #
+
+    # ---------------------------------------
+    def test_long_string_zero_length(self):
+        """
+        0 length long string
+        """
+        self.failUnlessEqual(self.callFunc('encode_longstr', ''), '\x00\x00\x00\x00', '0 length long string encoding FAILED...')
+
+    # -------------------------------------------
+    def test_long_string_positive_length(self):
+        """
+        positive length long string
+        """
+        self.failUnlessEqual(self.callFunc('encode_longstr', 'hello world'), '\x00\x00\x00\x0bhello world', 'positive length long string encoding FAILED...')
+
+    # ------------------------------------
+    def test_long_string_decode(self):
+        """
+        long string decode
+        """
+        self.failUnlessEqual(self.readFunc('decode_longstr', '\x00\x00\x00\x0bhello world'), 'hello world', 'long string decode FAILED...')
+
+
+# --------------------------------------
+# --------------------------------------
+class TimestampTestCase(BaseDataTypes):
+
+    """
+    No need of any test cases here as timestamps are implemented as long long which is tested above
+    """
+    pass
+
+# ---------------------------------------
+# ---------------------------------------
+class FieldTableTestCase(BaseDataTypes):
+
+    """
+    Handles Field Tables
+
+    Only S/I type messages seem to be implemented currently
+    """
+
+    # -------------------------
+    def __init__(self, *args):
+        """
+        sets constants for use in tests
+        """
+
+        BaseDataTypes.__init__(self, *args)
+        self.const_field_table_dummy_dict = {'$key1':'value1','$key2':'value2'}
+        self.const_field_table_dummy_dict_encoded = '\x00\x00\x00\x22\x05$key2S\x00\x00\x00\x06value2\x05$key1S\x00\x00\x00\x06value1'
+
+    # -------------------------------------------
+    def test_field_table_name_value_pair(self):
+        """
+        valid name value pair
+        """
+        self.failUnlessEqual(self.callFunc('encode_table', {'$key1':'value1'}), '\x00\x00\x00\x11\x05$key1S\x00\x00\x00\x06value1', 'valid name value pair encoding FAILED...')
+
+    # -------------------------------------------
+    def test_field_table_invalid_field_name(self):
+        """
+        invalid field name
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_table, {'1key1':'value1'})
+
+    # ----------------------------------------------------
+    def test_field_table_invalid_field_name_length(self):
+        """
+        field names can have a maximum length of 128 chars
+        """
+        self.failUnlessRaises(Exception, self.codec.encode_table, {'x'*129:'value1'})
+
+    # ---------------------------------------------------
+    def test_field_table_multiple_name_value_pair(self):
+        """
+        multiple name value pair
+        """
+        self.failUnlessEqual(self.callFunc('encode_table', self.const_field_table_dummy_dict), self.const_field_table_dummy_dict_encoded, 'multiple name value pair encoding FAILED...')
+
+    # ------------------------------------
+    def test_field_table_decode(self):
+        """
+        field table decode
+        """
+        self.failUnlessEqual(self.readFunc('decode_table', self.const_field_table_dummy_dict_encoded), self.const_field_table_dummy_dict, 'field table decode FAILED...')
+
+
+# ------------------------------------
+# ------------------------------------
+class ContentTestCase(BaseDataTypes):
+
+    """
+    Handles Content data types
+    """
+
+    # -----------------------------
+    def test_content_inline(self):
+        """
+        inline content
+        """
+        self.failUnlessEqual(self.callFunc('encode_content', 'hello inline message'), '\x00\x00\x00\x00\x14hello inline message', 'inline content encoding FAILED...')
+
+    # --------------------------------
+    def test_content_reference(self):
+        """
+        reference content
+        """
+        self.failUnlessEqual(self.callFunc('encode_content', ReferenceId('dummyId')), '\x01\x00\x00\x00\x07dummyId', 'reference content encoding FAILED...')
+
+    # ------------------------------------
+    def test_content_inline_decode(self):
+        """
+        inline content decode
+        """
+        self.failUnlessEqual(self.readFunc('decode_content', '\x00\x00\x00\x00\x14hello inline message'), 'hello inline message', 'inline content decode FAILED...')
+
+    # ------------------------------------
+    def test_content_reference_decode(self):
+        """
+        reference content decode
+        """
+        self.failUnlessEqual(self.readFunc('decode_content', '\x01\x00\x00\x00\x07dummyId').id, 'dummyId', 'reference content decode FAILED...')
+
+# ------------------------ #
+# Pre - existing test code #
+# ------------------------ #
+
+# ---------------------
+def test(type, value):
+    """
+    old test function cut/copy/paste from qpid/codec.py
+    """
+    if isinstance(value, (list, tuple)):
+      values = value
+    else:
+      values = [value]
+    stream = StringIO()
+    codec = Codec(stream)
+    for v in values:
+      codec.encode(type, v)
+    codec.flush()
+    enc = stream.getvalue()
+    stream.reset()
+    dup = []
+    for i in xrange(len(values)):
+      dup.append(codec.decode(type))
+    if values != dup:
+      raise AssertionError("%r --> %r --> %r" % (values, enc, dup))
+
+# -----------------------
+def dotest(type, value):
+    """
+    old test function cut/copy/paste from qpid/codec.py
+    """
+    args = (type, value)
+    test(*args)
+
+# -------------
+def oldtests():
+    """
+    old test function cut/copy/paste from qpid/codec.py
+    """
+    for value in ("1", "0", "110", "011", "11001", "10101", "10011"):
+      for i in range(10):
+        dotest("bit", map(lambda x: x == "1", value*i))
+
+    for value in ({}, {"asdf": "fdsa", "fdsa": 1, "three": 3}, {"one": 1}):
+      dotest("table", value)
+
+    for type in ("octet", "short", "long", "longlong"):
+      for value in range(0, 256):
+        dotest(type, value)
+
+    for type in ("shortstr", "longstr"):
+      for value in ("", "a", "asdf"):
+        dotest(type, value)
+
+# -----------------------------------------
+class oldTests(unittest.TestCase):
+
+	"""
+	class to handle pre-existing test cases
+	"""
+
+	# ---------------------------
+	def test_oldtestcases(self):
+		"""
+		call the old tests
+		"""
+		return oldtests()
+
+# ---------------------------
+# ---------------------------
+if __name__ == '__main__':
+
+    codec_test_suite = unittest.TestSuite()
+
+    #adding all the test suites...
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(IntegerTestCase))
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(BitTestCase))
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(StringTestCase))
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(TimestampTestCase))
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(FieldTableTestCase))
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(ContentTestCase))
+
+    #loading pre-existing test case from qpid/codec.py
+    codec_test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(oldTests))
+
+    run_output_stream = StringIO()
+    test_runner = unittest.TextTestRunner(run_output_stream, '', '')
+    test_result = test_runner.run(codec_test_suite)
+
+    print '\n%d test run...' % (test_result.testsRun)
+
+    if test_result.wasSuccessful():
+        print '\nAll tests successful\n'
+
+    if test_result.failures:
+        print '\n----------'
+        print '%d FAILURES:' % (len(test_result.failures))
+        print '----------\n'
+        for failure in test_result.failures:
+            print str(failure[0]) + ' ... FAIL'
+
+    if test_result.errors:
+        print '\n---------'
+        print '%d ERRORS:' % (len(test_result.errors))
+        print '---------\n'
+
+        for error in test_result.errors:
+            print str(error[0]) + ' ... ERROR'
+
+    f = open('codec_unit_test_output.txt', 'w')
+    f.write(str(run_output_stream.getvalue()))
+    f.close()
diff --git a/qpid/python/tests/queue.py b/qpid/python/tests/queue.py
new file mode 100644
index 0000000000..d2e495d207
--- /dev/null
+++ b/qpid/python/tests/queue.py
@@ -0,0 +1,76 @@
+# Do not delete - marks this directory as a python package.
+
+#
+# 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.
+#
+import threading, time
+from unittest import TestCase
+from qpid.queue import Queue, Empty, Closed
+
+
+class QueueTest (TestCase):
+
+  # The qpid queue class just provides sime simple extensions to
+  # python's standard queue data structure, so we don't need to test
+  # all the queue functionality.
+
+  def test_listen(self):
+    LISTEN = object()
+    GET = object()
+    EMPTY = object()
+
+    q = Queue(0)
+    values = []
+    q.listen(lambda x: values.append((LISTEN, x)))
+    q.put(1)
+    assert values[-1] == (LISTEN, 1)
+    q.put(2)
+    assert values[-1] == (LISTEN, 2)
+
+    class Getter(threading.Thread):
+
+      def run(self):
+        try:
+          values.append((GET, q.get(timeout=10)))
+        except Empty:
+          values.append(EMPTY)
+
+    g = Getter()
+    g.start()
+    # let the other thread reach the get
+    time.sleep(2)
+    q.put(3)
+    g.join()
+
+    assert values[-1] == (GET, 3)
+
+    q.put(4)
+    assert values[-1] == (LISTEN, 4)
+
+  def test_close(self):
+    q = Queue(0)
+    q.put(1); q.put(2); q.put(3); q.close()
+    assert q.get() == 1
+    assert q.get() == 2
+    assert q.get() == 3
+    for i in range(10):
+      try:
+        q.get()
+        raise AssertionError("expected Closed")
+      except Closed:
+        pass