text: #
  # 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.
  #
  
  # setup, usage, teardown, errors(sync), errors(async), stress, soak,
  # boundary-conditions, config
  
  import errno, os, socket, sys, time
  from qpid import compat
  from qpid.compat import set
  from qpid.messaging import *
  from qpid.messaging.transports import TRANSPORTS
  from qpid.tests.messaging import Base
  from threading import Thread
  
  class SetupTests(Base):
  
    def testEstablish(self):
      self.conn = Connection.establish(self.broker, **self.connection_options())
      self.ping(self.conn.session())
  
    def testOpen(self):
      self.conn = Connection(self.broker, **self.connection_options())
      self.conn.open()
      self.ping(self.conn.session())
  
    def testOpenReconnectURLs(self):
      options = self.connection_options()
      options["reconnect_urls"] = [self.broker, self.broker]
      self.conn = Connection(self.broker, **options)
      self.conn.open()
      self.ping(self.conn.session())
  
    def testTcpNodelay(self):
      self.conn = Connection.establish(self.broker, tcp_nodelay=True)
      assert self.conn._driver._transport.socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
  
    def testConnectError(self):
      try:
        # Specifying port 0 yields a bad address on Windows; port 4 is unassigned
        self.conn = Connection.establish("localhost:4")
        assert False, "connect succeeded"
      except ConnectError, e:
        assert "refused" in str(e)
  
    def testGetError(self):
      self.conn = Connection("localhost:0")
      try:
        self.conn.open()
        assert False, "connect succeeded"
      except ConnectError, e:
        assert self.conn.get_error() == e
  
    def use_fds(self):
      fds = []
      try:
        while True:
          fds.append(os.open(getattr(os, "devnull", "/dev/null"), os.O_RDONLY))
      except OSError, e:
        if e.errno != errno.EMFILE:
          raise e
        else:
          return fds
  
    def testOpenCloseResourceLeaks(self):
      fds = self.use_fds()
      try:
        for i in range(32):
          if fds: os.close(fds.pop())
        for i in xrange(64):
          conn = Connection.establish(self.broker, **self.connection_options())
          conn.close()
      finally:
        while fds:
          os.close(fds.pop())
  
    def testOpenFailResourceLeaks(self):
      fds = self.use_fds()
      try:
        for i in range(32):
          if fds: os.close(fds.pop())
        for i in xrange(64):
          conn = Connection("localhost:0", **self.connection_options())
          # XXX: we need to force a waiter to be created for this test
          # to work
          conn._lock.acquire()
          conn._wait(lambda: False, timeout=0.001)
          conn._lock.release()
          try:
            conn.open()
          except ConnectError, e:
            pass
      finally:
        while fds:
          os.close(fds.pop())
  
    def testReconnect(self):
      options = self.connection_options()
      real = TRANSPORTS["tcp"]
  
      class flaky:
  
        def __init__(self, conn, host, port):
          self.real = real(conn, host, port)
          self.sent_count = 0
          self.recv_count = 0
  
        def fileno(self):
          return self.real.fileno()
  
        def reading(self, reading):
          return self.real.reading(reading)
  
        def writing(self, writing):
          return self.real.writing(writing)
  
        def send(self, bytes):
          if self.sent_count > 2048:
            raise socket.error("fake error")
          n = self.real.send(bytes)
          self.sent_count += n
          return n
  
        def recv(self, n):
          if self.recv_count > 2048:
            return ""
          bytes = self.real.recv(n)
          self.recv_count += len(bytes)
          return bytes
  
        def close(self):
          self.real.close()
  
      TRANSPORTS["flaky"] = flaky
  
      options["reconnect"] = True
      options["reconnect_interval"] = 0
      options["reconnect_limit"] = 100
      options["reconnect_log"] = False
      options["transport"] = "flaky"
  
      self.conn = Connection.establish(self.broker, **options)
      ssn = self.conn.session()
      snd = ssn.sender("test-reconnect-queue; {create: always, delete: always}")
      rcv = ssn.receiver(snd.target)
  
      msgs = [self.message("testReconnect", i) for i in range(20)]
      for m in msgs:
        snd.send(m)
  
      content = set()
      drained = []
      duplicates = []
      try:
        while True:
          m = rcv.fetch(timeout=0)
          if m.content not in content:
            content.add(m.content)
            drained.append(m)
          else:
            duplicates.append(m)
          ssn.acknowledge(m)
      except Empty:
        pass
      # XXX: apparently we don't always get duplicates, should figure out why
      #assert duplicates, "no duplicates"
      assert len(drained) == len(msgs)
      for m, d in zip(msgs, drained):
        # XXX: we should figure out how to provide proper end to end
        # redelivered
        self.assertEcho(m, d, d.redelivered)
  
  class ConnectionTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def testCheckClosed(self):
      assert not self.conn.check_closed()
  
    def testSessionAnon(self):
      ssn1 = self.conn.session()
      ssn2 = self.conn.session()
      self.ping(ssn1)
      self.ping(ssn2)
      assert ssn1 is not ssn2
  
    def testSessionNamed(self):
      ssn1 = self.conn.session("one")
      ssn2 = self.conn.session("two")
      self.ping(ssn1)
      self.ping(ssn2)
      assert ssn1 is not ssn2
      assert ssn1 is self.conn.session("one")
      assert ssn2 is self.conn.session("two")
  
    def testDetach(self):
      ssn = self.conn.session()
      self.ping(ssn)
      self.conn.detach()
      try:
        self.ping(ssn)
        assert False, "ping succeeded"
      except Detached:
        # this is the expected failure when pinging on a detached
        # connection
        pass
      self.conn.attach()
      self.ping(ssn)
  
    def testClose(self):
      self.conn.close()
      assert not self.conn.attached()
  
    def testSimultaneousClose(self):
      ssns = [self.conn.session() for i in range(3)]
      for s in ssns:
        for i in range(3):
          s.receiver("amq.topic")
          s.sender("amq.topic")
  
      def closer(errors):
        try:
          self.conn.close()
        except:
          _, e, _ = sys.exc_info()
          errors.append(compat.format_exc(e))
  
      t1_errors = []
      t2_errors = []
      t1 = Thread(target=lambda: closer(t1_errors))
      t2 = Thread(target=lambda: closer(t2_errors))
      t1.start()
      t2.start()
      t1.join(self.delay())
      t2.join(self.delay())
  
      assert not t1_errors, t1_errors[0]
      assert not t2_errors, t2_errors[0]
  
  class hangable:
  
    def __init__(self, conn, host, port):
      self.tcp = TRANSPORTS["tcp"](conn, host, port)
      self.hung = False
  
    def hang(self):
      self.hung = True
  
    def fileno(self):
      return self.tcp.fileno()
  
    def reading(self, reading):
      if self.hung:
        return True
      else:
        return self.tcp.reading(reading)
  
    def writing(self, writing):
      if self.hung:
        return False
      else:
        return self.tcp.writing(writing)
  
    def send(self, bytes):
      if self.hung:
        return 0
      else:
        return self.tcp.send(bytes)
  
    def recv(self, n):
      if self.hung:
        return ""
      else:
        return self.tcp.recv(n)
  
    def close(self):
      self.tcp.close()
  
  TRANSPORTS["hangable"] = hangable
  
  class TimeoutTests(Base):
  
    def setup_connection(self):
      options = self.connection_options()
      options["transport"] = "hangable"
      return Connection.establish(self.broker, **options)
  
    def setup_session(self):
      return self.conn.session()
  
    def setup_sender(self):
      return self.ssn.sender("amq.topic")
  
    def setup_receiver(self):
      return self.ssn.receiver("amq.topic; {link: {reliability: unreliable}}")
  
    def teardown_connection(self, conn):
      try:
        conn.detach(timeout=0)
      except Timeout:
        pass
  
    def hang(self):
      self.conn._driver._transport.hang()
  
    def timeoutTest(self, method):
      self.hang()
      try:
        method(timeout=self.delay())
        assert False, "did not time out"
      except Timeout:
        pass
  
    def testSenderSync(self):
      self.snd.send(self.content("testSenderSync"), sync=False)
      self.timeoutTest(self.snd.sync)
  
    def testSenderClose(self):
      self.snd.send(self.content("testSenderClose"), sync=False)
      self.timeoutTest(self.snd.close)
  
    def testReceiverClose(self):
      self.timeoutTest(self.rcv.close)
  
    def testSessionSync(self):
      self.snd.send(self.content("testSessionSync"), sync=False)
      self.timeoutTest(self.ssn.sync)
  
    def testSessionClose(self):
      self.timeoutTest(self.ssn.close)
  
    def testConnectionDetach(self):
      self.timeoutTest(self.conn.detach)
  
    def testConnectionClose(self):
      self.timeoutTest(self.conn.close)
  
    def testConnectionOpen(self):
      options = self.connection_options()
      options["reconnect"] = True
      options["reconnect_timeout"] = self.delay()
      try:
        bad_conn = Connection.establish("badhostname", **options)
        assert False, "did not time out"
      except Timeout:
        pass
  
  ACK_QC = 'test-ack-queue; {create: always}'
  ACK_QD = 'test-ack-queue; {delete: always}'
  
  class SessionTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def setup_session(self):
      return self.conn.session()
  
    def testSender(self):
      snd = self.ssn.sender('test-snd-queue; {create: sender, delete: receiver}',
                            durable=self.durable())
      snd2 = self.ssn.sender(snd.target, durable=self.durable())
      assert snd is not snd2
      snd2.close()
  
      content = self.content("testSender")
      snd.send(content)
      rcv = self.ssn.receiver(snd.target)
      msg = rcv.fetch(0)
      assert msg.content == content
      self.ssn.acknowledge(msg)
  
    def testReceiver(self):
      rcv = self.ssn.receiver('test-rcv-queue; {create: always}')
      rcv2 = self.ssn.receiver(rcv.source)
      assert rcv is not rcv2
      rcv2.close()
  
      content = self.content("testReceiver")
      snd = self.ssn.sender(rcv.source, durable=self.durable())
      snd.send(content)
      msg = rcv.fetch(0)
      assert msg.content == content
      self.ssn.acknowledge(msg)
      snd2 = self.ssn.receiver('test-rcv-queue; {delete: always}')
  
    def testDetachedReceiver(self):
      self.conn.detach()
      rcv = self.ssn.receiver("test-dis-rcv-queue; {create: always, delete: always}")
      m = self.content("testDetachedReceiver")
      self.conn.attach()
      snd = self.ssn.sender("test-dis-rcv-queue")
      snd.send(m)
      self.drain(rcv, expected=[m])
  
    def testNextReceiver(self):
      ADDR = 'test-next-rcv-queue; {create: always, delete: always}'
      rcv1 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
      rcv2 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
      rcv3 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
  
      snd = self.ssn.sender(ADDR)
  
      msgs = []
      for i in range(10):
        content = self.content("testNextReceiver", i)
        snd.send(content)
        msgs.append(content)
  
      fetched = []
      try:
        while True:
          rcv = self.ssn.next_receiver(timeout=self.delay())
          assert rcv in (rcv1, rcv2, rcv3)
          assert rcv.available() > 0
          fetched.append(rcv.fetch().content)
      except Empty:
        pass
      assert msgs == fetched, "expecting %s, got %s" % (msgs, fetched)
      self.ssn.acknowledge()
      #we set the capacity to 0 to prevent the deletion of the queue -
      #triggered the deletion policy when the first receiver is closed -
      #resulting in session exceptions being issued for the remaining
      #active subscriptions:
      for r in [rcv1, rcv2, rcv3]:
        r.capacity = 0
  
    # XXX, we need a convenient way to assert that required queues are
    # empty on setup, and possibly also to drain queues on teardown
    def ackTest(self, acker, ack_capacity=None):
      # send a bunch of messages
      snd = self.ssn.sender(ACK_QC, durable=self.durable())
      contents = [self.content("ackTest", i) for i in range(15)]
      for c in contents:
        snd.send(c)
  
      # drain the queue, verify the messages are there and then close
      # without acking
      rcv = self.ssn.receiver(ACK_QC)
      self.drain(rcv, expected=contents)
      self.ssn.close()
  
      # drain the queue again, verify that they are all the messages
      # were requeued, and ack this time before closing
      self.ssn = self.conn.session()
      if ack_capacity is not None:
        self.ssn.ack_capacity = ack_capacity
      rcv = self.ssn.receiver(ACK_QC)
      self.drain(rcv, expected=contents)
      acker(self.ssn)
      self.ssn.close()
  
      # drain the queue a final time and verify that the messages were
      # dequeued
      self.ssn = self.conn.session()
      rcv = self.ssn.receiver(ACK_QD)
      self.assertEmpty(rcv)
  
    def testAcknowledge(self):
      self.ackTest(lambda ssn: ssn.acknowledge())
  
    def testAcknowledgeAsync(self):
      self.ackTest(lambda ssn: ssn.acknowledge(sync=False))
  
    def testAcknowledgeAsyncAckCap0(self):
      try:
        try:
          self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 0)
          assert False, "acknowledge shouldn't succeed with ack_capacity of zero"
        except InsufficientCapacity:
          pass
      finally:
        self.ssn.ack_capacity = UNLIMITED
        self.drain(self.ssn.receiver(ACK_QD))
        self.ssn.acknowledge()
  
    def testAcknowledgeAsyncAckCap1(self):
      self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 1)
  
    def testAcknowledgeAsyncAckCap5(self):
      self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 5)
  
    def testAcknowledgeAsyncAckCapUNLIMITED(self):
      self.ackTest(lambda ssn: ssn.acknowledge(sync=False), UNLIMITED)
  
    def testRelease(self):
      msgs = [self.message("testRelease", i) for i in range(3)]
      snd = self.ssn.sender("test-release-queue; {create: always, delete: always}")
      for m in msgs:
        snd.send(m)
      rcv = self.ssn.receiver(snd.target)
      echos = self.drain(rcv, expected=msgs)
      self.ssn.acknowledge(echos[0])
      self.ssn.acknowledge(echos[1], Disposition(RELEASED, set_redelivered=True))
      self.ssn.acknowledge(echos[2], Disposition(RELEASED))
      self.drain(rcv, limit=1, expected=msgs[1:2], redelivered=True)
      self.drain(rcv, expected=msgs[2:3])
      self.ssn.acknowledge()
  
    def testReject(self):
      msgs = [self.message("testReject", i) for i in range(3)]
      snd = self.ssn.sender("""
        test-reject-queue; {
          create: always,
          delete: always,
          node: {
            x-declare: {
              alternate-exchange: 'amq.topic'
            }
          }
        }
  """)
      for m in msgs:
        snd.send(m)
      rcv = self.ssn.receiver(snd.target)
      rej = self.ssn.receiver("amq.topic")
      echos = self.drain(rcv, expected=msgs)
      self.ssn.acknowledge(echos[0])
      self.ssn.acknowledge(echos[1], Disposition(REJECTED))
      self.ssn.acknowledge(echos[2],
                           Disposition(REJECTED, code=0, text="test-reject"))
      self.drain(rej, expected=msgs[1:])
      self.ssn.acknowledge()
  
    def send(self, ssn, target, base, count=1):
      snd = ssn.sender(target, durable=self.durable())
      messages = []
      for i in range(count):
        c = self.message(base, i)
        snd.send(c)
        messages.append(c)
      snd.close()
      return messages
  
    def txTest(self, commit):
      TX_Q = 'test-tx-queue; {create: sender, delete: receiver}'
      TX_Q_COPY = 'test-tx-queue-copy; {create: always, delete: always}'
      txssn = self.conn.session(transactional=True)
      messages = self.send(self.ssn, TX_Q, "txTest", 3)
      txrcv = txssn.receiver(TX_Q)
      txsnd = txssn.sender(TX_Q_COPY, durable=self.durable())
      rcv = self.ssn.receiver(txrcv.source)
      copy_rcv = self.ssn.receiver(txsnd.target)
      self.assertEmpty(copy_rcv)
      for i in range(3):
        m = txrcv.fetch(0)
        txsnd.send(m)
        self.assertEmpty(copy_rcv)
      txssn.acknowledge()
      if commit:
        txssn.commit()
        self.assertEmpty(rcv)
        self.drain(copy_rcv, expected=messages)
      else:
        txssn.rollback()
        self.drain(rcv, expected=messages, redelivered=True)
        self.assertEmpty(copy_rcv)
      self.ssn.acknowledge()
  
    def testCommit(self):
      self.txTest(True)
  
    def testRollback(self):
      self.txTest(False)
  
    def txTestSend(self, commit):
      TX_SEND_Q = 'test-tx-send-queue; {create: sender, delete: receiver}'
      txssn = self.conn.session(transactional=True)
      messages = self.send(txssn, TX_SEND_Q, "txTestSend", 3)
      rcv = self.ssn.receiver(TX_SEND_Q)
      self.assertEmpty(rcv)
  
      if commit:
        txssn.commit()
        self.drain(rcv, expected=messages)
        self.ssn.acknowledge()
      else:
        txssn.rollback()
        self.assertEmpty(rcv)
        txssn.commit()
        self.assertEmpty(rcv)
  
    def testCommitSend(self):
      self.txTestSend(True)
  
    def testRollbackSend(self):
      self.txTestSend(False)
  
    def txTestAck(self, commit):
      TX_ACK_QC = 'test-tx-ack-queue; {create: always}'
      TX_ACK_QD = 'test-tx-ack-queue; {delete: always}'
      txssn = self.conn.session(transactional=True)
      txrcv = txssn.receiver(TX_ACK_QC)
      self.assertEmpty(txrcv)
      messages = self.send(self.ssn, TX_ACK_QC, "txTestAck", 3)
      self.drain(txrcv, expected=messages)
  
      if commit:
        txssn.acknowledge()
      else:
        txssn.rollback()
        self.drain(txrcv, expected=messages, redelivered=True)
        txssn.acknowledge()
        txssn.rollback()
        self.drain(txrcv, expected=messages, redelivered=True)
        txssn.commit() # commit without ack
        self.assertEmpty(txrcv)
  
      txssn.close()
  
      txssn = self.conn.session(transactional=True)
      txrcv = txssn.receiver(TX_ACK_QC)
      self.drain(txrcv, expected=messages, redelivered=True)
      txssn.acknowledge()
      txssn.commit()
      rcv = self.ssn.receiver(TX_ACK_QD)
      self.assertEmpty(rcv)
      txssn.close()
      self.assertEmpty(rcv)
  
    def testCommitAck(self):
      self.txTestAck(True)
  
    def testRollbackAck(self):
      self.txTestAck(False)
  
    def testDoubleCommit(self):
      ssn = self.conn.session(transactional=True)
      snd = ssn.sender("amq.direct/doubleCommit")
      rcv = ssn.receiver("amq.direct/doubleCommit")
      msgs = [self.message("testDoubleCommit", i, subject="doubleCommit") for i in range(3)]
      for m in msgs:
        snd.send(m)
      ssn.commit()
      self.drain(rcv, expected=msgs)
      ssn.acknowledge()
      ssn.commit()
  
    def testClose(self):
      self.ssn.close()
      try:
        self.ping(self.ssn)
        assert False, "ping succeeded"
      except Detached:
        pass
  
    def testRxCallback(self):
      """Verify that the callback is invoked when a message is received.
      """
      ADDR = 'test-rx_callback-queue; {create: always, delete: receiver}'
      class CallbackHandler:
          def __init__(self):
              self.handler_called = False
          def __call__(self, ssn):
              self.handler_called = True
              self.ssn = ssn
      cb = CallbackHandler()
      self.ssn.set_message_received_notify_handler(cb)
      rcv = self.ssn.receiver(ADDR)
      rcv.capacity = UNLIMITED
      snd = self.ssn.sender(ADDR)
      assert not cb.handler_called
      snd.send("Ping")
      deadline = time.time() + self.timeout()
      while time.time() < deadline:
          if cb.handler_called:
              break;
      assert cb.handler_called
      assert cb.ssn == self.ssn
      snd.close()
      rcv.close()
  
  
  RECEIVER_Q = 'test-receiver-queue; {create: always, delete: always}'
  
  class ReceiverTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def setup_session(self):
      return self.conn.session()
  
    def setup_sender(self):
      return self.ssn.sender(RECEIVER_Q)
  
    def setup_receiver(self):
      return self.ssn.receiver(RECEIVER_Q)
  
    def send(self, base, count = None, sync=True):
      content = self.content(base, count)
      self.snd.send(content, sync=sync)
      return content
  
    def testFetch(self):
      try:
        msg = self.rcv.fetch(0)
        assert False, "unexpected message: %s" % msg
      except Empty:
        pass
      try:
        start = time.time()
        msg = self.rcv.fetch(self.delay())
        assert False, "unexpected message: %s" % msg
      except Empty:
        elapsed = time.time() - start
        assert elapsed >= self.delay()
  
      one = self.send("testFetch", 1)
      two = self.send("testFetch", 2)
      three = self.send("testFetch", 3)
      msg = self.rcv.fetch(0)
      assert msg.content == one
      msg = self.rcv.fetch(self.delay())
      assert msg.content == two
      msg = self.rcv.fetch()
      assert msg.content == three
      self.ssn.acknowledge()
  
    def fetchFromClosedTest(self, entry):
      entry.close()
      try:
        msg = self.rcv.fetch(0)
        assert False, "unexpected result: %s" % msg
      except Empty, e:
        assert False, "unexpected exception: %s" % e
      except LinkClosed, e:
        pass
  
    def testFetchFromClosedReceiver(self):
      self.fetchFromClosedTest(self.rcv)
  
    def testFetchFromClosedSession(self):
      self.fetchFromClosedTest(self.ssn)
  
    def testFetchFromClosedConnection(self):
      self.fetchFromClosedTest(self.conn)
  
    def fetchFromConcurrentCloseTest(self, entry):
      def closer():
        self.sleep()
        entry.close()
      t = Thread(target=closer)
      t.start()
      try:
        msg = self.rcv.fetch()
        assert False, "unexpected result: %s" % msg
      except Empty, e:
        assert False, "unexpected exception: %s" % e
      except LinkClosed, e:
        pass
      t.join()
  
    def testFetchFromConcurrentCloseReceiver(self):
      self.fetchFromConcurrentCloseTest(self.rcv)
  
    def testFetchFromConcurrentCloseSession(self):
      self.fetchFromConcurrentCloseTest(self.ssn)
  
    def testFetchFromConcurrentCloseConnection(self):
      self.fetchFromConcurrentCloseTest(self.conn)
  
    def testCapacityIncrease(self):
      content = self.send("testCapacityIncrease")
      self.sleep()
      assert self.rcv.available() == 0
      self.rcv.capacity = UNLIMITED
      self.sleep()
      assert self.rcv.available() == 1
      msg = self.rcv.fetch(0)
      assert msg.content == content
      assert self.rcv.available() == 0
      self.ssn.acknowledge()
  
    def testCapacityDecrease(self):
      self.rcv.capacity = UNLIMITED
      one = self.send("testCapacityDecrease", 1)
      self.sleep()
      assert self.rcv.available() == 1
      msg = self.rcv.fetch(0)
      assert msg.content == one
  
      self.rcv.capacity = 0
  
      two = self.send("testCapacityDecrease", 2)
      self.sleep()
      assert self.rcv.available() == 0
      msg = self.rcv.fetch(0)
      assert msg.content == two
  
      self.ssn.acknowledge()
  
    def capacityTest(self, capacity, threshold=None):
      if threshold is not None:
        self.rcv.threshold = threshold
      self.rcv.capacity = capacity
      self.assertAvailable(self.rcv, 0)
  
      for i in range(2*capacity):
        self.send("capacityTest(%s, %s)" % (capacity, threshold), i, sync=False)
      self.snd.sync()
      self.sleep()
      self.assertAvailable(self.rcv)
  
      first = capacity/2
      second = capacity - first
      self.drain(self.rcv, limit = first)
      self.sleep()
      self.assertAvailable(self.rcv)
      self.drain(self.rcv, limit = second)
      self.sleep()
      self.assertAvailable(self.rcv)
  
      drained = self.drain(self.rcv)
      assert len(drained) == capacity, "%s, %s" % (len(drained), drained)
      self.assertAvailable(self.rcv, 0)
  
      self.ssn.acknowledge()
  
    def testCapacity5(self):
      self.capacityTest(5)
  
    def testCapacity5Threshold1(self):
      self.capacityTest(5, 1)
  
    def testCapacity10(self):
      self.capacityTest(10)
  
    def testCapacity10Threshold1(self):
      self.capacityTest(10, 1)
  
    def testCapacity100(self):
      self.capacityTest(100)
  
    def testCapacity100Threshold1(self):
      self.capacityTest(100, 1)
  
    def testCapacityUNLIMITED(self):
      self.rcv.capacity = UNLIMITED
      self.assertAvailable(self.rcv, 0)
  
      for i in range(10):
        self.send("testCapacityUNLIMITED", i)
      self.sleep()
      self.assertAvailable(self.rcv, 10)
  
      self.drain(self.rcv)
      self.assertAvailable(self.rcv, 0)
  
      self.ssn.acknowledge()
  
    def testAvailable(self):
      self.rcv.capacity = UNLIMITED
      assert self.rcv.available() == 0
  
      for i in range(3):
        self.send("testAvailable", i)
      self.sleep()
      assert self.rcv.available() == 3
  
      for i in range(3, 10):
        self.send("testAvailable", i)
      self.sleep()
      assert self.rcv.available() == 10
  
      self.drain(self.rcv, limit=3)
      assert self.rcv.available() == 7
  
      self.drain(self.rcv)
      assert self.rcv.available() == 0
  
      self.ssn.acknowledge()
  
    def testDoubleClose(self):
      m1 = self.content("testDoubleClose", 1)
      m2 = self.content("testDoubleClose", 2)
  
      snd = self.ssn.sender("""test-double-close; {
    create: always,
    delete: sender,
    node: {
      type: topic
    }
  }
  """)
      r1 = self.ssn.receiver(snd.target)
      r2 = self.ssn.receiver(snd.target)
      snd.send(m1)
      self.drain(r1, expected=[m1])
      self.drain(r2, expected=[m1])
      r1.close()
      snd.send(m2)
      self.drain(r2, expected=[m2])
      r2.close()
  
    # XXX: need testClose
  
    def testMode(self):
      msgs = [self.content("testMode", 1),
              self.content("testMode", 2),
              self.content("testMode", 3)]
  
      for m in msgs:
        self.snd.send(m)
  
      rb = self.ssn.receiver('test-receiver-queue; {mode: browse}')
      rc = self.ssn.receiver('test-receiver-queue; {mode: consume}')
      self.drain(rb, expected=msgs)
      self.drain(rc, expected=msgs)
      rb2 = self.ssn.receiver(rb.source)
      self.assertEmpty(rb2)
      self.drain(self.rcv, expected=[])
  
    def testUnsettled(self):
      # just tests the code path and not the value
      rcv = self.ssn.receiver('test-receiver-unsettled-queue; {create: always, delete: always}')
      rcv.unsettled()
  
    def unreliabilityTest(self, mode="unreliable"):
      msgs = [self.message("testUnreliable", i) for i in range(3)]
      snd = self.ssn.sender("test-unreliability-queue; {create: sender, delete: receiver}")
      rcv = self.ssn.receiver(snd.target)
      for m in msgs:
        snd.send(m)
  
      # close without ack on reliable receiver, messages should be requeued
      ssn = self.conn.session()
      rrcv = ssn.receiver("test-unreliability-queue")
      self.drain(rrcv, expected=msgs)
      ssn.close()
  
      # close without ack on unreliable receiver, messages should not be requeued
      ssn = self.conn.session()
      urcv = ssn.receiver("test-unreliability-queue; {link: {reliability: %s}}" % mode)
      self.drain(urcv, expected=msgs, redelivered=True)
      ssn.close()
  
      self.assertEmpty(rcv)
  
    def testUnreliable(self):
      self.unreliabilityTest(mode="unreliable")
  
    def testAtMostOnce(self):
      self.unreliabilityTest(mode="at-most-once")
  
  class AddressTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def setup_session(self):
      return self.conn.session()
  
    def badOption(self, options, error):
      try:
        self.ssn.sender("test-bad-options-snd; %s" % options)
        assert False
      except InvalidOption, e:
        assert "error in options: %s" % error == str(e), e
  
      try:
        self.ssn.receiver("test-bad-options-rcv; %s" % options)
        assert False
      except InvalidOption, e:
        assert "error in options: %s" % error == str(e), e
  
    def testIllegalKey(self):
      self.badOption("{create: always, node: "
                     "{this-property-does-not-exist: 3}}",
                     "node: this-property-does-not-exist: "
                     "illegal key")
  
    def testWrongValue(self):
      self.badOption("{create: asdf}", "create: asdf not in "
                     "('always', 'sender', 'receiver', 'never')")
  
    def testWrongType1(self):
      self.badOption("{node: asdf}",
                     "node: asdf is not a map")
  
    def testWrongType2(self):
      self.badOption("{node: {durable: []}}",
                     "node: durable: [] is not a bool")
  
    def testCreateQueue(self):
      snd = self.ssn.sender("test-create-queue; {create: always, delete: always, "
                            "node: {type: queue, durable: False, "
                            "x-declare: {auto_delete: true}}}")
      content = self.content("testCreateQueue")
      snd.send(content)
      rcv = self.ssn.receiver("test-create-queue")
      self.drain(rcv, expected=[content])
  
    def createExchangeTest(self, props=""):
      addr = """test-create-exchange; {
                  create: always,
                  delete: always,
                  node: {
                    type: topic,
                    durable: False,
                    x-declare: {auto_delete: true, %s}
                  }
                }""" % props
      snd = self.ssn.sender(addr)
      snd.send("ping")
      rcv1 = self.ssn.receiver("test-create-exchange/first")
      rcv2 = self.ssn.receiver("test-create-exchange/first")
      rcv3 = self.ssn.receiver("test-create-exchange/second")
      for r in (rcv1, rcv2, rcv3):
        try:
          r.fetch(0)
          assert False
        except Empty:
          pass
      msg1 = Message(self.content("testCreateExchange", 1), subject="first")
      msg2 = Message(self.content("testCreateExchange", 2), subject="second")
      snd.send(msg1)
      snd.send(msg2)
      self.drain(rcv1, expected=[msg1.content])
      self.drain(rcv2, expected=[msg1.content])
      self.drain(rcv3, expected=[msg2.content])
  
    def testCreateExchange(self):
      self.createExchangeTest()
  
    def testCreateExchangeDirect(self):
      self.createExchangeTest("type: direct")
  
    def testCreateExchangeTopic(self):
      self.createExchangeTest("type: topic")
  
    def testDeleteBySender(self):
      snd = self.ssn.sender("test-delete; {create: always}")
      snd.send("ping")
      snd.close()
      snd = self.ssn.sender("test-delete; {delete: always}")
      snd.send("ping")
      snd.close()
      try:
        self.ssn.sender("test-delete")
      except NotFound, e:
        assert "no such queue" in str(e)
  
    def testDeleteByReceiver(self):
      rcv = self.ssn.receiver("test-delete; {create: always, delete: always}")
      try:
        rcv.fetch(0)
      except Empty:
        pass
      rcv.close()
  
      try:
        self.ssn.receiver("test-delete")
        assert False
      except NotFound, e:
        assert "no such queue" in str(e)
  
    def testDeleteSpecial(self):
      snd = self.ssn.sender("amq.topic; {delete: always}")
      snd.send("asdf")
      try:
        snd.close()
        assert False, "successfully deleted amq.topic"
      except SessionError, e:
        assert e.code == 530
      # XXX: need to figure out close after error
      self.conn._remove_session(self.ssn)
  
    def testNodeBindingsQueue(self):
      snd = self.ssn.sender("""
  test-node-bindings-queue; {
    create: always,
    delete: always,
    node: {
      x-bindings: [{exchange: "amq.topic", key: "a.#"},
                   {exchange: "amq.direct", key: "b"},
                   {exchange: "amq.topic", key: "c.*"}]
    }
  }
  """)
      snd.send("one")
      snd_a = self.ssn.sender("amq.topic/a.foo")
      snd_b = self.ssn.sender("amq.direct/b")
      snd_c = self.ssn.sender("amq.topic/c.bar")
      snd_a.send("two")
      snd_b.send("three")
      snd_c.send("four")
      rcv = self.ssn.receiver("test-node-bindings-queue")
      self.drain(rcv, expected=["one", "two", "three", "four"])
  
    def testNodeBindingsTopic(self):
      rcv = self.ssn.receiver("test-node-bindings-topic-queue; {create: always, delete: always}")
      rcv_a = self.ssn.receiver("test-node-bindings-topic-queue-a; {create: always, delete: always}")
      rcv_b = self.ssn.receiver("test-node-bindings-topic-queue-b; {create: always, delete: always}")
      rcv_c = self.ssn.receiver("test-node-bindings-topic-queue-c; {create: always, delete: always}")
      snd = self.ssn.sender("""
  test-node-bindings-topic; {
    create: always,
    delete: always,
    node: {
      type: topic,
      x-bindings: [{queue: test-node-bindings-topic-queue, key: "#"},
                   {queue: test-node-bindings-topic-queue-a, key: "a.#"},
                   {queue: test-node-bindings-topic-queue-b, key: "b"},
                   {queue: test-node-bindings-topic-queue-c, key: "c.*"}]
    }
  }
  """)
      m1 = Message("one")
      m2 = Message(subject="a.foo", content="two")
      m3 = Message(subject="b", content="three")
      m4 = Message(subject="c.bar", content="four")
      snd.send(m1)
      snd.send(m2)
      snd.send(m3)
      snd.send(m4)
      self.drain(rcv, expected=[m1, m2, m3, m4])
      self.drain(rcv_a, expected=[m2])
      self.drain(rcv_b, expected=[m3])
      self.drain(rcv_c, expected=[m4])
  
    def testLinkBindings(self):
      m_a = self.message("testLinkBindings", 1, subject="a")
      m_b = self.message("testLinkBindings", 2, subject="b")
  
      self.ssn.sender("test-link-bindings-queue; {create: always, delete: always}")
      snd = self.ssn.sender("amq.topic")
  
      snd.send(m_a)
      snd.send(m_b)
      snd.close()
  
      rcv = self.ssn.receiver("test-link-bindings-queue")
      self.assertEmpty(rcv)
  
      snd = self.ssn.sender("""
  amq.topic; {
    link: {
      x-bindings: [{queue: test-link-bindings-queue, key: a}]
    }
  }
  """)
  
      snd.send(m_a)
      snd.send(m_b)
  
      self.drain(rcv, expected=[m_a])
      rcv.close()
  
      rcv = self.ssn.receiver("""
  test-link-bindings-queue; {
    link: {
      x-bindings: [{exchange: "amq.topic", key: b}]
    }
  }
  """)
  
      snd.send(m_a)
      snd.send(m_b)
  
      self.drain(rcv, expected=[m_a, m_b])
  
    def testSubjectOverride(self):
      snd = self.ssn.sender("amq.topic/a")
      rcv_a = self.ssn.receiver("amq.topic/a")
      rcv_b = self.ssn.receiver("amq.topic/b")
      m1 = self.content("testSubjectOverride", 1)
      m2 = self.content("testSubjectOverride", 2)
      snd.send(m1)
      snd.send(Message(subject="b", content=m2))
      self.drain(rcv_a, expected=[m1])
      self.drain(rcv_b, expected=[m2])
  
    def testSubjectDefault(self):
      m1 = self.content("testSubjectDefault", 1)
      m2 = self.content("testSubjectDefault", 2)
      snd = self.ssn.sender("amq.topic/a")
      rcv = self.ssn.receiver("amq.topic")
      snd.send(m1)
      snd.send(Message(subject="b", content=m2))
      e1 = rcv.fetch(timeout=0)
      e2 = rcv.fetch(timeout=0)
      assert e1.subject == "a", "subject: %s" % e1.subject
      assert e2.subject == "b", "subject: %s" % e2.subject
      self.assertEmpty(rcv)
  
    def doReliabilityTest(self, reliability, messages, expected):
      snd = self.ssn.sender("amq.topic")
      rcv = self.ssn.receiver("amq.topic; {link: {reliability: %s}}" % reliability)
      for m in messages:
        snd.send(m)
      self.conn.detach()
      self.conn.attach()
      self.drain(rcv, expected=expected)
  
    def testReliabilityUnreliable(self):
      msgs = [self.message("testReliabilityUnreliable", i) for i in range(3)]
      self.doReliabilityTest("unreliable", msgs, [])
  
    def testReliabilityAtLeastOnce(self):
      msgs = [self.message("testReliabilityAtLeastOnce", i) for i in range(3)]
      self.doReliabilityTest("at-least-once", msgs, msgs)
  
    def testLinkName(self):
      msgs = [self.message("testLinkName", i) for i in range(3)]
      snd = self.ssn.sender("amq.topic")
      trcv = self.ssn.receiver("amq.topic; {link: {name: test-link-name}}")
      qrcv = self.ssn.receiver("test-link-name")
      for m in msgs:
        snd.send(m)
      self.drain(qrcv, expected=msgs)
  
    def testAssert1(self):
      try:
        snd = self.ssn.sender("amq.topic; {assert: always, node: {type: queue}}")
        assert 0, "assertion failed to trigger"
      except AssertionFailed, e:
        pass
      except NotFound, e:  # queue named amp.topic not found
        pass
  
    def testAssert2(self):
      snd = self.ssn.sender("amq.topic; {assert: always}")
  
  NOSUCH_Q = "this-queue-should-not-exist"
  UNPARSEABLE_ADDR = "name/subject; {bad options"
  UNLEXABLE_ADDR = "\0x0\0x1\0x2\0x3"
  
  class AddressErrorTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def setup_session(self):
      return self.conn.session()
  
    def senderErrorTest(self, addr, exc, check=lambda e: True):
      try:
        self.ssn.sender(addr, durable=self.durable())
        assert False, "sender creation succeeded"
      except exc, e:
        assert check(e), "unexpected error: %s" % compat.format_exc(e)
  
    def receiverErrorTest(self, addr, exc, check=lambda e: True):
      try:
        self.ssn.receiver(addr)
        assert False, "receiver creation succeeded"
      except exc, e:
        assert check(e), "unexpected error: %s" % compat.format_exc(e)
  
    def testNoneTarget(self):
      self.senderErrorTest(None, MalformedAddress)
  
    def testNoneSource(self):
      self.receiverErrorTest(None, MalformedAddress)
  
    def testNoTarget(self):
      self.senderErrorTest(NOSUCH_Q, NotFound, lambda e: NOSUCH_Q in str(e))
  
    def testNoSource(self):
      self.receiverErrorTest(NOSUCH_Q, NotFound, lambda e: NOSUCH_Q in str(e))
  
    def testUnparseableTarget(self):
      self.senderErrorTest(UNPARSEABLE_ADDR, MalformedAddress,
                           lambda e: "expecting COLON" in str(e))
  
    def testUnparseableSource(self):
      self.receiverErrorTest(UNPARSEABLE_ADDR, MalformedAddress,
                             lambda e: "expecting COLON" in str(e))
  
    def testUnlexableTarget(self):
      self.senderErrorTest(UNLEXABLE_ADDR, MalformedAddress,
                           lambda e: "unrecognized characters" in str(e))
  
    def testUnlexableSource(self):
      self.receiverErrorTest(UNLEXABLE_ADDR, MalformedAddress,
                             lambda e: "unrecognized characters" in str(e))
  
    def testInvalidMode(self):
      self.receiverErrorTest('name; {mode: "this-is-a-bad-receiver-mode"}',
                             InvalidOption,
                             lambda e: "not in ('browse', 'consume')" in str(e))
  
  SENDER_Q = 'test-sender-q; {create: always, delete: always}'
  
  class SenderTests(Base):
  
    def setup_connection(self):
      return Connection.establish(self.broker, **self.connection_options())
  
    def setup_session(self):
      return self.conn.session()
  
    def setup_sender(self):
      return self.ssn.sender(SENDER_Q)
  
    def setup_receiver(self):
      return self.ssn.receiver(SENDER_Q)
  
    def checkContent(self, content):
      self.snd.send(content)
      msg = self.rcv.fetch(0)
      assert msg.content == content
  
      out = Message(content)
      self.snd.send(out)
      echo = self.rcv.fetch(0)
      assert out.content == echo.content
      assert echo.content == msg.content
      self.ssn.acknowledge()
  
    def testSendString(self):
      self.checkContent(self.content("testSendString"))
  
    def testSendList(self):
      self.checkContent(["testSendList", 1, 3.14, self.test_id])
  
    def testSendMap(self):
      self.checkContent({"testSendMap": self.test_id, "pie": "blueberry", "pi": 3.14})
  
    def asyncTest(self, capacity):
      self.snd.capacity = capacity
      msgs = [self.content("asyncTest", i) for i in range(15)]
      for m in msgs:
        self.snd.send(m, sync=False)
      self.drain(self.rcv, timeout=self.delay(), expected=msgs)
      self.ssn.acknowledge()
  
    def testSendAsyncCapacity0(self):
      try:
        self.asyncTest(0)
        assert False, "send shouldn't succeed with zero capacity"
      except InsufficientCapacity:
        # this is expected
        pass
  
    def testSendAsyncCapacity1(self):
      self.asyncTest(1)
  
    def testSendAsyncCapacity5(self):
      self.asyncTest(5)
  
    def testSendAsyncCapacityUNLIMITED(self):
      self.asyncTest(UNLIMITED)
  
    def testCapacityTimeout(self):
      self.snd.capacity = 1
      msgs = []
      caught = False
      while len(msgs) < 100:
        m = self.content("testCapacity", len(msgs))
        try:
          self.snd.send(m, sync=False, timeout=0)
          msgs.append(m)
        except InsufficientCapacity:
          caught = True
          break
      self.snd.sync()
      self.drain(self.rcv, expected=msgs)
      self.ssn.acknowledge()
      assert caught, "did not exceed capacity"
  
    def testEINTR(self):
      m1 = self.content("testEINTR", 0)
      m2 = self.content("testEINTR", 1)
  
      self.snd.send(m1, timeout=self.timeout())
      try:
        os.setuid(500)
        assert False, "setuid should fail"
      except:
        pass
      self.snd.send(m2, timeout=self.timeout())
  
  
  class ErrorCallbackTests(Base):
  
    class Callback:
      def __init__(self, name):
        self.name = name
        self.obj = None
        self.exc = None
  
      def __call__(self, obj, exc):
        self.obj = obj
        self.exc = exc
  
    def testConnectErrorCallback(self):
      cb = ErrorCallbackTests.Callback("connection")
      self.conn = Connection("localhost:4")
      self.conn.set_async_exception_notify_handler(cb)
      try:
        self.conn.open()
        assert False, "connect succeeded"
      except Exception:
        assert self.conn == cb.obj, cb.obj
        assert cb.name == "connection"
        assert cb.exc is not None
  
    def testSessionErrorCallback(self):
      ccb = ErrorCallbackTests.Callback("connection")
      self.conn = Connection.establish(self.broker, **self.connection_options())
      self.conn.set_async_exception_notify_handler(ccb)
      scb = ErrorCallbackTests.Callback("session")
      self.ssn = self.conn.session(transactional=True)
      self.ssn.set_async_exception_notify_handler(scb)
      self.conn.detach()
      try:
        self.ping(self.ssn)
        assert False, "session succeeded"
      except Exception:
        assert self.ssn == scb.obj, scb.obj
        assert scb.name == "session"
        assert scb.exc is not None
        # connection callback should be empty
        assert ccb.obj == None, ccb.obj
  
    def testSenderErrorCallback(self):
      ccb = ErrorCallbackTests.Callback("connection")
      conn = Connection(self.broker, **self.connection_options())
      conn.set_async_exception_notify_handler(ccb)
      scb = ErrorCallbackTests.Callback("session")
      ssn = conn.session()
      ssn.set_async_exception_notify_handler(scb)
      snd = ssn.sender(NOSUCH_Q)
      sndcb = ErrorCallbackTests.Callback("sender")
      snd.set_async_exception_notify_handler(sndcb)
      conn.open()
      try:
          snd.send(self.message("HI"))
          assert False, "send worked"
      except Exception:
          assert snd == sndcb.obj, sndcb.obj
          assert sndcb.name == "sender"
          assert sndcb.exc is not None
          # connection and session callbacks are empty
          assert ccb.obj == None, ccb.obj
          assert scb.obj == None, scb.obj
  
    def testReceiverErrorCallback(self):
      ccb = ErrorCallbackTests.Callback("connection")
      self.conn = Connection(self.broker, **self.connection_options())
      self.conn.set_async_exception_notify_handler(ccb)
      scb = ErrorCallbackTests.Callback("session")
      self.ssn = self.conn.session()
      self.ssn.set_async_exception_notify_handler(scb)
      self.recv = self.ssn.receiver(NOSUCH_Q)
      rcb = ErrorCallbackTests.Callback("receiver")
      self.recv.set_async_exception_notify_handler(rcb)
      self.conn.open()
      try:
          self.recv.fetch()
          assert False, "fetch worked"
      except Exception:
          assert self.recv == rcb.obj, rcb.obj
          assert rcb.name == "receiver"
          assert rcb.exc is not None
          # connection and session callbacks are empty
          assert ccb.obj == None, ccb.obj
          assert scb.obj == None, scb.obj
  