Add test with Cinder notifications

Change-Id: Iac3064aa4aa492c663f2545cf63e3a5be2ac0700

stacklight_tests/elasticsearch_kibana/api.py

@@ -153,3 +153,20 @@ class ElasticsearchPluginApi(base_test.PluginApi):
         logger.info("Check that the instance was deleted")
         os_conn.verify_srv_deleted(instance)
         return instance.id
+
+    def make_volume_actions(self):
+        cinder = self.helpers.os_conn.cinder
+        logger.info("Create a volume")
+        volume = cinder.volumes.create(size=1)
+        self.helpers.wait_for_resource_status(
+            cinder.volumes, volume.id, "available")
+        logger.info("Update the volume")
+        if cinder.version == 1:
+            cinder.volumes.update(volume, display_name="updated_volume")
+        else:
+            cinder.volumes.update(volume, name="updated_volume")
+        self.helpers.wait_for_resource_status(
+            cinder.volumes, volume.id, "available")
+        logger.info("Delete the volume")
+        cinder.volumes.delete(volume)
+        return volume.id

stacklight_tests/toolchain/api.py

@@ -311,3 +311,14 @@ class ToolchainApi(object):
             [hit["_source"]["event_type"] for hit in output["hits"]["hits"]]))
         self.helpers.check_notifications(notification_list,
                                          neutron_event_types)
+
+    def check_cinder_notifications(self):
+        cinder_event_types = ["volume.update.start", "volume.update.end"]
+        volume_id = self.ELASTICSEARCH_KIBANA.make_volume_actions()
+        output = self.ELASTICSEARCH_KIBANA.query_elasticsearch(
+            index_type="notification",
+            query_filter='volume_id:"{}"'.format(volume_id), size=500)
+        notification_list = list(set([hit["_source"]["event_type"]
+                                      for hit in output["hits"]["hits"]]))
+        self.helpers.check_notifications(notification_list,
+                                         cinder_event_types)

stacklight_tests/toolchain/test_functional.py

@@ -213,3 +213,24 @@ class TestFunctionalToolchain(api.ToolchainApi):
         self.check_plugins_online()
 
         self.check_neutron_notifications()
+
+    @test(depends_on_groups=["deploy_toolchain"],
+          groups=["check_cinder_notifications_toolchain", "toolchain",
+                  "functional", "query_elasticsearch"])
+    @log_snapshot_after_test
+    def check_cinder_notifications_toolchain(self):
+        """Check that Cinder notifications are present in Elasticsearch
+
+        Scenario:
+            1. Revert snapshot with 3 deployed nodes
+            2. Create a volume and update it
+            3. Check that Cinder notifications are present in current
+               Elasticsearch index
+
+        Duration 25m
+        """
+        self.env.revert_snapshot("deploy_toolchain")
+
+        self.check_plugins_online()
+
+        self.check_cinder_notifications()