feat: Add steps APIs to the testmonitor client

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,6 +271,7 @@ 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
 ~~~~~~~~

examples/testmonitor/steps.py

@@ -0,0 +1,115 @@
+from nisystemlink.clients.testmonitor import TestMonitorClient
+from nisystemlink.clients.testmonitor.models import (
+    CreateMultipleStepsRequest,
+    CreateResultRequest,
+    CreateStepRequest,
+    NamedValue,
+    QueryStepsRequest,
+    QueryStepValuesRequest,
+    Status,
+    StepField,
+    StepIdResultIdPair,
+    UpdateMultipleStepsRequest,
+    UpdateStepRequest,
+)
+
+
+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(CreateMultipleStepsRequest(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 name of a step
+update_response = client.update_steps(
+    UpdateMultipleStepsRequest(
+        steps=[
+            UpdateStepRequest(
+                step_id=step.step_id,
+                result_id=step.result_id,
+                name="updated name",
+            )
+            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(CreateMultipleStepsRequest(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,165 @@ def delete_results(
                 or provided an invalid argument.
         """
         ...
+
+    @post("steps")
+    def create_steps(
+        self, steps: models.CreateMultipleStepsRequest
+    ) -> models.CreateStepsPartialSuccess:
+        """Creates one or more steps.
+
+        Args:
+            steps: A list of steps to create.
+
+        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.
+        """
+        ...
+
+    @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")
+    def update_steps(
+        self, steps: models.UpdateMultipleStepsRequest
+    ) -> 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.
+
+        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.
+        """
+        ...
+
+    @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.
+        """
+        ...
+
+    @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.
+        """
+        ...

nisystemlink/clients/testmonitor/models/__init__.py

@@ -1,17 +1,32 @@
 from ._api_info import Operation, V2Operations, ApiInfo
 from ._result import Result
 from ._status import StatusType, Status
+from ._step import Step, NamedValue, StepData
+from ._paged_results import PagedResults
+from ._paged_steps import PagedSteps
+from ._delete_results_partial_success import DeleteResultsPartialSuccess
+from ._delete_steps_partial_success import DeleteStepsPartialSuccess, StepIdResultIdPair
 from ._create_results_partial_success import CreateResultsPartialSuccess
+from ._create_steps_partial_success import CreateStepsPartialSuccess
 from ._update_results_partial_success import UpdateResultsPartialSuccess
-from ._delete_results_partial_success import DeleteResultsPartialSuccess
-from ._paged_results import PagedResults
+from ._update_steps_partial_success import UpdateStepsPartialSuccess
 from ._create_result_request import CreateResultRequest
+from ._create_steps_request import CreateMultipleStepsRequest, CreateStepRequest
 from ._update_result_request import UpdateResultRequest
+from ._update_steps_request import UpdateMultipleStepsRequest, UpdateStepRequest
 from ._query_results_request import (
     QueryResultsRequest,
     QueryResultValuesRequest,
     ResultField,
     ResultProjection,
 )
+from ._query_steps_request import (
+    QueryStepsRequest,
+    QueryStepValuesRequest,
+    StepOrderBy,
+    StepField,
+    StepProjection,
+    StepResponseFormat,
+)
 
 # flake8: noqa

nisystemlink/clients/testmonitor/models/_create_steps_partial_success.py

@@ -0,0 +1,25 @@
+from typing import List, Optional
+
+from nisystemlink.clients.core import ApiError
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+from nisystemlink.clients.testmonitor.models._create_steps_request import (
+    CreateStepRequest,
+)
+from nisystemlink.clients.testmonitor.models._step import Step
+
+
+class CreateStepsPartialSuccess(JsonModel):
+    steps: List[Step]
+    """The list of steps that were successfully created."""
+
+    failed: Optional[List[CreateStepRequest]] = None
+    """The list of step requests that failed.
+
+    If this is `None`, then all steps were successfully created.
+    """
+
+    error: Optional[ApiError] = None
+    """Error messages for steps that were not created.
+
+    If this is `None`, then all steps were successfully created.
+    """