Adds Zaqar consumer-producer scenario

This scenario for Zaqar benchmarks a serial consumer-producer
use case: creates a queue, sends messages to it, retrieves
messages from it and deletes the queue.

Some changes have been added to the fakes to test this
scenario properly.

Change-Id: I49d5956bbf571840829b3e9b1411a907748cf1ed

doc/samples/tasks/scenarios/zaqar/producer_consumer.json

@@ -0,0 +1,16 @@
+{
+    "ZaqarBasic.producer_consumer": [
+        {
+            "args": {
+                "name_length": 10,
+                "min_msg_count": 50,
+                "max_msg_count": 200
+            },
+            "runner": {
+                "type": "constant",
+                "times": 100,
+                "concurrency": 10
+            }
+        }
+    ]
+}

doc/samples/tasks/scenarios/zaqar/producer_consumer.yaml

@@ -0,0 +1,11 @@
+---
+  ZaqarBasic.producer_consumer:
+    -
+      args:
+        name_length: 10
+        min_msg_count: 50
+        max_msg_count: 200
+      runner:
+        type: "constant"
+        times: 100
+        concurrency: 10

rally-jobs/rally-zaqar.yaml

@@ -10,3 +10,17 @@
       sla:
         failure_rate:
           max: 0
+
+  ZaqarBasic.producer_consumer:
+    -
+      args:
+        name_length: 10
+        min_msg_count: 50
+        max_msg_count: 200
+      runner:
+        type: "constant"
+        times: 100
+        concurrency: 10
+      sla:
+        failure_rate:
+          max: 0

rally/benchmark/scenarios/zaqar/basic.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2014 Red Hat, Inc.
+#
 # Licensed 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
@@ -10,6 +12,8 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import random
+
 from rally.benchmark.scenarios import base
 from rally.benchmark.scenarios.zaqar import utils as zutils
 from rally.benchmark import validation
@@ -20,11 +24,35 @@ class ZaqarBasic(zutils.ZaqarScenario):
     @validation.number("name_length", minval=10)
     @base.scenario(context={"cleanup": ["zaqar"]})
     def create_queue(self, name_length=10, **kwargs):
-        """Creates Zaqar queue with random name.
+        """Creates Zaqar queue with random name
 
         :param name_length: length of generated (random) part of name
-        :param **kwargs: Other optional parameters to create queues like
-                        "metadata".
+        :param kwargs: other optional parameters to create queues like
+                       "metadata"
         """
 
         self._queue_create(name_length=name_length, **kwargs)
+
+    @validation.number("name_length", minval=10)
+    @base.scenario(context={"cleanup": ["zaqar"]})
+    def producer_consumer(self, name_length=10,
+                          min_msg_count=50, max_msg_count=200, **kwargs):
+        """Serial producer/consumer
+
+        Creates a Zaqar queue with random name, sends a set of messages
+        and then retrieves an iterator containing those
+
+        :param name_length: length of generated (random) part of name
+        :param min_msg_count: min number of messages to be posted
+        :param max_msg_count: max number of messages to be posted
+        :param kwargs: other optional parameters to create queues like
+                       'metadata'
+        """
+
+        queue = self._queue_create(name_length=name_length, **kwargs)
+        msg_count = random.randint(min_msg_count, max_msg_count)
+        messages = [{'body': {'id': idx}, 'ttl': 360} for idx
+                    in range(msg_count)]
+        self._messages_post(queue, messages, min_msg_count, max_msg_count)
+        self._messages_list(queue)
+        self._queue_delete(queue)

rally/benchmark/scenarios/zaqar/utils.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2014 Red Hat, Inc.
+#
 # Licensed 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
@@ -17,13 +19,42 @@ class ZaqarScenario(base.Scenario):
 
     @base.atomic_action_timer('zaqar.create_queue')
     def _queue_create(self, name_length=10, **kwargs):
-        """Creates Zaqar queue with random name.
+        """Creates Zaqar queue with random name
 
         :param name_length: length of generated (random) part of name
-        :param **kwargs: Other optional parameters to create queues like
-                        "metadata".
-        :returns: zaqar queue instance
+        :param **kwargs: other optional parameters to create queues like
+                         "metadata"
+        :returns: Zaqar queue instance
         """
 
         name = self._generate_random_name(length=name_length)
         return self.clients("zaqar").queue(name, **kwargs)
+
+    @base.atomic_action_timer('zaqar.delete_queue')
+    def _queue_delete(self, queue):
+        """Removes a Zaqar queue
+
+        :param queue: queue to remove
+        """
+
+        queue.delete()
+
+    def _messages_post(self, queue, messages, min_msg_count, max_msg_count):
+        """Post a list of messages to a given Zaqar queue
+
+        :param queue: post the messages to queue
+        :param messages: messages to post
+        """
+        with base.AtomicAction(self, 'zaqar.post_between_%s_and_%s_messages' %
+                               (min_msg_count, max_msg_count)):
+            queue.post(messages)
+
+    @base.atomic_action_timer('zaqar.list_messages')
+    def _messages_list(self, queue):
+        """Gets messages from a given Zaqar queue
+
+        :param queue: get messages from queue
+        :return: messages iterator
+        """
+
+        return queue.messages()

