def test_timestamp_config()

in qpid_tests/broker_0_10/management.py [0:0]

• 44 lines of code
• 8 McCabe index (conditional complexity)

    def test_timestamp_config(self):
        """
        Test message timestamping control.
        """
        self.startQmf()
        conn = self.connect()
        session = conn.session("timestamp-session")

        #verify that receive message timestamping is OFF by default
        broker = self.qmf.getObjects(_class="broker")[0]
        rc = broker.getTimestampConfig()
        self.assertEqual(rc.status, 0)
        self.assertEqual(rc.text, "OK")

        #try to enable it
        rc = broker.setTimestampConfig(True)
        self.assertEqual(rc.status, 0)
        self.assertEqual(rc.text, "OK")

        rc = broker.getTimestampConfig()
        self.assertEqual(rc.status, 0)
        self.assertEqual(rc.text, "OK")
        self.assertEqual(rc.receive, True)

        # setup a connection & session to the broker
        url = "%s://%s:%d" % (self.broker.scheme or "amqp", self.broker.host or "localhost", self.broker.port or 5672)
        conn = qpid.messaging.Connection(url)
        conn.open()
        sess = conn.session()

        #send a message to a queue
        sender = sess.sender("ts-q; {create:sender, delete:receiver}")
        sender.send( qpid.messaging.Message(content="abc") )

        #receive message from queue, and verify timestamp is present
        receiver = sess.receiver("ts-q")
        try:
            msg = receiver.fetch(timeout=1)
        except Empty:
            assert(False)
        self.assertEqual("abc", msg.content)
        self.assertEqual(True, "x-amqp-0-10.timestamp" in msg.properties)
        assert(msg.properties["x-amqp-0-10.timestamp"])

        #try to disable it
        rc = broker.setTimestampConfig(False)
        self.assertEqual(rc.status, 0)
        self.assertEqual(rc.text, "OK")

        rc = broker.getTimestampConfig()
        self.assertEqual(rc.status, 0)
        self.assertEqual(rc.text, "OK")
        self.assertEqual(rc.receive, False)

        #send another message to the queue
        sender.send( qpid.messaging.Message(content="def") )

        #receive message from queue, and verify timestamp is NOT PRESENT
        receiver = sess.receiver("ts-q")
        try:
            msg = receiver.fetch(timeout=1)
        except Empty:
            assert(False)
        self.assertEqual("def", msg.content)
        self.assertEqual(False, "x-amqp-0-10.timestamp" in msg.properties)