feat: Add steps APIs to the testmonitor client (#96)

Signed-off-by: Santosh Srinivasulu <[email protected]>
Co-authored-by: Priyadarshini Piramanayagam <[email protected]>
Co-authored-by: Suganth Sadaiyappan <[email protected]>
Co-authored-by: Suganth Sadaiyappan <[email protected]>

Author: 4 people

docs/api_reference/testmonitor.rst

@@ -14,6 +14,14 @@ nisystemlink.clients.testmonitor
    .. automethod:: update_results
    .. automethod:: delete_result
    .. automethod:: delete_results
+   .. automethod:: create_steps
+   .. automethod:: query_steps
+   .. automethod:: query_step_values
+   .. automethod:: delete_steps
+   .. automethod:: delete_step
+   .. automethod:: update_steps
+   .. automethod:: get_steps
+   .. automethod:: get_step
 
 .. automodule:: nisystemlink.clients.testmonitor.models
    :members:

docs/getting_started.rst

@@ -261,7 +261,7 @@ Overview
 ~~~~~~~~
 
 The :class:`.TestMonitorClient` class is the primary entry point of the Test Monitor API
-used to interact with test results (Results).
+used to interact with test results (Results) and test steps (Steps).
 
 When constructing a :class:`.TestMonitorClient`, you can pass an
 :class:`.HttpConfiguration` (like one retrieved from the
@@ -271,12 +271,19 @@ default connection. The default connection depends on your environment.
 With a :class:`.TestMonitorClient` object, you can:
 
 * Create, update, query, and delete results
+* Create, update, query, and delete steps
 
 Examples
 ~~~~~~~~
 
 Create, query, update, and delete some results
 
-.. literalinclude:: ../examples/result/results.py
+.. literalinclude:: ../examples/testmonitor/results.py
+   :language: python
+   :linenos:
+
+Create, update, query, and delete steps
+
+.. literalinclude:: ../examples/testmonitor/steps.py
    :language: python
    :linenos:

examples/testmonitor/steps.py

@@ -0,0 +1,130 @@
+from nisystemlink.clients.testmonitor import TestMonitorClient
+from nisystemlink.clients.testmonitor.models import (
+    CreateResultRequest,
+    CreateStepRequest,
+    NamedValue,
+    QueryStepsRequest,
+    QueryStepValuesRequest,
+    StepField,
+    StepIdResultIdPair,
+    UpdateStepRequest,
+)
+from nisystemlink.clients.testmonitor.models._status import Status
+from nisystemlink.clients.testmonitor.models._step_data import Measurement, StepData
+
+
+def create_test_result():
+    """Create example result on your server."""
+    new_results = [
+        CreateResultRequest(
+            part_number="Example 123 AA",
+            program_name="Example Name",
+            host_name="Example Host",
+            status=Status.PASSED(),
+            keywords=["original keyword"],
+            properties={"original property key": "yes"},
+        )
+    ]
+    create_response = client.create_results(new_results)
+    return create_response
+
+
+# Server configuration is not required when used with Systemlink Client or run throught Jupyter on SLE
+server_configuration = None
+
+# # Example of setting up the server configuration to point to your instance of SystemLink Enterprise
+# server_configuration = HttpConfiguration(
+#     server_uri="https://yourserver.yourcompany.com",
+#     api_key="YourAPIKeyGeneratedFromSystemLink",
+# )
+
+client = TestMonitorClient(configuration=server_configuration)
+
+# create a result to attach the steps to
+create_response = create_test_result()
+
+# Create the step requests
+result_id = create_response.results[0].result_id
+step_requests = [
+    CreateStepRequest(
+        step_id="step1",
+        name="step1",
+        result_id=result_id,
+        inputs=[
+            NamedValue(name="Temperature", value="35"),
+            NamedValue(name="Voltage", value="5"),
+        ],
+    ),
+    CreateStepRequest(
+        step_id="step2",
+        name="step2",
+        result_id=result_id,
+    ),
+]
+
+# Create the steps
+create_response = client.create_steps(steps=step_requests)
+created_steps = create_response.steps
+print(create_response)
+
+# You can query steps based on any field using DynamicLinq syntax.
+# These are just some representative examples.
+# Query based on result id
+query_response = client.query_steps(
+    QueryStepsRequest(filter=f'resultId == "{result_id}"')
+)
+queried_steps = query_response.steps
+
+# query step name using query step values
+query_values_response = client.query_step_values(
+    QueryStepValuesRequest(
+        filter=f'resultId == "{result_id}"',
+        field=StepField.NAME,
+    )
+)
+
+# update the data of the step
+# extra properties of the measurements will be converted to string if not already a string
+update_response = client.update_steps(
+    steps=[
+        UpdateStepRequest(
+            step_id=step.step_id,
+            result_id=step.result_id,
+            data=StepData(
+                text="My output string",
+                parameters=[
+                    Measurement(
+                        name="Temperature",
+                        status="Passed",
+                        measurement="35",
+                        lowLimit="30",
+                        highLimit="40",
+                        units="C",
+                        comparisonType="Numeric",
+                        spec_id="spec1",
+                        spec_info={
+                            "specKey": 10
+                        },  # will be converted to string as '{"specKey": 10}'
+                    )
+                ],
+            ),
+        )
+        for step in created_steps
+    ]
+)
+
+# delete all steps at once
+delete_response = client.delete_steps(
+    steps=[
+        StepIdResultIdPair(step_id=step.step_id, result_id=step.result_id)
+        for step in queried_steps
+    ]
+)
+
+create_response = client.create_steps(steps=step_requests)
+created_steps = create_response.steps
+
+# delete steps one by one
+for step in created_steps:
+    if step.step_id and step.result_id:
+        client.delete_step(result_id=step.result_id, step_id=step.step_id)

nisystemlink/clients/testmonitor/_test_monitor_client.py

@@ -9,7 +9,7 @@
     CreateResultRequest,
     UpdateResultRequest,
 )
-from uplink import Field, Query, retry, returns
+from uplink import Field, Path, Query, retry, returns
 
 from . import models
 
@@ -31,7 +31,7 @@ def __init__(self, configuration: Optional[core.HttpConfiguration] = None):
                 is used to obtain the configuration.
 
         Raises:
-            ApiException: if unable to communicate with the Spec Service.
+            ApiException: if unable to communicate with the TestMonitor Service.
         """
         if configuration is None:
             configuration = core.HttpConfigurationManager.get_configuration()
@@ -197,3 +197,190 @@ def delete_results(
                 or provided an invalid argument.
         """
         ...
+
+    @post(
+        "steps",
+        args=[Field("steps"), Field("updateResultTotalTime")],
+    )
+    def create_steps(
+        self,
+        steps: List[models.CreateStepRequest],
+        update_result_total_time: bool = False,
+    ) -> models.CreateStepsPartialSuccess:
+        """Creates one or more steps.
+
+        Args:
+            steps: A list of steps to create.
+            update_result_total_time: Determine test result total time from the step total times.
+                Defaults to False.
+
+        Returns:
+            A list of steps that were successfully created and ones that failed to be created.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...
+
+    @post("delete-steps", args=[Field("steps")])
+    def delete_steps(
+        self, steps: List[models.StepIdResultIdPair]
+    ) -> Optional[models.DeleteStepsPartialSuccess]:
+        """Deletes one or more steps by global ID.
+
+        Args:
+            steps: A list of step IDs and result IDs. Note that these are the global IDs and not the
+            `step_id` that is local to a product and workspace.
+
+        Returns:
+            None if all deletes succeed otherwise a list of which IDs failed and which succeeded.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there
+            invalid arguments.
+        """
+        ...
+
+    @delete(
+        "results/{resultId}/steps/{stepId}",
+        args=[Path("resultId"), Path("stepId"), Query("updateResultTotalTime")],
+    )
+    def delete_step(
+        self,
+        result_id: str,
+        step_id: str,
+        update_result_total_time: Optional[bool] = False,
+    ) -> None:
+        """Deletes a single step.
+
+        Args:
+            result_id: The resultId of the step to delete.
+            step_id: The stepId of the step to delete.
+            update_result_total_time: Determine test result total time from the step total times.
+                Defaults to False.
+
+        Returns:
+            None
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...
+
+    @post("query-steps")
+    def query_steps(self, query: models.QueryStepsRequest) -> models.PagedSteps:
+        """Queries for steps that match the filters.
+
+        Args:
+            query: The query contains a product ID as well as a filter for steps under that product.
+
+        Returns:
+            A list of steps that match the filter.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...
+
+    @post(
+        "update-steps",
+        args=[
+            Field("steps"),
+            Field("updateResultTotalTime"),
+            Field("replaceKeywords"),
+            Field("replaceProperties"),
+        ],
+    )
+    def update_steps(
+        self,
+        steps: List[models.UpdateStepRequest],
+        update_result_total_time: bool = False,
+        replace_keywords: bool = False,
+        replace_properties: bool = False,
+    ) -> models.UpdateStepsPartialSuccess:
+        """Updates one or more steps.
+
+        Update requires that the version field matches the version being updated from.
+
+        Args:
+            steps: a list of steps that are to be updated. Must include the global ID and
+            each step being updated must match the version currently on the server.
+            update_result_total_time: Determine test result total time from the step total times.
+                Defaults to False.
+            replace_keywords: Replace with existing keywords instead of merging them.
+                Defaults to False.
+            replace_properties: Replace with existing properties instead of merging them.
+                Defaults to False.
+
+        Returns
+            A list of steps that were successfully updated and a list of ones that were not along
+            with error messages for updates that failed.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...
+
+    @get(
+        "steps",
+        args=[Query("continuationToken"), Query("take"), Query("returnCount")],
+    )
+    def get_steps(
+        self,
+        continuation_token: Optional[str] = None,
+        take: Optional[int] = None,
+        return_count: Optional[bool] = None,
+    ) -> models.PagedSteps:
+        """Reads a list of steps.
+
+        Args:
+            continuation_token: The token used to paginate steps.
+            take: The number of steps to get in this request.
+            return_count: Whether or not to return the total number of steps available.
+
+        Returns:
+            A list of steps.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments..
+        """
+        ...
+
+    @get("results/{resultId}/steps/{stepId}", args=[Path("resultId"), Path("stepId")])
+    def get_step(self, result_id: str, step_id: str) -> models.Step:
+        """Gets a single step.
+
+        Args:
+            result_id: The resultId of the step to get.
+            step_id: The stepId of the step to get.
+
+        Returns:
+            The step.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...
+
+    @returns.json  # type: ignore
+    @post("query-step-values")
+    def query_step_values(self, query: models.QueryStepValuesRequest) -> List[str]:
+        """Queries values for a step field.
+
+        Args:
+            query: The query parameters.
+
+        Returns:
+            A list of values for the specified step field.
+
+        Raises:
+            ApiException: if unable to communicate with the `/nitestmonitor` service or if there are
+            invalid arguments.
+        """
+        ...