feat: Export Results as Dataframe (#100)

Co-authored-by: Priyadarshini Piramanayagam <priydarshini.piramanayagam@emerson.com>

Author: priyadarshini-ni

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,129 @@
+from typing import Any, Dict, List
+
+import pandas as pd
+from nisystemlink.clients.testmonitor.models import Result
+from nisystemlink.clients.testmonitor.utilities.constants import DataFrameHeaders
+
+
+def convert_results_to_dataframe(
+    results: List[Result], set_id_as_index: bool = True
+) -> pd.DataFrame:
+    """Creates a Pandas DataFrame for the results.
+
+    Args:
+        results: The list of results to be included in the dataframe.
+        set_id_as_index: If true, result id will be set as index for the dataframe.
+                         If false, index will not be set.
+                         Default value is false.
+
+    Returns:
+        A Pandas DataFrame with the each result fields having a separate column.
+        Following fields are split into sub-columns.
+            - status_type_summary: All the entries will be split into separate columns.
+            For example, status_type_summary.LOOPING, status_type_summary.PASSED, etc
+            - status: Split into status.status_type and status.status_name columns.
+            - Properties: All the properties will be split into separate columns. For example,
+            properties.property1, properties.property2, etc.
+    """
+    results_dict = [result.dict(exclude_none=True) for result in results]
+    results_dict_with_normalized_status = __normalize_results_status(results_dict)
+    normalized_dataframe = pd.json_normalize(
+        results_dict_with_normalized_status, sep="."
+    )
+    normalized_dataframe = __format_results_columns(
+        results_dataframe=normalized_dataframe
+    )
+    if set_id_as_index and "id" in normalized_dataframe.columns:
+        normalized_dataframe.set_index("id", inplace=True)
+
+    return normalized_dataframe
+
+
+def __normalize_results_status(
+    results_dict: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    """Gets dictionary of results data and modifies the status object.
+
+    Args:
+        results: List of results.
+
+    Returns:
+        A list of result fields as dictionary. If status.status_type is "CUSTOM"
+            the status field takes the value of "status_name", else value of "status_type" is used.
+    """
+    for result in results_dict:
+        status = result.get("status", {})
+        if status.get("status_type") == "CUSTOM":
+            result["status"] = status["status_name"]
+        else:
+            result["status"] = status["status_type"].value
+
+    return results_dict
+
+
+def __format_results_columns(results_dataframe: pd.DataFrame) -> pd.DataFrame:
+    """Format results column to keep properties at the end.
+
+    Args:
+        results_dataframe: Dataframe of results.
+
+    Returns:
+        Formatted dataframe of results.
+    """
+    column_headers = results_dataframe.columns.to_list()
+    standard_column_headers = [
+        header for header in column_headers if __is_standard_column_header(header)
+    ]
+    status_type_summary_header = [
+        header
+        for header in column_headers
+        if __is_status_type_summary_header(header=header)
+    ]
+    properties_headers = [
+        header for header in column_headers if __is_property_header(header=header)
+    ]
+    standard_column_headers += status_type_summary_header + properties_headers
+
+    return results_dataframe.reindex(columns=standard_column_headers, copy=False)
+
+
+def __is_standard_column_header(header: str) -> bool:
+    """Check if column header is not status type summary or property.
+
+    Args:
+        header: column header for results dataframe.
+
+    Returns:
+        True if header doesn't start with 'status_type_summary.', 'properties.'. Else returns false.
+
+    """
+    return not (
+        __is_status_type_summary_header(header=header)
+        or __is_property_header(header=header)
+    )
+
+
+def __is_status_type_summary_header(header: str) -> bool:
+    """Check if column header is not a status type summary.
+
+    Args:
+        header: column header for results dataframe.
+
+    Returns:
+        True if header contains 'status_type_summary.'. Else returns false.
+
+    """
+    return header.startswith(DataFrameHeaders.STATUS_TYPE_SUMMARY_HEADER_PREFIX)
+
+
+def __is_property_header(header: str) -> bool:
+    """Check if column header is not a property.
+
+    Args:
+        header: column header for results dataframe.
+
+    Returns:
+        True if header contains 'properties.'. Else returns false.
+
+    """
+    return header.startswith(DataFrameHeaders.PROPERTY_COLUMN_HEADER_PREFIX)

nisystemlink/clients/testmonitor/utilities/constants.py

@@ -0,0 +1,4 @@
+class DataFrameHeaders:
+    STATUS_TYPE_SUMMARY_HEADER_PREFIX = "status_type_summary."
+
+    PROPERTY_COLUMN_HEADER_PREFIX = "properties."

tests/testmonitor/__init__.py

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

tests/testmonitor/test_testmonitor_dataframe_utilities.py

@@ -0,0 +1,217 @@
+import datetime
+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=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            updated_at=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            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,
+        ),
+        Result(
+            status=Status.FAILED(),
+            started_at=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            updated_at=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            program_name="My Program Name",
+            id=uuid.uuid1().hex,
+            part_number=uuid.uuid1().hex,
+            total_time_in_seconds=16.76845106446358,
+            keywords=[],
+            properties={"property3": "value3"},
+            file_ids=[uuid.uuid1().hex],
+            status_type_summary={StatusType.PASSED: 0, StatusType.FAILED: 1},
+            workspace=uuid.uuid1().hex,
+        ),
+        Result(
+            status=Status(status_type=StatusType.CUSTOM, status_name="custom_status"),
+            started_at=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            updated_at=datetime.datetime(
+                2018, 5, 7, 18, 58, 5, 219692, tzinfo=datetime.timezone.utc
+            ),
+            program_name="My Program Name",
+            id=uuid.uuid1().hex,
+            file_ids=[uuid.uuid1().hex],
+            status_type_summary={StatusType.PASSED: 0, StatusType.FAILED: 1},
+            workspace=uuid.uuid1().hex,
+        ),
+    ]
+
+    return results
+
+
+@pytest.mark.enterprise
+class TestTestmonitorDataframeUtilities:
+    def test__convert_results_with_all_fields_to_dataframe__returns_whole_results_dataframe(
+        self, results
+    ):
+        expected_results_dataframe = self.__get_expected_results_dataframe(
+            results=results
+        )
+
+        results_dataframe = convert_results_to_dataframe(
+            results=results, set_id_as_index=False
+        )
+
+        assert not results_dataframe.empty
+        assert len(results_dataframe.columns.tolist()) == 20
+        pd.testing.assert_frame_equal(
+            results_dataframe, expected_results_dataframe, check_dtype=True
+        )
+        assert isinstance(results_dataframe["status"].iloc[0], str)
+        assert results_dataframe["started_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["updated_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["file_ids"].dtype == "object"
+        assert isinstance(results_dataframe["file_ids"].iloc[0], List)
+        assert results_dataframe["data_table_ids"].dtype == "object"
+        assert isinstance(results_dataframe["data_table_ids"].iloc[0], List)
+        assert results_dataframe["keywords"].dtype == "object"
+        assert isinstance(results_dataframe["keywords"].iloc[0], List)
+
+    def test__convert_results_with_specific_fields_to_dataframe__returns_results_dataframe_with_specific_fields(
+        self, results
+    ):
+        results = results[1:]
+        expected_results_dataframe = self.__get_expected_results_dataframe(
+            results=results
+        )
+
+        results_dataframe = convert_results_to_dataframe(
+            results=results, set_id_as_index=False
+        )
+
+        assert not results_dataframe.empty
+        assert len(results_dataframe.columns.tolist()) == 13
+        pd.testing.assert_frame_equal(
+            results_dataframe, expected_results_dataframe, check_dtype=True
+        )
+        assert isinstance(results_dataframe["status"].iloc[0], str)
+        assert results_dataframe["started_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["updated_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["file_ids"].dtype == "object"
+        assert isinstance(results_dataframe["file_ids"].iloc[0], List)
+        assert results_dataframe["keywords"].dtype == "object"
+        assert isinstance(results_dataframe["keywords"].iloc[0], List)
+
+    def test__convert_results_to_dataframe_with_id_index__returns_results_dataframe_with_id_index(
+        self, results
+    ):
+        expected_results_dataframe = self.__get_expected_results_dataframe(
+            results=results
+        )
+        expected_results_dataframe = expected_results_dataframe.set_index("id")
+
+        results_dataframe = convert_results_to_dataframe(results=results)
+
+        assert not results_dataframe.empty
+        assert len(results_dataframe.columns.tolist()) == 19
+        pd.testing.assert_frame_equal(
+            results_dataframe, expected_results_dataframe, check_dtype=True
+        )
+        assert isinstance(results_dataframe["status"].iloc[0], str)
+        assert results_dataframe["started_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["updated_at"].dtype == "datetime64[ns, UTC]"
+        assert results_dataframe["file_ids"].dtype == "object"
+        assert isinstance(results_dataframe["file_ids"].iloc[0], List)
+        assert results_dataframe["data_table_ids"].dtype == "object"
+        assert isinstance(results_dataframe["data_table_ids"].iloc[0], List)
+        assert results_dataframe["keywords"].dtype == "object"
+        assert isinstance(results_dataframe["keywords"].iloc[0], List)
+
+    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
+
+    def __get_expected_results_dataframe(self, results: List[Result]):
+        results_dict = []
+        for result in results:
+            status = {
+                "status": (
+                    result.status.status_type.value
+                    if result.status and result.status.status_type != "CUSTOM"
+                    else result.status.status_name if result.status else None
+                )
+            }
+            status_type_summary = (
+                {
+                    f"status_type_summary.{key}": value
+                    for key, value in result.status_type_summary.items()
+                }
+                if result.status_type_summary
+                else {}
+            )
+            properties = (
+                {f"properties.{key}": value for key, value in result.properties.items()}
+                if result.properties
+                else {}
+            )
+            results_dict.append(
+                {
+                    **{
+                        "started_at": result.started_at,
+                        "updated_at": result.updated_at,
+                        "program_name": result.program_name,
+                        "id": result.id,
+                        "system_id": result.system_id,
+                        "host_name": result.host_name,
+                        "part_number": result.part_number,
+                        "serial_number": result.serial_number,
+                        "total_time_in_seconds": result.total_time_in_seconds,
+                        "keywords": result.keywords,
+                        "operator": result.operator,
+                        "file_ids": result.file_ids,
+                        "data_table_ids": result.data_table_ids,
+                        "workspace": result.workspace,
+                    },
+                    **status,
+                    **status_type_summary,
+                    **properties,
+                }
+            )
+
+        results_df = pd.DataFrame(results_dict)
+        results_df = results_df[
+            ["status"] + [col for col in results_df.columns if col != "status"]
+        ]
+        results_df.dropna(axis="columns", how="all", inplace=True)
+
+        return results_df