Merge branch 'users/ammar/feat/testplan' of https://github.com/aravindhan-ni/nisystemlink-clients-python into users/ammar/feat/test_plan_template_api

Author: Ammar Husain Mian Fazulul

docs/api_reference.rst

@@ -16,6 +16,7 @@ API Reference
    api_reference/file
    api_reference/notebook
    api_reference/feeds
+   api_reference/test_plan
    api_reference/test_plan_templates
 
 Indices and tables

docs/api_reference/test_plan.rst

@@ -0,0 +1,19 @@
+.. _api_tag_page:
+
+nisystemlink.clients.test_plan.test_plan
+======================
+
+.. autoclass:: nisystemlink.clients.test_plan.test_plan.TestPlanClient
+   :exclude-members: __init__
+
+   .. automethod:: __init__
+   .. automethod:: create_test_plans
+   .. automethod:: delete_test_plans
+   .. automethod:: query_test_plans
+   .. automethod:: schedule_test_plans
+   .. automethod:: update_test_plans
+   .. automethod:: get_test_plan
+
+.. automodule:: nisystemlink.clients.test_plan.test_plan.models
+   :members:
+   :imported-members:

docs/getting_started.rst

@@ -321,6 +321,32 @@ Create, query, retry, and cancel notebook executions.
    :language: python
    :linenos:
 
+TestPlan API
+-------
+
+Overview
+~~~~~~~~
+
+The :class:`.TestPlanClient` class is the primary entry point of the TestPlan API.
+
+When constructing a :class:`.TestPlanClient`, you can pass an
+:class:`.HttpConfiguration` (like one retrieved from the
+:class:`.HttpConfigurationManager`), or let :class:`.TestPlanClient` use the
+default connection. The default connection depends on your environment.
+
+With a :class:`.TestPlanClient` object, you can:
+
+* Create, query, get, update, schedule and delete TestPlans
+
+Examples
+~~~~~~~~
+
+Create, query, get, update, schedule and delete TestPlans
+
+.. literalinclude:: ../examples/test_plan/test_plan/test_plans.py
+   :language: python
+   :linenos:
+
 TestPlanTemplate API
 -------
 

examples/test_plan/test_plan/test_plan.py

@@ -0,0 +1,72 @@
+from nisystemlink.clients.core._http_configuration import HttpConfiguration
+from nisystemlink.clients.test_plan.test_plan import TestPlanClient
+from nisystemlink.clients.test_plan.test_plan.models import (
+    CreateTestPlanRequest,
+    CreateTestPlansRequest,
+    QueryTestPlansRequest,
+    ScheduleTestPlanRequest,
+    ScheduleTestPlansRequest,
+    UpdateTestPlanRequest,
+    UpdateTestPlansRequest,
+)
+
+# Setup the server configuration to point to your instance of SystemLink Enterprise
+server_configuration = HttpConfiguration(
+    server_uri="https://yourserver.yourcompany.com",
+    api_key="YourAPIKeyGeneratedFromSystemLink",
+)
+client = TestPlanClient(configuration=server_configuration)
+
+create_test_plans_request = CreateTestPlansRequest(
+    testPlans=[
+        CreateTestPlanRequest(
+            name="Python integration test plan", state="NEW", partNumber="px40482"
+        )
+    ]
+)
+
+# create a test plan
+created_test_plans = client.create_test_plans(create_request=create_test_plans_request)
+
+created_test_plan_id = created_test_plans.created_test_plans[0].id
+
+# Query test plan using id.
+query_test_plans_request = QueryTestPlansRequest(
+    skip=0,
+    take=1,
+    descending=False,
+    returnCount=False,
+)
+client.query_test_plans(query_request=query_test_plans_request)
+
+# Get test plan
+get_test_plan = client.get_test_plan(test_plan_id=created_test_plan_id)
+
+# Update test plan
+update_test_plans_request = UpdateTestPlansRequest(
+    testPlans=[
+        UpdateTestPlanRequest(
+            id=created_test_plan_id,
+            name="Updated Test Plan",
+        )
+    ]
+)
+updated_test_plan = client.update_test_plans(update_request=update_test_plans_request)
+
+# Schedule the test plan
+schedule_test_plans_request = ScheduleTestPlansRequest(
+    test_plans=[
+        ScheduleTestPlanRequest(
+            id=created_test_plan_id,
+            planned_start_date_time="2025-05-20T15:07:42.527Z",
+            estimated_end_date_time="2025-05-20T15:07:42.527Z",
+            system_id="fake-system",
+        )
+    ]
+)
+schedule_test_plan_response = client.schedule_test_plans(
+    schedule_request=schedule_test_plans_request
+)
+
+# Delete test plan
+client.delete_test_plans(ids=[created_test_plan_id])

