refactor:DataframeUtilitiesWithoutBatchQuery

Author: RSam-NI

nisystemlink/clients/testmonitor/utilities/__init__.py

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

nisystemlink/clients/testmonitor/utilities/_client_utilities.py

@@ -1,44 +1,10 @@
-from typing import List, Optional
+from typing import List
 
 import pandas as pd
-from nisystemlink.clients.testmonitor._test_monitor_client import TestMonitorClient
-from nisystemlink.clients.testmonitor.models import (
-    Result,
-    ResultProjection,
-)
-from nisystemlink.clients.testmonitor.utilities._client_utilities import (
-    __batch_query_results,
-)
+from nisystemlink.clients.testmonitor.models import Result
 
 
-def get_results_dataframe(
-    client: TestMonitorClient,
-    query_filter: str,
-    column_projection: Optional[List[ResultProjection]] = None,
-) -> pd.DataFrame:
-    """Fetches test results and normalizes them into a Pandas DataFrame.
-
-    Args:
-        client: The TestMonitorClient to use for the request.
-        query_filter: The result query filter in Dynamic Linq format.
-        column_projection: List of columns to retrieve when querying the results.
-            Fields you do not specify are excluded. Returns all fields if no value is specified.
-
-    Returns:
-        A Pandas DataFrame containing the results.
-
-    Raises:
-        ApiException: If unable to communicate with the `/nitestmonitor` service
-            or provided an invalid argument.
-    """
-    queried_results = __batch_query_results(client, query_filter, column_projection)
-
-    results_dataframe = __normalize_results(queried_results)
-
-    return results_dataframe
-
-
-def __normalize_results(results: List[Result]) -> pd.DataFrame:
+def convert_results_to_dataframe(results: List[Result]) -> pd.DataFrame:
     """Normalizes the results into a Pandas DataFrame.
 
     Args:

nisystemlink/clients/testmonitor/utilities/_constants.py

@@ -1,15 +1,13 @@
 import uuid
 from typing import Dict, List, Optional
 
-import pandas as pd
 import pytest
 from nisystemlink.clients.core._http_configuration import HttpConfiguration
 from nisystemlink.clients.testmonitor import TestMonitorClient
 from nisystemlink.clients.testmonitor.models import (
     CreateResultRequest,
     CreateResultsPartialSuccess,
     Result,
-    ResultProjection,
     Status,
     UpdateResultRequest,
 )
@@ -19,7 +17,6 @@
     QueryResultValuesRequest,
     ResultField,
 )
-from nisystemlink.clients.testmonitor.utilities import get_results_dataframe
 
 
 @pytest.fixture(scope="class")
@@ -397,92 +394,8 @@ def test__update_properties_append__properties_appended(
         )
         assert updated_result.properties[new_key] == new_properties[new_key]
 
-    def test__get_results_dataframe_without_column_projection__returns_complete_results_dataframe(
-        self, client: TestMonitorClient, create_results, unique_identifier
-    ):
-        create_results_request = CreateResultRequest(
-            part_number=unique_identifier,
-            program_name="Test Program",
-            status=Status.PASSED(),
-        )
-        create_results([create_results_request, create_results_request])
-        query_results_filter = f'partNumber="{unique_identifier}"'
-        query_response: PagedResults = client.query_results(
-            QueryResultsRequest(filter=query_results_filter)
-        )
-        expected_results_dataframe = self.__get_expected_results_dataframe(
-            query_response.results
-        )
-
-        results_dataframe = get_results_dataframe(
-            client, query_filter=query_results_filter
-        )
-
-        assert not results_dataframe.empty
-        assert isinstance(results_dataframe, pd.DataFrame)
-        assert len(results_dataframe) == 2
-        assert len(results_dataframe.columns.tolist()) == 22
-        assert results_dataframe.equals(expected_results_dataframe)
-
-    def test__get_results_dataframe_with_column_projection__returns_dataframe_with_projected_columns(
-        self, client: TestMonitorClient, create_results, unique_identifier
-    ):
-        create_results_request = CreateResultRequest(
-            part_number=unique_identifier,
-            program_name="Test Program",
-            status=Status.PASSED(),
-        )
-        create_results([create_results_request, create_results_request])
-        query_results_filter = f'partNumber="{unique_identifier}"'
-        query_response: PagedResults = client.query_results(
-            QueryResultsRequest(
-                filter=query_results_filter,
-                projection=[
-                    ResultProjection.PART_NUMBER,
-                    ResultProjection.PROGRAM_NAME,
-                ],
-            )
-        )
-        expected_results_dataframe = self.__get_expected_results_dataframe(
-            query_response.results
-        )
-
-        results_dataframe = get_results_dataframe(
-            client,
-            query_filter=query_results_filter,
-            column_projection=[
-                ResultProjection.PART_NUMBER,
-                ResultProjection.PROGRAM_NAME,
-            ],
-        )
-
-        assert not results_dataframe.empty
-        assert isinstance(results_dataframe, pd.DataFrame)
-        assert len(results_dataframe) == 2
-        assert len(results_dataframe.columns.tolist()) == 2
-        assert results_dataframe.equals(expected_results_dataframe)
-
-    def test__get_results_dataframe_with_no_results__returns_empty_dataframe(
-        self, client: TestMonitorClient, create_results, unique_identifier
-    ):
-        invalid_part_number = unique_identifier
-
-        results_dataframe = get_results_dataframe(
-            client, query_filter=f'partNumber="{invalid_part_number}"'
-        )
-
-        assert isinstance(results_dataframe, pd.DataFrame)
-        assert results_dataframe.empty
-
     def __map_result_to_update_result_request(
         self, result: Result
     ) -> UpdateResultRequest:
         result_dict = result.dict(exclude={"status_type_summary", "updated_at"})
         return UpdateResultRequest(**result_dict)
-
-    def __get_expected_results_dataframe(self, results: List[Result]) -> List[dict]:
-        expected_results_dict = [result.dict(exclude_unset=True) for result in results]
-        expected_results_dataframe = pd.json_normalize(expected_results_dict, sep=".")
-        expected_results_dataframe.dropna(axis="columns", how="all", inplace=True)
-
-        return expected_results_dataframe

nisystemlink/clients/testmonitor/utilities/_dataframe_utilities.py

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

tests/integration/testmonitor/test_testmonitor.py

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

tests/unit/__init__.py

@@ -0,0 +1,66 @@
+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(status_type=StatusType.PASSED),
+            id=uuid.uuid1().hex,
+            part_number=uuid.uuid1().hex,
+            keywords=["keyword1", "keyword2"],
+            properties={"property1": "value1", "property2": "value2"},
+        ),
+        Result(
+            status=Status(status_type=StatusType.PASSED),
+            id=uuid.uuid1().hex,
+            part_number=uuid.uuid1().hex,
+            keywords=[],
+        ),
+        Result(
+            status=Status(status_type=StatusType.PASSED),
+            id=uuid.uuid1().hex,
+            part_number=uuid.uuid1().hex,
+            properties={
+                "property1": "value1",
+                "property2": "value2",
+                "property3": "value3",
+            },
+        ),
+    ]
+
+    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) == 3
+        assert len(results_dataframe.columns.tolist()) == 7
+        assert results_dataframe.equals(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