| # |
| # 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 __future__ import unicode_literals |
| from __future__ import division |
| from __future__ import absolute_import |
| from __future__ import print_function |
| |
| import os |
| import sys |
| from time import sleep |
| from signal import SIGINT |
| from subprocess import PIPE |
| |
| from proton import Message |
| |
| from system_test import TestCase |
| from system_test import Qdrouterd |
| from system_test import main_module |
| from system_test import TIMEOUT |
| from system_test import Process |
| from system_test import AsyncTestSender |
| from system_test import unittest |
| |
| |
| class ThreeRouterTest(TestCase): |
| |
| @classmethod |
| def setUpClass(cls): |
| """ |
| Create a mesh of three routers. Reject any links or messages sent to |
| an unavailable address. |
| """ |
| super(ThreeRouterTest, cls).setUpClass() |
| |
| def router(name, extra_config): |
| config = [ |
| |
| ('router', {'id': name, |
| 'mode': 'interior', |
| "defaultDistribution": "unavailable"}), |
| ('listener', {'role': 'normal', |
| 'port': cls.tester.get_port(), |
| "linkCapacity": '100'}), |
| |
| ('address', {'prefix': 'closest', 'distribution': 'closest'}), |
| ('address', {'prefix': 'balanced', 'distribution': 'balanced'}), |
| ('address', {'prefix': 'multicast', 'distribution': 'multicast'}), |
| ] + extra_config |
| |
| config = Qdrouterd.Config(config) |
| |
| cls.routers.append(cls.tester.qdrouterd(name, config, wait=False)) |
| |
| cls.routers = [] |
| |
| inter_router_A = cls.tester.get_port() |
| inter_router_B = cls.tester.get_port() |
| inter_router_C = cls.tester.get_port() |
| |
| router('RouterA', |
| [('listener', {'role': 'inter-router', 'port': inter_router_A}), |
| ('connector', {'role': 'inter-router', 'port': inter_router_B})]) |
| |
| router('RouterB', |
| [('listener', {'role': 'inter-router', 'port': inter_router_B}), |
| ('connector', {'role': 'inter-router', 'port': inter_router_C})]) |
| |
| router('RouterC', |
| [('listener', {'role': 'inter-router', 'port': inter_router_C}), |
| ('connector', {'role': 'inter-router', 'port': inter_router_A})]) |
| |
| cls.RouterA = cls.routers[0] |
| cls.RouterB = cls.routers[1] |
| cls.RouterC = cls.routers[2] |
| |
| cls.RouterA.wait_router_connected('RouterB') |
| cls.RouterA.wait_router_connected('RouterC') |
| cls.RouterB.wait_router_connected('RouterA') |
| cls.RouterB.wait_router_connected('RouterC') |
| cls.RouterC.wait_router_connected('RouterA') |
| cls.RouterC.wait_router_connected('RouterB') |
| |
| def server_address(self, router): |
| return router.addresses[0] |
| |
| def server_port(self, router): |
| return router.ports[0] # first listener is for client connection |
| |
| def server_host(self, router): |
| fam = router.ports_family |
| return router.get_host(fam.get(self.server_port(router), |
| "IPv4")) |
| |
| def spawn_receiver(self, router, count, address, *extra_args): |
| cmd = ["test-receiver", |
| "-a", "%s:%s" % (self.server_host(router), |
| self.server_port(router)), |
| "-c", str(count), "-s", address] + list(extra_args) |
| # env = dict(os.environ, PN_TRACE_FRM="1") |
| # return self.popen(cmd, expect=Process.EXIT_OK, env=env) |
| return self.popen(cmd, expect=Process.EXIT_OK) |
| |
| def spawn_sender(self, router, count, address, *extra_args): |
| cmd = ["test-sender", |
| "-a", "%s:%s" % (self.server_host(router), |
| self.server_port(router)), |
| "-c", str(count), "-t", address] + list(extra_args) |
| # env = dict(os.environ, PN_TRACE_FRM="1") |
| # return self.popen(cmd, expect=Process.EXIT_OK, env=env) |
| return self.popen(cmd, expect=Process.EXIT_OK) |
| |
| def _rx_failover(self, extra_tx_args=None, extra_rx_args=None): |
| # Have a single sender transmit unsettled as fast as possible |
| # non-stop. Have a single receiver that consumes a small number of |
| # messages before failing over to a different router in the mesh |
| extra_tx = extra_tx_args or [] |
| extra_rx = extra_rx_args or [] |
| total = 100 |
| router_index = 0 |
| tx = self.spawn_sender(self.RouterC, 0, "balanced/foo", *extra_tx) |
| while total > 0: |
| rx = self.spawn_receiver(self.routers[router_index], 5, |
| "balanced/foo", *extra_rx) |
| if rx.wait(timeout=TIMEOUT): |
| raise Exception("Receiver failed to consume all messages") |
| total -= 5 |
| router_index += 1 |
| if router_index == len(self.routers): |
| router_index = 0 |
| tx.send_signal(SIGINT) |
| out_text, out_err = tx.communicate(timeout=TIMEOUT) |
| if tx.returncode: |
| raise Exception("Sender failed: %s %s" |
| % (out_text, out_err)) |
| |
| def test_01_rx_failover_clean(self): |
| """ |
| Runs the receiver failover test. In this test the receiver cleanly |
| shuts down the AMQP endpoint before failing over. |
| """ |
| self._rx_failover() |
| |
| def test_02_rx_failover_dirty(self): |
| """ |
| Runs the receiver failover test. In this test the receiver abruptly |
| drops the TCP connection simulating a client crash. |
| """ |
| tcp_drop = ["-E"] |
| self._rx_failover(extra_rx_args=tcp_drop) |
| |
| def test_03_unavailable_link_attach(self): |
| """ |
| Attempt to attach a link to an unavailable address, expect the router |
| to detach it |
| """ |
| ats = AsyncTestSender(self.server_address(self.RouterA), |
| "an/unavailable/address") |
| try: |
| ats.wait() |
| self.assertTrue(False) # expect exception |
| except AsyncTestSender.TestSenderException as exc: |
| self.assertIn("link error", ats.error) |
| |
| def test_04_unavailable_anonymous_link_attach(self): |
| """ |
| Attempt to attach an anonymous link and send a message to an |
| unavailable address. Expect to allow the link, but REJECT the message |
| """ |
| message = Message(body="REJECTED!!!") |
| message.address = "another/unavailable/address" |
| ats = AsyncTestSender(self.server_address(self.RouterA), |
| target=None, |
| message=message) |
| ats.wait() |
| self.assertEqual(0, ats.accepted) |
| self.assertEqual(1, ats.rejected) |
| |
| def test_05_unavailable_anonymous_link_send(self): |
| """ |
| Attach an anonymous link and send to a configured address (no |
| subscribers). Expect to allow the link, but RELEASE the message |
| """ |
| message = Message(body="Release me, let me go...") |
| message.address = "closest/foo" |
| ats = AsyncTestSender(self.server_address(self.RouterA), |
| target=None, |
| message=message) |
| ats.wait() |
| self.assertEqual(0, ats.accepted) |
| self.assertEqual(1, ats.released) |
| self.assertEqual(0, ats.rejected) |
| |
| def test_06_parallel_priority(self): |
| """ |
| Create 10 senders each with a different priority. Send large messages |
| - large enough to trigger Qx flow control (sender argument "-sx"). |
| Ensure all messages arrive as expected. |
| """ |
| priorities = 10 |
| send_batch = 25 |
| |
| total = priorities * send_batch |
| rx = self.spawn_receiver(self.RouterC, |
| count=total, |
| address="closest/test_06_address") |
| self.RouterA.wait_address("closest/test_06_address") |
| |
| senders = [self.spawn_sender(self.RouterA, |
| send_batch, |
| "closest/test_06_address", |
| "-sx", "-p%s" % p) |
| for p in range(priorities)] |
| |
| if rx.wait(timeout=TIMEOUT): |
| raise Exception("Receiver failed to consume all messages") |
| for tx in senders: |
| out_text, out_err = tx.communicate(timeout=TIMEOUT) |
| if tx.returncode: |
| raise Exception("Sender failed: %s %s" % (out_text, out_err)) |
| |
| |
| if __name__ == '__main__': |
| unittest.main(main_module()) |
