Merge "Fixing ordering of 'severity' in alarms"

aodh/storage/sqlalchemy/utils.py

@@ -16,6 +16,7 @@ import operator
 from sqlalchemy import and_
 from sqlalchemy import asc
 from sqlalchemy import desc
+from sqlalchemy import func
 from sqlalchemy import not_
 from sqlalchemy import or_
 
@@ -88,8 +89,13 @@ class QueryTransformer(object):
             for field in orderby:
                 attr, order = list(field.items())[0]
                 ordering_function = self.ordering_functions[order]
-                self.query = self.query.order_by(ordering_function(
-                    getattr(self.table, attr)))
+                if attr == 'severity':
+                    self.query = self.query.order_by(ordering_function(
+                        func.field(getattr(self.table, attr), 'low',
+                                   'moderate', 'critical')))
+                else:
+                    self.query = self.query.order_by(ordering_function(
+                        getattr(self.table, attr)))
         else:
             self.query = self.query.order_by(desc(self.table.timestamp))
 

aodh/tests/functional/api/v2/test_complex_query_scenarios.py

@@ -21,6 +21,7 @@ from oslo_utils import timeutils
 
 from aodh.storage import models
 from aodh.tests.functional.api import v2 as tests_api
+from aodh.tests.functional import db as tests_db
 
 
 admin_header = {"X-Roles": "admin",
@@ -194,6 +195,28 @@ class TestQueryAlarmsController(tests_api.FunctionalTest):
         for alarm in data.json:
             self.assertEqual("alarm", alarm["state"])
 
+    @tests_db.run_with('mysql', 'pgsql', 'sqlite')
+    def test_query_with_orderby_severity(self):
+        orderby = '[{"severity": "ASC"}]'
+        data = self.post_json(self.alarm_url,
+                              headers=admin_header,
+                              params={"orderby": orderby})
+        alarms = list(data.json)
+        severities = [a['severity'] for a in alarms]
+        severity_choices = ['low', 'moderate', 'critical']
+        sorted_severities = sorted(severities, key=severity_choices.index)
+        self.assertEqual(sorted_severities, severities)
+
+        orderby = '[{"severity": "DESC"}]'
+        data = self.post_json(self.alarm_url,
+                              headers=admin_header,
+                              params={"orderby": orderby})
+        alarms = list(data.json)
+        severities = [a['severity'] for a in alarms]
+        sorted_severities = sorted(severities, key=severity_choices.index,
+                                   reverse=True)
+        self.assertEqual(sorted_severities, severities)
+
     def test_limit_should_be_positive(self):
         data = self.post_json(self.alarm_url,
                               headers=admin_header,