feat: Export Results as Dataframe

nisystemlink/clients/testmonitor/utilities/__init__.py

@@ -0,0 +1,3 @@
+from ._dataframe_utilities import convert_results_to_dataframe
+
+# flake8: noqa

nisystemlink/clients/testmonitor/utilities/_dataframe_utilities.py

@@ -0,0 +1,25 @@
+from typing import List
+
+import pandas as pd
+from nisystemlink.clients.testmonitor.models import Result
+
+
+def convert_results_to_dataframe(results: List[Result]) -> pd.DataFrame:
+    """Normalizes the results into a Pandas DataFrame.
+
+    Args:
+        results: The list of results to normalize.
+
+    Returns:
+        A Pandas DataFrame with the normalized queried results.
+        status_type_summary will be normalized into the respective status types.
+        For example, status_type_summary.LOOPING, status_type_summary.PASSED.
+        Status is normalized into status.status_type and status.status_name.
+        Properties are normalized into respective properties. For example,
+        properties.property1, properties.property2 and etc.
+    """
+    results_dict = [result.dict(exclude_unset=True) for result in results]
+    normalized_dataframe = pd.json_normalize(results_dict, sep=".")
+    normalized_dataframe.dropna(axis="columns", how="all", inplace=True)
+
+    return normalized_dataframe

pyproject.toml

@@ -38,6 +38,7 @@ requests = "^2.28.1"
 uplink   = "^0.9.7"
 pydantic = "^1.10.2"
 pyyaml = "^6.0.1"
+pandas = "^2.1.0"
 
 [tool.poetry.group.dev.dependencies]
 black               = ">=22.10,<25.0"

tests/unit/__init__.py

@@ -0,0 +1 @@
+# flake8: noqa

tests/unit/testmonitor/__init__.py

@@ -0,0 +1 @@
+# flake8: noqa

tests/unit/testmonitor/test_testmonitor_dataframe_utilities.py

@@ -0,0 +1,65 @@
+import uuid
+from typing import List
+
+import pandas as pd
+import pytest
+from nisystemlink.clients.testmonitor.models._result import Result
+from nisystemlink.clients.testmonitor.models._status import Status, StatusType
+from nisystemlink.clients.testmonitor.utilities._dataframe_utilities import (
+    convert_results_to_dataframe,
+)
+
+
+@pytest.fixture(scope="class")
+def results() -> List[Result]:
+    """Sample results for testing purposes."""
+    results = [
+        Result(
+            status = Status.PASSED(),
+            started_at = "2018-05-07T18:58:05.219692Z",
+            updated_at = "2018-05-07T18:58:05.219692Z",
+            program_name = "My Program Name",
+            id = uuid.uuid1().hex,
+            system_id = uuid.uuid1().hex,
+            host_name = "host name",
+            part_number = uuid.uuid1().hex,
+            serial_number = uuid.uuid1().hex,
+            total_time_in_seconds = 16.76845106446358,
+            keywords = ["keyword1", "keyword2"],
+            properties = {"property1": "value1", "property2": "value2"},
+            operator = "sample operator",
+            file_ids = [uuid.uuid1().hex, uuid.uuid1().hex],
+            data_table_ids = [uuid.uuid1().hex, uuid.uuid1().hex],
+            status_type_summary = {StatusType.PASSED: 1, StatusType.FAILED: 0},
+            workspace = uuid.uuid1().hex,
+        ),
+    ]
+
+    return results
+
+
+@pytest.mark.enterprise
+class TestTestmonitorDataframeUtilities:
+    def test__convert_results_to_dataframe__returns_results_dataframe(self, results):
+        expected_results_dict = []
+        for result in results:
+            expected_results_dict.append(result.dict(exclude_unset=True))
+        expected_results_dataframe = pd.json_normalize(expected_results_dict, sep=".")
+        expected_results_dataframe.dropna(axis="columns", how="all", inplace=True)
+
+        results_dataframe = convert_results_to_dataframe(results=results)
+
+        assert not results_dataframe.empty
+        assert isinstance(results_dataframe, pd.DataFrame)
+        assert len(results_dataframe) == 1
+        assert len(results_dataframe.columns.tolist()) == 19
+        assert results_dataframe.equals(expected_results_dataframe), expected_results_dataframe
+        pd.testing.assert_frame_equal(results_dataframe, expected_results_dataframe, check_dtype=True), expected_results_dataframe
+
+    def test__convert_results_to_dataframe_with_no_results__returns_empty_dataframe(
+        self,
+    ):
+        results_dataframe = convert_results_to_dataframe(results=[])
+
+        assert isinstance(results_dataframe, pd.DataFrame)
+        assert results_dataframe.empty