tests/system_tests_drain.py

#
# 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 system_test import TestCase, Qdrouterd, main_module
from system_tests_drain_support import DrainMessagesHandler, DrainOneMessageHandler
from system_tests_drain_support import DrainNoMessagesHandler, DrainNoMoreMessagesHandler

class DrainSupportTest(TestCase):

    @classmethod
    def setUpClass(cls):
        """Start a router and a messenger"""
        super(DrainSupportTest, cls).setUpClass()
        name = "test-router"
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'QDR'}),

            # Setting the linkCapacity to 10 will allow the sender to send a burst of 10 messages
            ('listener', {'port': cls.tester.get_port(), 'linkCapacity': 10}),

        ])

        cls.router = cls.tester.qdrouterd(name, config)
        cls.router.wait_ready()
        cls.address = cls.router.addresses[0]

    def test_drain_support_1_all_messages(self):
        drain_support = DrainMessagesHandler(self.address)
        drain_support.run()
        self.assertEqual(drain_support.error, None)

    def test_drain_support_2_one_message(self):
        drain_support = DrainOneMessageHandler(self.address)
        drain_support.run()
        self.assertEqual(drain_support.error, None)

    def test_drain_support_3_no_messages(self):
        drain_support = DrainNoMessagesHandler(self.address)
        drain_support.run()
        self.assertEqual(drain_support.error, None)

    def test_drain_support_4_no_more_messages(self):
        drain_support = DrainNoMoreMessagesHandler(self.address)
        drain_support.run()
        self.assertEqual(drain_support.error, None)

if __name__ == '__main__':
    unittest.main(main_module())