nisystemlink/clients/test_plan/test_plan/__init__.py

@@ -0,0 +1,3 @@
+from ._test_plan_client import TestPlanClient
+
+# flake8: noqa

nisystemlink/clients/test_plan/test_plan/_test_plan_client.py

@@ -0,0 +1,157 @@
+from typing import Optional
+
+from nisystemlink.clients import core
+from nisystemlink.clients.core._http_configuration import HttpConfiguration
+from nisystemlink.clients.core._uplink._base_client import BaseClient
+from nisystemlink.clients.core._uplink._methods import get, post
+from uplink import retry
+
+from .models import (
+    _QueryTestPlansRequest,
+    CreateTestPlansRequest,
+    CreateTestPlansResponse,
+    DeleteTestPlansRequest,
+    QueryTestPlansRequest,
+    QueryTestPlansResponse,
+    ScheduleTestPlansRequest,
+    ScheduleTestPlansResponse,
+    TestPlan,
+    UpdateTestPlansRequest,
+    UpdateTestPlansResponse,
+)
+
+
+@retry(
+    when=retry.when.status(408, 429, 502, 503, 504),
+    stop=retry.stop.after_attempt(5),
+    on_exception=retry.CONNECTION_ERROR,
+)
+class TestPlanClient(BaseClient):
+    def __init__(self, configuration: Optional[HttpConfiguration] = None):
+        """Initialize an instance.
+
+        Args:
+            configuration: Defines the web server to connect to and information about
+                how to connect. If not provided, the
+                :class:`HttpConfigurationManager <nisystemlink.clients.core.HttpConfigurationManager>`
+                is used to obtain the configuration.
+
+        Raises:
+            ApiException: if unable to communicate with the WorkOrder Service.
+        """
+        if configuration is None:
+            configuration = core.HttpConfigurationManager.get_configuration()
+
+        super().__init__(configuration, base_path="/niworkorder/v1/")
+
+    @get("testplans/{test_plan_id}")
+    def get_test_plan(self, test_plan_id) -> TestPlan:
+        """Retrieve a test plan by its ID.
+
+        Args:
+            test_plan_id: The ID of the test plan to retrieve.
+
+        Returns:
+            The TestPlan object corresponding to the given ID.
+        """
+        ...
+
+    @post("testplans")
+    def create_test_plans(
+        self, create_request: CreateTestPlansRequest
+    ) -> CreateTestPlansResponse:
+        """Create a new test plan.
+
+        Args:
+            test_plan: The test plans to create.
+
+        Returns:
+            The created test plan object.
+        """
+        ...
+
+    @post("delete-testplans")
+    def delete_test_plans(self, ids: DeleteTestPlansRequest) -> None:
+        """Delete test plans by IDs.
+
+        Args:
+            test_plan_ids: A list of test plan IDs to delete.
+
+        Returns:
+            None
+        """
+        ...
+
+    @post("query-testplans")
+    def __query_test_plans(
+        self, query_request: _QueryTestPlansRequest
+    ) -> QueryTestPlansResponse:
+        """Query test plans.
+
+        Args:
+            query: The query to execute.
+
+        Returns:
+            A QueryTestPlansResponse object containing test plans that match the query.
+        """
+        ...
+
+    def query_test_plans(
+        self, query_request: QueryTestPlansRequest
+    ) -> QueryTestPlansResponse:
+        """Query test plans.
+
+        Args:
+            query: The query to execute.
+
+        Returns:
+            A QueryTestPlansResponse object containing test plans that match the query.
+        """
+        projection_str = (
+            [projection.name for projection in query_request.projection]
+            if query_request.projection
+            else None
+        )
+        query_params = {
+            "filter": query_request.filter,
+            "take": query_request.take,
+            "continuation_token": query_request.continuation_token,
+            "order_by": query_request.order_by,
+            "descending": query_request.descending,
+            "return_count": query_request.return_count,
+            "projection": projection_str,
+        }
+
+        query_params = {k: v for k, v in query_params.items() if v is not None}
+
+        query_test_plans_request = _QueryTestPlansRequest(**query_params)
+
+        return self.__query_test_plans(query_request=query_test_plans_request)
+
+    @post("schedule-testplans")
+    def schedule_test_plans(
+        self, schedule_request: ScheduleTestPlansRequest
+    ) -> ScheduleTestPlansResponse:
+        """Schedule a test plan.
+
+        Args:
+            schedule: The schedule to apply to the test plan.
+
+        Returns:
+            A ScheduleTestPlansResponse object containing the scheduled test plan.
+        """
+        ...
+
+    @post("update-testplans")
+    def update_test_plans(
+        self, update_request: UpdateTestPlansRequest
+    ) -> UpdateTestPlansResponse:
+        """Update a test plan.
+
+        Args:
+            test_plan: The test plan to update.
+
+        Returns:
+            The updated test plan object.
+        """
+        ...

