| from __future__ import absolute_import |
| # |
| # 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 absolute_import |
| |
| import os |
| |
| from .common import Test, Skipped, free_tcp_ports, \ |
| MessengerReceiverC, MessengerSenderC, \ |
| ReactorReceiverC, ReactorSenderC, \ |
| isSSLPresent |
| |
| # |
| # Tests that run the apps |
| # |
| |
| |
| class AppTests(Test): |
| |
| def __init__(self, *args): |
| Test.__init__(self, *args) |
| self.is_valgrind = False |
| |
| def default(self, name, value, **kwargs): |
| if self.is_valgrind: |
| default = kwargs.get("valgrind", value) |
| else: |
| default = value |
| return Test.default(self, name, default, **kwargs) |
| |
| @property |
| def iterations(self): |
| return int(self.default("iterations", 2, fast=1, valgrind=2)) |
| |
| @property |
| def send_count(self): |
| return int(self.default("send_count", 17, fast=1, valgrind=2)) |
| |
| @property |
| def target_count(self): |
| return int(self.default("target_count", 5, fast=1, valgrind=2)) |
| |
| @property |
| def send_batch(self): |
| return int(self.default("send_batch", 7, fast=1, valgrind=2)) |
| |
| @property |
| def forward_count(self): |
| return int(self.default("forward_count", 5, fast=1, valgrind=2)) |
| |
| @property |
| def port_count(self): |
| return int(self.default("port_count", 3, fast=1, valgrind=2)) |
| |
| @property |
| def sender_count(self): |
| return int(self.default("sender_count", 3, fast=1, valgrind=2)) |
| |
| def valgrind_test(self): |
| self.is_valgrind = True |
| |
| def setUp(self): |
| self.senders = [] |
| self.receivers = [] |
| |
| def tearDown(self): |
| pass |
| |
| def _do_test(self, iterations=1): |
| verbose = self.verbose |
| |
| for R in self.receivers: |
| R.start(verbose) |
| |
| for j in range(iterations): |
| for S in self.senders: |
| S.start(verbose) |
| |
| for S in self.senders: |
| S.wait() |
| #print("SENDER OUTPUT:") |
| #print( S.stdout() ) |
| assert S.status() == 0, ("Command '%s' failed status=%d: '%s' '%s'" |
| % (str(S.cmdline()), |
| S.status(), |
| S.stdout(), |
| S.stderr())) |
| |
| for R in self.receivers: |
| R.wait() |
| #print("RECEIVER OUTPUT") |
| #print( R.stdout() ) |
| assert R.status() == 0, ("Command '%s' failed status=%d: '%s' '%s'" |
| % (str(R.cmdline()), |
| R.status(), |
| R.stdout(), |
| R.stderr())) |
| |
| # |
| # Traffic passing tests based on the Messenger apps |
| # |
| |
| |
| class MessengerTests(AppTests): |
| |
| _timeout = 60 |
| |
| def _ssl_check(self): |
| if not isSSLPresent(): |
| raise Skipped("No SSL libraries found.") |
| if os.name == "nt": |
| raise Skipped("Windows SChannel lacks anonymous cipher support.") |
| |
| def __init__(self, *args): |
| AppTests.__init__(self, *args) |
| |
| def _do_oneway_test(self, receiver, sender, domain="amqp"): |
| """ Send N messages to a receiver. |
| Parameters: |
| iterations - repeat the senders this many times |
| target_count = # of targets to send to. |
| send_count = # messages sent to each target |
| """ |
| iterations = self.iterations |
| send_count = self.send_count |
| target_count = self.target_count |
| |
| send_total = send_count * target_count |
| receive_total = send_total * iterations |
| |
| port = free_tcp_ports()[0] |
| |
| receiver.subscriptions = ["%s://~0.0.0.0:%s" % (domain, port)] |
| receiver.receive_count = receive_total |
| receiver.timeout = MessengerTests._timeout |
| self.receivers.append(receiver) |
| |
| sender.targets = ["%s://0.0.0.0:%s/X%d" % (domain, port, j) for j in range(target_count)] |
| sender.send_count = send_total |
| sender.timeout = MessengerTests._timeout |
| self.senders.append(sender) |
| |
| self._do_test(iterations) |
| |
| def _do_echo_test(self, receiver, sender, domain="amqp"): |
| """ Send N messages to a receiver, which responds to each. |
| Parameters: |
| iterations - repeat the senders this many times |
| target_count - # targets to send to |
| send_count = # messages sent to each target |
| send_batch - wait for replies after this many messages sent |
| """ |
| iterations = self.iterations |
| send_count = self.send_count |
| target_count = self.target_count |
| send_batch = self.send_batch |
| |
| send_total = send_count * target_count |
| receive_total = send_total * iterations |
| |
| port = free_tcp_ports()[0] |
| |
| receiver.subscriptions = ["%s://~0.0.0.0:%s" % (domain, port)] |
| receiver.receive_count = receive_total |
| receiver.send_reply = True |
| receiver.timeout = MessengerTests._timeout |
| self.receivers.append(receiver) |
| |
| sender.targets = ["%s://0.0.0.0:%s/%dY" % (domain, port, j) for j in range(target_count)] |
| sender.send_count = send_total |
| sender.get_reply = True |
| sender.send_batch = send_batch |
| sender.timeout = MessengerTests._timeout |
| self.senders.append(sender) |
| |
| self._do_test(iterations) |
| |
| # Removed messenger "relay" tests. The test start-up is faulty: |
| # msgr-recv prints it's -X ready message when it starts to open a |
| # connection but it does not wait for the remote open. The relay |
| # tests depends on mapping a container name from an incoming |
| # connection. They can fail under if the sender starts before the |
| # connection is complete (esp. valgrind with SSL connections) We |
| # could fix the tests but since messenger is deprecated it does |
| # not seem worthwhile. |
| |
| def _do_star_topology_test(self, r_factory, s_factory, domain="amqp"): |
| """ |
| A star-like topology, with a central receiver at the hub, and senders at |
| the spokes. Each sender will connect to each of the ports the receiver is |
| listening on. Each sender will then create N links per each connection. |
| Each sender will send X messages per link, waiting for a response. |
| Parameters: |
| iterations - repeat the senders this many times |
| port_count - # of ports the receiver will listen on. Each sender connects |
| to all ports. |
| sender_count - # of senders |
| target_count - # of targets per connection |
| send_count - # of messages sent to each target |
| send_batch - # of messages to send before waiting for response |
| """ |
| iterations = self.iterations |
| port_count = self.port_count |
| sender_count = self.sender_count |
| target_count = self.target_count |
| send_count = self.send_count |
| send_batch = self.send_batch |
| |
| send_total = port_count * target_count * send_count |
| receive_total = send_total * sender_count * iterations |
| |
| ports = free_tcp_ports(port_count) |
| |
| receiver = r_factory() |
| receiver.subscriptions = ["%s://~0.0.0.0:%s" % (domain, port) for port in ports] |
| receiver.receive_count = receive_total |
| receiver.send_reply = True |
| receiver.timeout = MessengerTests._timeout |
| self.receivers.append(receiver) |
| |
| for i in range(sender_count): |
| sender = s_factory() |
| sender.targets = ["%s://0.0.0.0:%s/%d" % (domain, port, j) for port in ports for j in range(target_count)] |
| sender.send_count = send_total |
| sender.send_batch = send_batch |
| sender.get_reply = True |
| sender.timeout = MessengerTests._timeout |
| self.senders.append(sender) |
| |
| self._do_test(iterations) |
| |
| def test_oneway_C(self): |
| self._do_oneway_test(MessengerReceiverC(), MessengerSenderC()) |
| |
| def test_oneway_C_SSL(self): |
| self._ssl_check() |
| self._do_oneway_test(MessengerReceiverC(), MessengerSenderC(), "amqps") |
| |
| def test_echo_C(self): |
| self._do_echo_test(MessengerReceiverC(), MessengerSenderC()) |
| |
| def test_echo_C_SSL(self): |
| self._ssl_check() |
| self._do_echo_test(MessengerReceiverC(), MessengerSenderC(), "amqps") |
| |
| def test_star_topology_C(self): |
| self._do_star_topology_test(MessengerReceiverC, MessengerSenderC) |
| |
| def test_star_topology_C_SSL(self): |
| self._ssl_check() |
| self._do_star_topology_test(MessengerReceiverC, MessengerSenderC, "amqps") |
| |
| def test_oneway_reactor(self): |
| self._do_oneway_test(ReactorReceiverC(), ReactorSenderC()) |
