yardstick/tests/functional/common/messaging/test_messaging.py

   1 # Copyright (c) 2018 Intel Corporation
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import multiprocessing
  16 import os
  17 import time
  18
  19 from yardstick.common.messaging import consumer
  20 from yardstick.common.messaging import payloads
  21 from yardstick.common.messaging import producer
  22 from yardstick.tests.functional import base
  23
  24
  25 TOPIC = 'topic_MQ'
  26 METHOD_INFO = 'info'
  27
  28
  29 class DummyPayload(payloads.Payload):
  30     REQUIRED_FIELDS = {'version', 'data'}
  31
  32
  33 class DummyEndpoint(consumer.NotificationHandler):
  34
  35     def info(self, ctxt, **kwargs):
  36         if ctxt['pid'] == self._ctx_pid:
  37             self._queue.put('ID {}, data: {}'.format(self._id, kwargs['data']))
  38
  39
  40 class DummyConsumer(consumer.MessagingConsumer):
  41
  42     def __init__(self, id, ctx_pid, queue):
  43         self._id = id
  44         endpoints = [DummyEndpoint(id, ctx_pid, queue)]
  45         super(DummyConsumer, self).__init__(TOPIC, ctx_pid, endpoints)
  46
  47
  48 class DummyProducer(producer.MessagingProducer):
  49     pass
  50
  51
  52 def _run_consumer(id, ctx_pid, queue):
  53     _consumer = DummyConsumer(id, ctx_pid, queue)
  54     _consumer.start_rpc_server()
  55     _consumer.wait()
  56
  57
  58 class MessagingTestCase(base.BaseFunctionalTestCase):
  59
  60     @staticmethod
  61     def _terminate_consumers(num_consumers, processes):
  62         for i in range(num_consumers):
  63             processes[i].terminate()
  64
  65     def test_run_five_consumers(self):
  66         output_queue = multiprocessing.Queue()
  67         num_consumers = 10
  68         ctx_id = os.getpid()
  69         producer = DummyProducer(TOPIC, pid=ctx_id)
  70
  71         processes = []
  72         for i in range(num_consumers):
  73             processes.append(multiprocessing.Process(
  74                 name='consumer_{}'.format(i),
  75                 target=_run_consumer,
  76                 args=(i, ctx_id, output_queue)))
  77             processes[i].start()
  78         self.addCleanup(self._terminate_consumers, num_consumers, processes)
  79
  80         time.sleep(2)  # Let consumers to create the listeners
  81         producer.send_message(METHOD_INFO, DummyPayload(version=1,
  82                                                         data='message 0'))
  83         producer.send_message(METHOD_INFO, DummyPayload(version=1,
  84                                                         data='message 1'))
  85         time.sleep(2)  # Let consumers attend the calls
  86
  87         output = []
  88         while not output_queue.empty():
  89             output.append(output_queue.get(True, 1))
  90
  91         self.assertEqual(num_consumers * 2, len(output))
  92         for i in range(num_consumers):
  93             self.assertIn('ID {}, data: {}'.format(1, 'message 0'), output)
  94             self.assertIn('ID {}, data: {}'.format(1, 'message 1'), output)