nisystemlink/clients/test_plan/test_plan/models/__init__.py

@@ -0,0 +1,23 @@
+from ._create_test_plans_request import CreateTestPlansRequest
+from ._create_test_plan_request import CreateTestPlanRequest
+from ._create_test_plans_response import CreateTestPlansResponse
+from ._schedule_test_plan_request import ScheduleTestPlanRequest
+from ._schedule_test_plans_response import ScheduleTestPlansResponse
+from ._delete_test_plans_request import DeleteTestPlansRequest
+from ._execution_event import ExecutionEvent
+from ._query_test_plans_request import (
+    QueryTestPlansRequest,
+    TestPlanField,
+    _QueryTestPlansRequest,
+)
+from ._query_test_plans_response import QueryTestPlansResponse
+from ._schedule_test_plans_request import ScheduleTestPlansRequest
+from ._state import State
+from ._test_plan import TestPlan
+from ._update_test_plan_request import UpdateTestPlanRequest
+from ._update_test_plans_request import UpdateTestPlansRequest
+from ._update_test_plans_response import UpdateTestPlansResponse
+from ._workflow_definition import WorkflowDefinition
+from ._order_by import OrderBy
+
+# flake8: noqa

nisystemlink/clients/test_plan/test_plan/models/_create_test_plan_request.py

@@ -0,0 +1,67 @@
+from typing import Dict, List, Optional
+
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+
+from ...models._execution_definition import ExecutionDefinition
+
+
+class Dashboard(JsonModel):
+    """Represents a dashboard reference."""
+
+    id: Optional[str] = None
+    """ID of the dashboard"""
+
+    variables: Optional[Dict[str, str]] = None
+    """Variables for the dashboard"""
+
+
+class CreateTestPlanRequest(JsonModel):
+    """Represents the request body content for creating a test plan."""
+
+    name: Optional[str] = None
+    """The name of the test plan."""
+
+    template_id: Optional[str] = None
+    """The ID of the template to use for the test plan."""
+
+    state: Optional[str] = None
+    """The state of the test plan."""
+
+    description: Optional[str] = None
+    """A description of the test plan."""
+
+    assigned_to: Optional[str] = None
+    """The user or group assigned to the test plan."""
+
+    work_order_id: Optional[str] = None
+    """The work order ID associated with the test plan."""
+
+    estimated_duration_in_seconds: Optional[int] = None
+    """The estimated duration of the test plan in seconds."""
+
+    properties: Optional[Dict[str, str]] = None
+    """Additional properties for the test plan."""
+
+    part_number: Optional[str] = None
+    """The part number associated with the test plan."""
+
+    dut_id: Optional[str] = None
+    """The Device Under Test (DUT) ID."""
+
+    test_program: Optional[str] = None
+    """The test program associated with the test plan."""
+
+    systemtest_programfilter: Optional[str] = None
+    """The system filter to apply."""
+
+    workspace: Optional[str] = None
+    """The workspace associated with the test plan."""
+
+    file_ids_from_template: Optional[List[str]] = None
+    """List of file IDs from the template."""
+
+    dashboard: Optional[Dashboard] = None
+    """The dashboard associated with the test plan."""
+
+    execution_actions: Optional[List[ExecutionDefinition]] = None
+    """List of execution actions for the test plan."""