tests/unit/benchmark/scenarios/zaqar/test_basic.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2014 Red Hat, Inc.
+#
 # Licensed 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
@@ -21,10 +23,30 @@ BASIC = BASE + "basic.ZaqarBasic."
 
 class ZaqarBasicTestCase(test.TestCase):
 
-    @mock.patch(BASIC + "_generate_random_name")
+    @mock.patch(BASIC + "_generate_random_name", return_value="fizbit")
     def test_create_queue(self, mock_gen_name):
         scenario = basic.ZaqarBasic()
-        mock_gen_name.return_value = "fizbit"
         scenario._queue_create = mock.MagicMock()
         scenario.create_queue(name_length=10)
         scenario._queue_create.assert_called_once_with(name_length=10)
+
+    @mock.patch(BASIC + "_generate_random_name", return_value="kitkat")
+    def test_producer_consumer(self, mock_gen_name):
+        scenario = basic.ZaqarBasic()
+        messages = [{'body': {'id': idx}, 'ttl': 360} for idx
+                    in range(20)]
+        queue = mock.MagicMock()
+
+        scenario._queue_create = mock.MagicMock(return_value=queue)
+        scenario._messages_post = mock.MagicMock()
+        scenario._messages_list = mock.MagicMock()
+        scenario._queue_delete = mock.MagicMock()
+
+        scenario.producer_consumer(name_length=10, min_msg_count=20,
+                                   max_msg_count=20)
+
+        scenario._queue_create.assert_called_once_with(name_length=10)
+        scenario._messages_post.assert_called_once_with(queue, messages,
+                                                        20, 20)
+        scenario._messages_list.assert_called_once_with(queue)
+        scenario._queue_delete.assert_called_once_with(queue)

tests/unit/benchmark/scenarios/zaqar/test_utils.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2014 Red Hat, Inc.
+#
 # Licensed 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
@@ -26,11 +28,9 @@ class ZaqarScenarioTestCase(test.TestCase):
         self.assertIsNotNone(action_duration)
         self.assertIsInstance(action_duration, float)
 
-    @mock.patch(UTILS + "ZaqarScenario._generate_random_name")
+    @mock.patch(UTILS + "ZaqarScenario._generate_random_name",
+                return_value="kitkat")
     def test_queue_create(self, mock_gen_name):
-        name = "kitkat"
-        mock_gen_name.return_value = name
-
         queue = {}
         fake_zaqar = fakes.FakeZaqarClient()
         fake_zaqar.queue = mock.MagicMock(return_value=queue)
@@ -46,3 +46,36 @@ class ZaqarScenarioTestCase(test.TestCase):
         fake_zaqar.queue.assert_called_once_with("kitkat")
         self._test_atomic_action_timer(scenario.atomic_actions(),
                                        'zaqar.create_queue')
+
+    def test_queue_delete(self):
+        queue = fakes.FakeQueue()
+        queue.delete = mock.MagicMock()
+
+        scenario = utils.ZaqarScenario()
+        scenario._queue_delete(queue)
+        queue.delete.assert_called_once_with()
+        self._test_atomic_action_timer(scenario.atomic_actions(),
+                                       'zaqar.delete_queue')
+
+    def test_messages_post(self):
+        queue = fakes.FakeQueue()
+        queue.post = mock.MagicMock()
+
+        messages = [{'body': {'id': 'one'}, 'ttl': 100},
+                    {'body': {'id': 'two'}, 'ttl': 120},
+                    {'body': {'id': 'three'}, 'ttl': 140}]
+        min_msg_count = max_msg_count = len(messages)
+
+        scenario = utils.ZaqarScenario()
+        scenario._messages_post(queue, messages, min_msg_count, max_msg_count)
+        queue.post.assert_called_once_with(messages)
+
+    def test_messages_list(self):
+        queue = fakes.FakeQueue()
+        queue.messages = mock.MagicMock()
+
+        scenario = utils.ZaqarScenario()
+        scenario._messages_list(queue)
+        queue.messages.assert_called_once_with()
+        self._test_atomic_action_timer(scenario.atomic_actions(),
+                                       'zaqar.list_messages')

tests/unit/fakes.py

@@ -248,10 +248,13 @@ class FakeQueue(FakeResource):
         self.queue_name = name
         self.messages = FakeMessagesManager(name)
 
-    def post_message(self, messages):
+    def post(self, messages):
         for msg in messages:
             self.messages.create(**msg)
 
+    def messages(self):
+        return self.messages.list()
+
 
 class FakeDbInstance(FakeResource):
     pass