feat: Add steps APIs to the testmonitor client

nisystemlink/clients/testmonitor/_test_monitor_client.py

@@ -1,14 +1,16 @@
 """Implementation of TestMonitor Client"""
 
-from typing import Optional
+from typing import List, Optional
 
 from nisystemlink.clients import core
 from nisystemlink.clients.core._uplink._base_client import BaseClient
-from nisystemlink.clients.core._uplink._methods import get
+from nisystemlink.clients.core._uplink._methods import delete, get, post
+from uplink import Field, Path, Query, retry
 
 from . import models
 
 
+@retry(when=retry.when.status([429, 503, 504]), stop=retry.stop.after_attempt(5))
 class TestMonitorClient(BaseClient):
     # prevent pytest from thinking this is a test class
     __test__ = False
@@ -23,7 +25,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()
@@ -40,3 +42,164 @@ def api_info(self) -> models.ApiInfo:
             ApiException: if unable to communicate with the `ni``/nitestmonitor``` service.
         """
         ...
+
+    @post("steps")
+    def create_steps(
+        self, steps: models.CreateStepsRequest
+    ) -> 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.UpdateStepsRequest
+    ) -> Optional[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.
+        """
+        ...
+
+    @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,3 +1,18 @@
 from ._api_info import Operation, V2Operations, ApiInfo
+from ._step import Step, StatusObject, NamedValueObject, StepDataObject, StatusType
+from ._paged_steps import PagedSteps
+from ._delete_steps_partial_success import DeleteStepsPartialSuccess, StepIdResultIdPair
+from ._create_steps_partial_success import CreateStepsPartialSuccess
+from ._update_steps_partial_success import UpdateStepsPartialSuccess
+from ._create_steps_request import CreateStepsRequest, CreateStepRequestObject
+from ._update_steps_request import UpdateStepsRequest, UpdateStepRequestObject
+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 (
+    CreateStepRequestObject,
+)
+from nisystemlink.clients.testmonitor.models._step import Step
+
+
+class CreateStepsPartialSuccess(JsonModel):
+    steps: Optional[List[Step]] = None
+    """The list of steps that were successfully created."""
+
+    failed: Optional[List[CreateStepRequestObject]] = None
+    """The list of steps that were not created.
+
+    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.
+    """

nisystemlink/clients/testmonitor/models/_create_steps_request.py

@@ -0,0 +1,66 @@
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+from nisystemlink.clients.testmonitor.models._step import (
+    NamedValueObject,
+    StatusObject,
+    StepDataObject,
+)
+
+
+class StepRequestObjectBase(JsonModel):
+    step_id: str
+    """Step ID."""
+
+    result_id: str
+    """Result ID."""
+
+    parent_id: Optional[str] = None
+    """Parent step ID."""
+
+    data: Optional[StepDataObject] = None
+    """Data returned by the test step."""
+
+    data_model: Optional[str] = None
+    """Data model for the step."""
+
+    started_at: Optional[datetime] = None
+    """ISO-8601 formatted timestamp indicating when the test result began."""
+
+    status: Optional[StatusObject] = None
+    """The status of the step."""
+
+    step_type: Optional[str] = None
+    """Step type."""
+
+    total_time_in_seconds: Optional[float] = None
+    """Total number of seconds the step took to execute."""
+
+    inputs: Optional[List[NamedValueObject]] = None
+    """Inputs and their values passed to the test."""
+
+    outputs: Optional[List[NamedValueObject]] = None
+    """Outputs and their values logged by the test."""
+
+    keywords: Optional[List[str]] = None
+    """Words or phrases associated with the step."""
+
+    properties: Optional[Dict[str, str]] = None
+    """Test step properties."""
+
+
+class CreateStepRequestObject(StepRequestObjectBase):
+    name: str
+    """Step name."""
+
+    children: Optional[List["CreateStepRequestObject"]] = None
+    """Nested child steps."""
+
+
+class CreateStepsRequest(JsonModel):
+    steps: List[CreateStepRequestObject]
+    """Array of test steps to create."""
+
+    update_result_total_time: Optional[bool] = None
+    """Determine test result total time from the test step total times."""

nisystemlink/clients/testmonitor/models/_delete_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
+
+
+class StepIdResultIdPair(JsonModel):
+    step_id: str
+    """The ID of the step."""
+
+    result_id: str
+    """The ID of the result associated with the step."""
+
+
+class DeleteStepsPartialSuccess(JsonModel):
+    """The result of deleting multiple steps when one or more steps could not be deleted."""
+
+    steps: Optional[List[StepIdResultIdPair]] = None
+    """The step_id and result_id pairs of the steps that were successfully deleted."""
+
+    failed: Optional[List[StepIdResultIdPair]] = None
+    """The step_id and result_id pairs of the steps that could not be deleted."""
+
+    error: Optional[ApiError] = None
+    """The error that occurred when deleting the steps."""

nisystemlink/clients/testmonitor/models/_paged_steps.py

@@ -0,0 +1,14 @@
+from typing import List, Optional
+
+from nisystemlink.clients.core._uplink._with_paging import WithPaging
+from nisystemlink.clients.testmonitor.models._step import Step
+
+
+class PagedSteps(WithPaging):
+    """The response for a Steps query containing matched steps."""
+
+    steps: List[Step]
+    """A list of all the steps in this page."""
+
+    total_count: Optional[int] = None
+    """The total number of steps that match the query."""