ENH: add search_by_expression workflow

lib/workflows/search_by_expression.py

@@ -0,0 +1,64 @@
+from typing import List, Optional
+from enums.query.query_presets import QueryPresetsInteger
+from enums.query.sort_order import SortOrder
+import openstack_query
+
+# pylint:disable=too-many-arguments
+
+
+def search_by_expression(
+    cloud_account: str,
+    query_type: str,
+    search_mode: str,
+    property_to_search_by: str,
+    value: int,
+    properties_to_select: Optional[List[str]] = None,
+    output_type: Optional[str] = None,
+    group_by: Optional[str] = None,
+    sort_by: Optional[List[str]] = None,
+    **kwargs
+):
+    """
+    Method that builds and runs a query to find generic numerical resource based on a expression
+    (less-than, greater-than etc)
+    :param cloud_account: A string representing the cloud account to use - set in clouds.yaml
+    :param query_type: what Query object to use to run query with ServerQuery, UserQuery etc
+    :param search_mode: A string representing a preset Enum
+        LESS_THAN,
+        LESS_THAN_OR_EQUAL_TO,
+        GREATE_THAN,
+        GREATER_THAN_OR_EQUAL_TO
+    which dictates what query to perform
+    :param property_to_search_by: A string representing a datetime property Enum that the preset will be used on
+    :param value: A number to compare property against
+    :param properties_to_select: list of strings representing which properties to select
+    :param output_type: string representing how to output the query
+    :param group_by: an optional string representing a property to group results by
+    :param sort_by: an optional set of tuples representing way which properties to sort results by
+    :param kwargs: A set of optional meta params to pass to the query
+    """
+
+    query = getattr(openstack_query, query_type)()
+    if not properties_to_select:
+        query.select_all()
+    else:
+        query.select(*properties_to_select)
+
+    query.where(
+        preset=QueryPresetsInteger.from_string(search_mode),
+        prop=property_to_search_by,
+        value=value,
+    )
+
+    if sort_by:
+        query.sort_by(*[(p, SortOrder.DESC) for p in sort_by])
+    if group_by:
+        query.group_by(group_by)
+
+    query.run(cloud_account, **kwargs)
+    return {
+        "to_html": query.to_html(),
+        "to_string": query.to_string(),
+        "to_objects": query.to_objects(),
+        "to_props": query.to_props(),
+    }[output_type]

tests/lib/workflows/test_search_by_expression.py

@@ -0,0 +1,104 @@
+from unittest.mock import patch, NonCallableMock, MagicMock
+import pytest
+
+from enums.query.sort_order import SortOrder
+from workflows.search_by_expression import search_by_expression
+
+
+@patch("workflows.search_by_expression.openstack_query")
+@patch("workflows.search_by_expression.QueryPresetsInteger")
+@pytest.mark.parametrize(
+    "output_type", ["to_html", "to_string", "to_objects", "to_props"]
+)
+def test_search_by_expression_minimal(
+    mock_preset_enum, mock_openstack_query, output_type
+):
+    """
+    Runs search_by_expression only providing required values
+    """
+
+    mock_query = MagicMock()
+    mock_cloud_account = NonCallableMock()
+
+    mock_openstack_query.MockQuery.return_value = mock_query
+    params = {
+        "cloud_account": mock_cloud_account,
+        "query_type": "MockQuery",
+        "search_mode": NonCallableMock(),
+        "property_to_search_by": NonCallableMock(),
+        "output_type": output_type,
+        "value": 1,
+    }
+    res = search_by_expression(**params)
+
+    mock_query.select_all.assert_called_once()
+    mock_query.where.assert_called_once_with(
+        preset=mock_preset_enum.from_string.return_value,
+        prop=params["property_to_search_by"],
+        value=1,
+    )
+    mock_query.run.assert_called_once_with(mock_cloud_account)
+
+    assert (
+        res
+        == {
+            "to_html": mock_query.to_html.return_value,
+            "to_string": mock_query.to_string.return_value,
+            "to_objects": mock_query.to_objects.return_value,
+            "to_props": mock_query.to_props.return_value,
+        }[output_type]
+    )
+
+
+@patch("workflows.search_by_expression.openstack_query")
+@patch("workflows.search_by_expression.QueryPresetsInteger")
+@pytest.mark.parametrize(
+    "output_type", ["to_html", "to_string", "to_objects", "to_props"]
+)
+def test_search_by_expression_all(mock_preset_enum, mock_openstack_query, output_type):
+    """
+    Runs search_by_expression providing all available params
+    """
+
+    mock_query = MagicMock()
+    mock_cloud_account = NonCallableMock()
+
+    mock_openstack_query.MockQuery.return_value = mock_query
+    params = {
+        "cloud_account": mock_cloud_account,
+        "query_type": "MockQuery",
+        "search_mode": NonCallableMock(),
+        "property_to_search_by": NonCallableMock(),
+        "output_type": output_type,
+        "properties_to_select": ["prop1", "prop2"],
+        "value": 1,
+        "group_by": NonCallableMock(),
+        "sort_by": ["prop1", "prop2"],
+        "arg1": "val1",
+        "arg2": "val2",
+    }
+    res = search_by_expression(**params)
+
+    mock_query.select.assert_called_once_with(*params["properties_to_select"])
+    mock_query.where.assert_called_once_with(
+        preset=mock_preset_enum.from_string.return_value,
+        prop=params["property_to_search_by"],
+        value=1,
+    )
+    mock_query.sort_by.assert_called_once_with(
+        *[(p, SortOrder.DESC) for p in params["sort_by"]]
+    )
+    mock_query.group_by.assert_called_once_with(params["group_by"])
+    mock_query.run.assert_called_once_with(
+        params["cloud_account"], arg1="val1", arg2="val2"
+    )
+
+    assert (
+        res
+        == {
+            "to_html": mock_query.to_html.return_value,
+            "to_string": mock_query.to_string.return_value,
+            "to_objects": mock_query.to_objects.return_value,
+            "to_props": mock_query.to_props.return_value,
+        }[output_type]
+    )