Improve unit test for entity ordering

server/tests/automated/test_entity_ordering.py

@@ -1,14 +1,14 @@
 import operator
 from time import sleep
 
-from typing import Sequence
+from typing import Sequence, Mapping
 
 from tests.automated import TestService
 
 
 class TestEntityOrdering(TestService):
     test_comment = "Entity ordering test"
-    only_fields = ["id", "started", "comment"]
+    only_fields = ["id", "started", "comment", "execution.parameters"]
 
     def setUp(self, **kwargs):
         super().setUp(**kwargs)
@@ -27,6 +27,9 @@ class TestEntityOrdering(TestService):
         # sort by the same field that we use for the search
         self._assertGetTasksWithOrdering(order_by="comment")
 
+        # sort by parameter which type is not part of db schema
+        self._assertGetTasksWithOrdering(order_by="execution.parameters.test")
+
     def test_order_with_paging(self):
         order_field = "started"
         # all results in one page
@@ -52,7 +55,7 @@ class TestEntityOrdering(TestService):
     def _get_page_tasks(self, order_by, page: int, page_size: int) -> Sequence:
         return self.api.tasks.get_all_ex(
             only_fields=self.only_fields,
-            order_by=[order_by] if order_by else None,
+            order_by=[order_by] if isinstance(order_by, str) else order_by,
             comment=self.test_comment,
             page=page,
             page_size=page_size,
@@ -63,12 +66,19 @@ class TestEntityOrdering(TestService):
         Assert that vals are sorted in the ascending or descending order
         with None values are always coming from the end
         """
-        if None in vals:
+        empty = [None, "", [], {}]
-            first_null_idx = vals.index(None)
+        empty_value = None
-            none_tail = vals[first_null_idx:]
+        idx = 0
-            vals = vals[:first_null_idx]
+        for idx, val in enumerate(vals):
-            self.assertTrue(all(val is None for val in none_tail))
+            if val in empty:
-            self.assertTrue(all(val is not None for val in vals))
+                empty_value = val
+                break
+
+        if idx < len(vals) - 1:
+            none_tail = vals[idx:]
+            vals = vals[:idx]
+            self.assertTrue(all(val == empty_value for val in none_tail))
+            self.assertTrue(all(val != empty_value for val in vals))
 
         if ascending:
             cmp = operator.le
@@ -76,10 +86,18 @@ class TestEntityOrdering(TestService):
             cmp = operator.ge
         self.assertTrue(all(cmp(i, j) for i, j in zip(vals, vals[1:])))
 
+    def _get_value_for_path(self, data: Mapping, field_path: Sequence[str]):
+        val = None
+        for name in field_path:
+            val = data.get(name)
+            data = val if isinstance(val, dict) else {}
+
+        return val
+
     def _assertGetTasksWithOrdering(self, order_by: str = None, **kwargs):
         tasks = self.api.tasks.get_all_ex(
             only_fields=self.only_fields,
-            order_by=[order_by] if order_by else None,
+            order_by=[order_by] if isinstance(order_by, str) else order_by,
             comment=self.test_comment,
             **kwargs,
         ).tasks
@@ -87,12 +105,17 @@ class TestEntityOrdering(TestService):
         if order_by:
             # test that the output is correctly ordered
             field_name = order_by if not order_by.startswith("-") else order_by[1:]
-            field_vals = [t.get(field_name) for t in tasks]
+            field_vals = [self._get_value_for_path(t, field_name.split(".")) for t in tasks]
             self._assertSorted(field_vals, ascending=not order_by.startswith("-"))
 
     def _create_tasks(self):
-        tasks = [self._temp_task() for _ in range(10)]
+        tasks = [
-        for _, task in zip(range(5), tasks):
+            self._temp_task(
+                **(dict(execution={"parameters": {"test": f"{i}"} if i >= 5 else {}}))
+            )
+            for i in range(10)
+        ]
+        for idx, task in zip(range(5), tasks):
             self.api.tasks.started(task=task)
             sleep(0.1)
         return tasks