diff --git a/docs/api_reference.rst b/docs/api_reference.rst index 14ec073c..a7856fdd 100644 --- a/docs/api_reference.rst +++ b/docs/api_reference.rst @@ -18,6 +18,7 @@ API Reference api_reference/feeds api_reference/assetmanagement api_reference/systems + api_reference/test_plan Indices and tables ------------------ diff --git a/docs/api_reference/test_plan.rst b/docs/api_reference/test_plan.rst new file mode 100644 index 00000000..7e3f5669 --- /dev/null +++ b/docs/api_reference/test_plan.rst @@ -0,0 +1,22 @@ +.. _api_tag_page: + +nisystemlink.clients.test_plan +====================== + +.. autoclass:: nisystemlink.clients.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 + .. automethod:: create_test_plan_templates + .. automethod:: query_test_plan_templates + .. automethod:: delete_test_plan_templates + +.. automodule:: nisystemlink.clients.test_plan.models + :members: + :imported-members: \ No newline at end of file diff --git a/docs/getting_started.rst b/docs/getting_started.rst index 0e8d6e65..a4893dee 100644 --- a/docs/getting_started.rst +++ b/docs/getting_started.rst @@ -372,3 +372,36 @@ Create, query, and remove some systems. .. literalinclude:: ../examples/systems/systems.py :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 +* Create, query and delete test plan templates + +Examples +~~~~~~~~ + +Create, query, get, update, schedule and delete TestPlans + +.. literalinclude:: ../examples/test_plan/test_plans.py + :language: python + :linenos: + +Create, query and delete test plan templates. + +.. literalinclude:: ../examples/test_plan/test_plan_templates.py + :language: python + :linenos: diff --git a/examples/test_plan/test_plan_templates.py b/examples/test_plan/test_plan_templates.py new file mode 100644 index 00000000..3af5fdaf --- /dev/null +++ b/examples/test_plan/test_plan_templates.py @@ -0,0 +1,82 @@ +from nisystemlink.clients.core._http_configuration import HttpConfiguration +from nisystemlink.clients.test_plan import TestPlanClient +from nisystemlink.clients.test_plan.models import ( + CreateTestPlanTemplateRequest, + Dashboard, + Job, + JobExecution, + ManualExecution, + QueryTestPlanTemplatesRequest, +) + + +# 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) + +# Test plan template request metadata +create_test_plan_template_request = [ + CreateTestPlanTemplateRequest( + name="Python integration test plan template", + template_group="sample template group", + product_families=["FamilyA", "FamilyB"], + part_numbers=["PN-1001", "PN-1002"], + summary="Template for running integration test plans", + description="This template defines execution steps for integration workflows.", + test_program="TP-INT-002", + estimated_duration_in_seconds=86400, + system_filter="os:linux AND arch:x64", + execution_actions=[ + ManualExecution(action="boot", type="MANUAL"), + JobExecution( + action="run", + type="JOB", + jobs=[ + Job( + functions=["run_test_suite"], + arguments=[["test_suite.py"]], + metadata={"env": "staging"}, + ) + ], + systemId="system-001", + ), + ], + file_ids=["file1", "file2"], + workspace="your_workspace_id", + properties={"env": "staging", "priority": "high"}, + dashboard=Dashboard( + id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"} + ), + ) +] + +# Create a test plan template +create_test_plan_template_response = client.create_test_plan_templates( + test_plan_templates=create_test_plan_template_request +) + +create_test_plan_template_id = None + +if ( + create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id +): + create_test_plan_template_id = str( + create_test_plan_template_response.created_test_plan_templates[0].id + ) + +# Query test plan templates using id +query_test_plan_template_request = QueryTestPlanTemplatesRequest( + filter=f'id="{create_test_plan_template_id}"', take=1 +) + +client.query_test_plan_templates( + query_test_plan_templates=query_test_plan_template_request +) + +# Delete the created test plan template. +if create_test_plan_template_id is not None: + client.delete_test_plan_templates(ids=[create_test_plan_template_id]) diff --git a/examples/test_plan/test_plans.py b/examples/test_plan/test_plans.py new file mode 100644 index 00000000..c4ca165b --- /dev/null +++ b/examples/test_plan/test_plans.py @@ -0,0 +1,108 @@ +from datetime import datetime + +from nisystemlink.clients.core._http_configuration import HttpConfiguration +from nisystemlink.clients.test_plan import TestPlanClient +from nisystemlink.clients.test_plan.models import ( + CreateTestPlanRequest, + Dashboard, + Job, + JobExecution, + ManualExecution, + 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 = [ + CreateTestPlanRequest( + name="Python integration test plan", + state="NEW", + description="Test plan for verifying integration flow", + assigned_to="test.user@example.com", + estimated_duration_in_seconds=86400, + properties={"env": "staging", "priority": "high"}, + part_number="px40482", + dut_id="Sample-Dut_Id", + test_program="TP-Integration-001", + system_filter="os:linux AND arch:x64", + workspace="your_workspace_id", + file_ids_from_template=["file1", "file2"], + dashboard=Dashboard( + id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"} + ), + execution_actions=[ + ManualExecution(action="boot", type="MANUAL"), + JobExecution( + action="run", + type="JOB", + jobs=[ + Job( + functions=["run_test_suite"], + arguments=[["test_suite.py"]], + metadata={"env": "staging"}, + ) + ], + systemId="system-001", + ), + ], + ) +] + +# create a test plan +created_test_plans_response = client.create_test_plans( + test_plans=create_test_plans_request +) + +if created_test_plans_response.created_test_plans: + created_test_plan_id = created_test_plans_response.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( + test_plans=[ + 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=datetime.strptime( + "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ), + estimated_end_date_time=datetime.strptime( + "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ), + 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]) diff --git a/nisystemlink/clients/test_plan/__init__.py b/nisystemlink/clients/test_plan/__init__.py new file mode 100644 index 00000000..013b65cb --- /dev/null +++ b/nisystemlink/clients/test_plan/__init__.py @@ -0,0 +1,3 @@ +from ._test_plan_client import TestPlanClient + +# flake8: noqa diff --git a/nisystemlink/clients/test_plan/_test_plan_client.py b/nisystemlink/clients/test_plan/_test_plan_client.py new file mode 100644 index 00000000..248e7940 --- /dev/null +++ b/nisystemlink/clients/test_plan/_test_plan_client.py @@ -0,0 +1,159 @@ +from typing import List, 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 nisystemlink.clients.test_plan import models +from uplink import Field, retry + + +@retry( + when=retry.when.status(408, 429, 502, 503, 504), + stop=retry.stop.after_attempt(5), + on_exception=retry.CONNECTION_ERROR, +) +class TestPlanClient(BaseClient): + __test__ = False + + 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 ` + 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: str) -> models.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", args=[Field("testPlans")]) + def create_test_plans( + self, test_plans: List[models.CreateTestPlanRequest] + ) -> models.CreateTestPlansPartialSuccessResponse: + """Create a new test plan. + + Args: + test_plan: The test plans to create. + + Returns: + The created test plan object. + """ + ... + + @post("delete-testplans", args=[Field("ids")]) + def delete_test_plans(self, ids: List[str]) -> 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: models.QueryTestPlansRequest + ) -> models.PagedTestPlans: + """Query test plans. + + Args: + query: The query to execute. + + Returns: + A PagedTestPlans object containing test plans that match the query. + """ + ... + + @post("schedule-testplans") + def schedule_test_plans( + self, schedule_request: models.ScheduleTestPlansRequest + ) -> models.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: models.UpdateTestPlansRequest + ) -> models.UpdateTestPlansResponse: + """Update a test plan. + + Args: + test_plan: The test plan to update. + + Returns: + The updated test plan object. + """ + ... + + @post("testplan-templates", args=[Field("testPlanTemplates")]) + def create_test_plan_templates( + self, test_plan_templates: List[models.CreateTestPlanTemplateRequest] + ) -> models.CreateTestPlanTemplatePartialSuccessResponse: + """Creates one or more test plan template and return errors for failed creations. + + Args: + test_plan_templates: A list of test plan templates to attempt to create. + + Returns: A list of created test plan templates, test plan templates that failed to create, and errors for + failures. + + Raises: ApiException: if unable to communicate with the `/niworkorder` service of provided invalid + arguments. + """ + ... + + @post("query-testplan-templates") + def query_test_plan_templates( + self, query_test_plan_templates: models.QueryTestPlanTemplatesRequest + ) -> models.PagedTestPlanTemplates: + """Queries one or more test plan templates and return errors for failed queries. + + Returns: A list of test plan templates, based on the query and errors for the wrong query. + + Raises: ApiException: if unable to communicate with the `/niworkorder` service of provided invalid + arguments. + """ + ... + + @post("delete-testplan-templates", args=[Field("ids")]) + def delete_test_plan_templates( + self, ids: List[str] + ) -> Optional[models.DeleteTestPlanTemplatesPartialSuccessResponse]: + """Deletes one or more test plan templates and return errors for failed deletion. + + Returns: + A partial success if any test plan templates failed to delete, or None if all + test plan templates were deleted successfully. + + Raises: ApiException: if unable to communicate with the `/niworkorder` service of provided invalid + arguments. + """ + ... diff --git a/nisystemlink/clients/test_plan/models/__init__.py b/nisystemlink/clients/test_plan/models/__init__.py new file mode 100644 index 00000000..1f7b7707 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/__init__.py @@ -0,0 +1,47 @@ +from ._create_test_plan_request import CreateTestPlanRequest +from ._create_test_plans_partial_success_response import ( + CreateTestPlansPartialSuccessResponse, +) +from ._dashboard import Dashboard, DashboardUrl +from ._update_test_plan_request import UpdateTestPlanRequest +from ._update_test_plans_request import UpdateTestPlansRequest +from ._update_test_plans_response import UpdateTestPlansResponse +from ._query_test_plans_request import QueryTestPlansRequest, TestPlanField +from ._paged_test_plans import PagedTestPlans +from ._execution_event import ( + ExecutionEvent, + NotebookExecutionEvent, + JobExecutionEvent, + ManualExecutionEvent, +) +from ._order_by import OrderBy +from ._test_plan import TestPlan +from ._state import State +from ._execution_definition import ( + ExecutionDefinition, + ManualExecution, + JobExecution, + NoneExecution, + NotebookExecution, + Job, +) +from ._schedule_test_plans_request import ScheduleTestPlansRequest +from ._schedule_test_plan_request import ScheduleTestPlanRequest +from ._schedule_test_plans_response import ScheduleTestPlansResponse + +from ._test_plan_templates import TestPlanTemplateBase, TestPlanTemplate +from ._create_test_plan_templates_partial_success_response import ( + CreateTestPlanTemplatePartialSuccessResponse, +) +from ._delete_test_plan_templates_partial_success_response import ( + DeleteTestPlanTemplatesPartialSuccessResponse, +) +from ._query_test_plan_templates_request import ( + QueryTestPlanTemplatesRequest, + TestPlanTemplateField, + TestPlanTemplateOrderBy, +) +from ._paged_test_plan_templates import PagedTestPlanTemplates +from ._create_test_plan_template_request import CreateTestPlanTemplateRequest + +# flake8: noqa diff --git a/nisystemlink/clients/test_plan/models/_create_test_plan_request.py b/nisystemlink/clients/test_plan/models/_create_test_plan_request.py new file mode 100644 index 00000000..df911c18 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_create_test_plan_request.py @@ -0,0 +1,58 @@ +from typing import Dict, List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._dashboard import Dashboard +from ._execution_definition import ExecutionDefinition + + +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.""" + + system_filter: 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.""" diff --git a/nisystemlink/clients/test_plan/models/_create_test_plan_template_request.py b/nisystemlink/clients/test_plan/models/_create_test_plan_template_request.py new file mode 100644 index 00000000..981ab732 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_create_test_plan_template_request.py @@ -0,0 +1,11 @@ +from ._test_plan_templates import TestPlanTemplateBase + + +class CreateTestPlanTemplateRequest(TestPlanTemplateBase): + """Contains information about a test plan template request.""" + + name: str + """Name of the test plan template.""" + + template_group: str + """The template group defined by the user.""" diff --git a/nisystemlink/clients/test_plan/models/_create_test_plan_templates_partial_success_response.py b/nisystemlink/clients/test_plan/models/_create_test_plan_templates_partial_success_response.py new file mode 100644 index 00000000..4eee6232 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_create_test_plan_templates_partial_success_response.py @@ -0,0 +1,25 @@ +from typing import List, Optional + +from nisystemlink.clients.core._api_error import ApiError +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._create_test_plan_template_request import CreateTestPlanTemplateRequest +from ._test_plan_templates import TestPlanTemplate + + +class CreateTestPlanTemplatePartialSuccessResponse(JsonModel): + + created_test_plan_templates: Optional[List[TestPlanTemplate]] = None + """The list of test plan templates that were successfully created.""" + + failed_test_plan_templates: Optional[List[CreateTestPlanTemplateRequest]] = None + """The list of test plan templates that were not created. + + If this is `None`, then all test plan templates were successfully created. + """ + + error: Optional[ApiError] = None + """Error messages for test plan templates that were not created. + + If this is `None`, then all test plan templates were successfully created. + """ diff --git a/nisystemlink/clients/test_plan/models/_create_test_plans_partial_success_response.py b/nisystemlink/clients/test_plan/models/_create_test_plans_partial_success_response.py new file mode 100644 index 00000000..1bdd19d2 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_create_test_plans_partial_success_response.py @@ -0,0 +1,22 @@ +from typing import List, Optional + +from nisystemlink.clients.core._api_error import ApiError +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._create_test_plan_request import CreateTestPlanRequest +from ._test_plan import TestPlan + + +class CreateTestPlansPartialSuccessResponse(JsonModel): + """Represents the response from creating test plans, including successfully created, + failed test plans, and any associated errors. + """ + + created_test_plans: Optional[List[TestPlan]] = None + """List of test plans that were successfully created.""" + + failed_test_plans: Optional[List[CreateTestPlanRequest]] = None + """List of test plans that failed to be created, with their request body content.""" + + error: Optional[ApiError] = None + """The error that occurred when creating the test plans.""" diff --git a/nisystemlink/clients/test_plan/models/_dashboard.py b/nisystemlink/clients/test_plan/models/_dashboard.py new file mode 100644 index 00000000..030f72ee --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_dashboard.py @@ -0,0 +1,20 @@ +from typing import Dict, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + + +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 DashboardUrl(Dashboard): + """Definition and URL of the dashboard reference associated with this test plan.""" + + url: Optional[str] = None + """URL of the dashboard reference associated with this test plan.""" diff --git a/nisystemlink/clients/test_plan/models/_delete_test_plan_templates_partial_success_response.py b/nisystemlink/clients/test_plan/models/_delete_test_plan_templates_partial_success_response.py new file mode 100644 index 00000000..27e192ab --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_delete_test_plan_templates_partial_success_response.py @@ -0,0 +1,19 @@ +from typing import List, Optional + +from nisystemlink.clients.core._api_error import ApiError +from nisystemlink.clients.core._uplink._json_model import JsonModel + + +class DeleteTestPlanTemplatesPartialSuccessResponse(JsonModel): + """The result of deleting multiple test plan templates + when one or more test plan templates could not be deleted. + """ + + deleted_test_plan_template_ids: List[str] + """The IDs of the test plan template that could not be deleted.""" + + failed_test_plan_template_ids: Optional[List[str]] = None + """The IDs of the test plan template that could not be deleted.""" + + error: Optional[ApiError] = None + """The error that occurred when deleting the test plan template.""" diff --git a/nisystemlink/clients/test_plan/models/_execution_definition.py b/nisystemlink/clients/test_plan/models/_execution_definition.py new file mode 100644 index 00000000..6257ab3d --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_execution_definition.py @@ -0,0 +1,80 @@ +from typing import Annotated, Any, Dict, List, Literal, Optional, Union + +from nisystemlink.clients.core._uplink._json_model import JsonModel +from pydantic import Field + + +class Job(JsonModel): + """Represents a job to be executed, including its functions, arguments, and metadata.""" + + functions: List[str] + """List of function names to execute.""" + + arguments: List[List[Any]] + """List of argument lists for each function.""" + + metadata: Dict[str, Any] + """Additional metadata for the job.""" + + +class NotebookExecution(JsonModel): + """Defines the execution of a notebook.""" + + action: str + """User defined action to perform in workflow (user defined).""" + + type: Literal["NOTEBOOK"] = Field(default="NOTEBOOK") + """Type of execution, default is 'NOTEBOOK'.""" + + notebookId: str + """ID of the notebook to execute.""" + + parameters: Optional[Dict[str, str]] = None + """ Dictionary of parameters that will be passed to the notebook when the execution is run.""" + + +class ManualExecution(JsonModel): + """Represents a manual execution definition.""" + + action: str + """User defined action to perform in workflow (user defined).""" + + type: Literal["MANUAL"] = Field(default="MANUAL") + """Type of execution, default is 'MANUAL'.""" + + +class JobExecution(JsonModel): + """Defines the execution of one or more jobs.""" + + action: str + """User defined action to perform in workflow (user defined).""" + + type: Literal["JOB"] = Field(default="JOB") + """Type of execution, default is 'JOB'.""" + + jobs: Optional[List[Job]] = None + """List of jobs to execute.""" + + systemId: Optional[str] = None + """Optional system ID where jobs will run.""" + + +class NoneExecution(JsonModel): + """Represents a definition where no execution is specified.""" + + action: str + """User defined action to perform in workflow (user defined).""" + + type: Literal["NONE"] = Field(default="NONE") + """Type of execution, default is 'NONE'.""" + + +ExecutionDefinition = Annotated[ + Union[ + NotebookExecution, + ManualExecution, + JobExecution, + NoneExecution, + ], + Field(discriminator="type"), +] diff --git a/nisystemlink/clients/test_plan/models/_execution_event.py b/nisystemlink/clients/test_plan/models/_execution_event.py new file mode 100644 index 00000000..6a51a835 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_execution_event.py @@ -0,0 +1,55 @@ +from datetime import datetime +from typing import Annotated, List, Literal, Optional, Union + +from nisystemlink.clients.core._uplink._json_model import JsonModel +from pydantic import Field + + +class ExecutionEventBase(JsonModel): + """Base class for execution events, containing common attributes such as action.""" + + action: Optional[str] = None + """The user-defined action that initiated the event.""" + + triggered_at: Optional[datetime] = None + """The time the event was triggered.""" + + triggered_by: Optional[str] = None + """The user who triggered the event.""" + + +class NotebookExecutionEvent(ExecutionEventBase): + """Represents an execution event that was triggered by a notebook execution.""" + + type: Literal["NOTEBOOK"] = Field(default="NOTEBOOK") + """Represents an execution event triggered by a notebook.""" + + execution_id: Optional[str] = None + """Includes the type identifier and the execution ID.""" + + +class JobExecutionEvent(ExecutionEventBase): + """A concrete execution event that represents an event triggered by a job.""" + + type: Literal["JOB"] = Field(default="JOB") + """Represents an execution event triggered by a job.""" + + job_ids: Optional[List[str]] = None + """Includes the type identifier and a list of job IDs.""" + + +class ManualExecutionEvent(ExecutionEventBase): + """A concrete execution event that represents an event triggered manually.""" + + type: Literal["MANUAL"] = Field(default="MANUAL") + """Represents an execution event triggered manually. Includes only the type identifier.""" + + +ExecutionEvent = Annotated[ + Union[ + NotebookExecutionEvent, + ManualExecutionEvent, + JobExecutionEvent, + ], + Field(discriminator="type"), +] diff --git a/nisystemlink/clients/test_plan/models/_order_by.py b/nisystemlink/clients/test_plan/models/_order_by.py new file mode 100644 index 00000000..546aeafd --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_order_by.py @@ -0,0 +1,8 @@ +from enum import Enum + + +class OrderBy(Enum): + """The state of the test plan.""" + + ID = "ID" + UPDATE_AT = "UPDATE_AT" diff --git a/nisystemlink/clients/test_plan/models/_paged_test_plan_templates.py b/nisystemlink/clients/test_plan/models/_paged_test_plan_templates.py new file mode 100644 index 00000000..37c3e0e3 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_paged_test_plan_templates.py @@ -0,0 +1,17 @@ +from typing import List, Optional + +from nisystemlink.clients.core._uplink._with_paging import WithPaging + +from ._test_plan_templates import TestPlanTemplate + + +class PagedTestPlanTemplates(WithPaging): + """The response containing the list of products, total count of products and the continuation + token if applicable. + """ + + test_plan_templates: List[TestPlanTemplate] + """A list of all the products in this page.""" + + total_count: Optional[int] = None + """The total number of products that match the query.""" diff --git a/nisystemlink/clients/test_plan/models/_paged_test_plans.py b/nisystemlink/clients/test_plan/models/_paged_test_plans.py new file mode 100644 index 00000000..dfb12ac1 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_paged_test_plans.py @@ -0,0 +1,17 @@ +from typing import List, Optional + +from nisystemlink.clients.core._uplink._with_paging import WithPaging + +from ._test_plan import TestPlan + + +class PagedTestPlans(WithPaging): + """The response containing the list of products, total count of products and the continuation + token if applicable. + """ + + test_plans: List[TestPlan] + """A list of all the products in this page.""" + + total_count: Optional[int] = None + """The total number of products that match the query.""" diff --git a/nisystemlink/clients/test_plan/models/_query_test_plan_templates_request.py b/nisystemlink/clients/test_plan/models/_query_test_plan_templates_request.py new file mode 100644 index 00000000..7ad558d8 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_query_test_plan_templates_request.py @@ -0,0 +1,101 @@ +from enum import Enum +from typing import List, Optional + +from nisystemlink.clients.core._uplink._with_paging import WithPaging + + +class TestPlanTemplateOrderBy(str, Enum): + """An enumeration by which test plan templates can be ordered/sorted.""" + + ID = "ID" + NAME = "NAME" + TEMPLATE_GROUP = "TEMPLATE_GROUP" + CREATED_AT = "CREATED_AT" + UPDATED_AT = "UPDATED_AT" + + +class TestPlanTemplateField(str, Enum): + """Model for an object describing an test plan template with all of its properties.""" + + __test__ = False + + ID = "ID" + NAME = "NAME" + TEMPLATE_GROUP = "TEMPLATE_GROUP" + PRODUCT_FAMILIES = "PRODUCT_FAMILIES" + PART_NUMBERS = "PART_NUMBERS" + SUMMARY = "SUMMARY" + DESCRIPTION = "DESCRIPTION" + TEST_PROGRAM = "TEST_PROGRAM" + ESTIMATED_DURATION_IN_SECONDS = "ESTIMATED_DURATION_IN_SECONDS" + SYSTEM_FILTER = "SYSTEM_FILTER" + EXECUTION_ACTIONS = "EXECUTION_ACTIONS" + FILE_IDS = "FILE_IDS" + WORKSPACE = "WORKSPACE" + PROPERTIES = "PROPERTIES" + DASHBOARD = "DASHBOARD" + CREATED_BY = "CREATED_BY" + UPDATED_BY = "UPDATED_BY" + CREATED_AT = "CREATED_AT" + UPDATED_AT = "UPDATED_AT" + + +class QueryTestPlanTemplatesRequest(WithPaging): + """Request information for the query test plan templates API.""" + + filter: Optional[str] = None + """The test plan template query filter in dynamic LINQ format. + + `id`: String representing the ID of a test plan template. Field supports only equals '=' and not + equal '!=' operators for filtering. + `productFamilies`: Array of strings representing the product families to which the test plan + template belongs. + `partNumbers`: Array of strings representing the part numbers of the products to which the test + plan template belongs. + `fileIds`: The array of file IDs associated with the test plan template. + `name`: String representing the name of a test plan template. + `summary`: String representing the summary of a test plan template. + `description`: String representing description of the test plan created from this template. + `templateGroup`: String representing the template group defined by the user. + `testProgram`: String representing the test program name of the test plan created from this + template. + `systemFilter`: String representing the LINQ filter used to filter the potential list of systems + capable of executing test plans created from this template. + `workspace`: String representing the workspace where the test plan template belongs. + `createdBy`: String representing the user who created the test plan template. + `updatedBy`: String representing the user who updated the test plan template. + `createdAt`: ISO-8601 formatted timestamp indicating when the test plan template was created. + `updatedAt`: ISO-8601 formatted timestamp indicating when the test plan template was most + recently updated. + `properties`: Collection of key-value pairs related to a test plan created from this template. + Example: properties.Any(key == "Location" & value == "Austin") + + See [Dynamic Linq](https://github.com/ni/systemlink-OpenAPI-documents/wiki/Dynamic-Linq-Query-Language) + documentation for more details. + + `"@0"`, `"@1"` etc. can be used in conjunction with the `substitutions` parameter to keep this + query string more simple and reusable.""" + + take: Optional[int] = None + """The maximum number of test plan templates to return.""" + + order_by: Optional[TestPlanTemplateOrderBy] = None + """Field by which test plan templates can be ordered/sorted.""" + + substitutions: Optional[List[str]] = None + """Makes substitutions in the query filter expression + using non-negative integers. These integers + use the @ symbol as a prefix. The filter + expression replaces each substitution + with the element at the corresponding + index in this list. The index is zero-based.""" + + descending: Optional[bool] = None + """Whether to return the test plan templates in the descending order. By default, test plan + templates are sorted in the ascending order.""" + + projection: Optional[List[TestPlanTemplateField]] = None + """ + Gets or sets the projection to be used when retrieving the assets. If not specified, + all properties will be returned. + """ diff --git a/nisystemlink/clients/test_plan/models/_query_test_plans_request.py b/nisystemlink/clients/test_plan/models/_query_test_plans_request.py new file mode 100644 index 00000000..91d44423 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_query_test_plans_request.py @@ -0,0 +1,70 @@ +import enum +from typing import List, Optional + +from nisystemlink.clients.core._uplink._with_paging import WithPaging + +from ._order_by import OrderBy + + +class TestPlanField(enum.Enum): + """Model for an object describing an test plan with all of its properties.""" + + __test__ = False + + ID = "ID" + TEMPLATE_ID = "TEMPLATE_ID" + NAME = "NAME" + STATE = "STATE" + SUBSTATE = "SUBSTATE" + DESCRIPTION = "DESCRIPTION" + ASSIGNED_TO = "ASSIGNED_TO" + WORK_ORDER_ID = "WORK_ORDER_ID" + WORK_ORDER_NAME = "WORK_ORDER_NAME" + WORKSPACE = "WORKSPACE" + CREATED_BY = "CREATED_BY" + UPDATED_BY = "UPDATED_BY" + CREATED_AT = "CREATED_AT" + UPDATED_AT = "UPDATED_AT" + PROPERTIES = "PROPERTIES" + PART_NUMBER = "PART_NUMBER" + DUT_ID = "DUT_ID" + TEST_PROGRAM = "TEST_PROGRAM" + SYSTEM_ID = "SYSTEM_ID" + FIXTURE_IDS = "FIXTURE_IDS" + SYSTEM_FILTER = "SYSTEM_FILTER" + PLANNED_START_DATE_TIME = "PLANNED_START_DATE_TIME" + ESTIMATED_END_DATE_TIME = "ESTIMATED_END_DATE_TIME" + ESTIMATED_DURATION_IN_SECONDS = "ESTIMATED_DURATION_IN_SECONDS" + FILE_IDS_FROM_TEMPLATE = "FILE_IDS_FROM_TEMPLATE" + EXECUTION_ACTIONS = "EXECUTION_ACTIONS" + EXECUTION_HISTORY = "EXECUTION_HISTORY" + DASHBOARD_URL = "DASHBOARD_URL" + DASHBOARD = "DASHBOARD" + WORKFLOW = "WORKFLOW" + + +class QueryTestPlansRequest(WithPaging): + """Represents the request body for querying test plans. + Allows filtering, sorting, and pagination of test plan results. + """ + + filter: Optional[str] = None + """A string expression to filter the test plans returned by the query.""" + + take: Optional[int] = None + """The maximum number of test plans to return in the response.""" + + order_by: Optional[OrderBy] = None + """The field name to use for sorting the test plans.""" + + descending: Optional[bool] = None + """Whether to sort the test plans in descending order.""" + + return_count: Optional[bool] = None + """Whether to include the total count of matching test plans in the response.""" + + projection: Optional[List[TestPlanField]] = None + """ + Gets or sets the projection to be used when retrieving the assets. If not specified, + all properties will be returned. + """ diff --git a/nisystemlink/clients/test_plan/models/_schedule_test_plan_request.py b/nisystemlink/clients/test_plan/models/_schedule_test_plan_request.py new file mode 100644 index 00000000..0c65bd46 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_schedule_test_plan_request.py @@ -0,0 +1,32 @@ +from datetime import datetime +from typing import List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + + +class ScheduleTestPlanRequest(JsonModel): + """Represents the request body content for scheduling a test plan.""" + + id: str + """Unique identifier for the test plan to be scheduled.""" + + assigned_to: Optional[str] = None + """(Optional) User or entity to whom the test plan is assigned.""" + + dut_id: Optional[str] = None + """(Optional) Identifier for the Device Under Test (DUT).""" + + system_id: Optional[str] = None + """(Optional) Identifier for the system where the test plan will run.""" + + planned_start_date_time: Optional[datetime] = None + """(Optional) Planned start date and time for the test plan execution (ISO 8601 format).""" + + estimated_end_date_time: Optional[datetime] = None + """(Optional) Estimated end date and time for the test plan execution (ISO 8601 format).""" + + estimated_duration_in_seconds: Optional[int] = None + """(Optional) Estimated duration of the test plan execution in seconds.""" + + fixture_ids: Optional[List[str]] = None + """(Optional) List of fixture identifiers associated with the test plan.""" diff --git a/nisystemlink/clients/test_plan/models/_schedule_test_plans_request.py b/nisystemlink/clients/test_plan/models/_schedule_test_plans_request.py new file mode 100644 index 00000000..68627d7e --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_schedule_test_plans_request.py @@ -0,0 +1,15 @@ +from typing import List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._schedule_test_plan_request import ScheduleTestPlanRequest + + +class ScheduleTestPlansRequest(JsonModel): + """Represents the request body for scheduling multiple test plans.""" + + test_plans: List[ScheduleTestPlanRequest] + """List of test plan scheduling content objects.""" + + replace: Optional[bool] = None + """(Optional) If true, replaces existing scheduled test plans.""" diff --git a/nisystemlink/clients/test_plan/models/_schedule_test_plans_response.py b/nisystemlink/clients/test_plan/models/_schedule_test_plans_response.py new file mode 100644 index 00000000..b28e6edb --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_schedule_test_plans_response.py @@ -0,0 +1,20 @@ +from typing import List, Optional + +from nisystemlink.clients.core._api_error import ApiError +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._schedule_test_plan_request import ScheduleTestPlanRequest +from ._test_plan import TestPlan + + +class ScheduleTestPlansResponse(JsonModel): + """Represents the response returned after attempting to schedule one or more test plans.""" + + scheduled_test_plans: Optional[List[TestPlan]] = None + """(Optional) List of successfully scheduled test plans.""" + + failed_test_plans: Optional[List[ScheduleTestPlanRequest]] = None + """(Optional) List of test plan requests that failed to schedule.""" + + error: Optional[ApiError] = None + """The error that occurred when scheduling the test plans.""" diff --git a/nisystemlink/clients/test_plan/models/_state.py b/nisystemlink/clients/test_plan/models/_state.py new file mode 100644 index 00000000..cd727442 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_state.py @@ -0,0 +1,14 @@ +from enum import Enum + + +class State(Enum): + """The state of the test plan.""" + + NEW = "NEW" + DEFINED = "DEFINED" + REVIEWED = "REVIEWED" + SCHEDULED = "SCHEDULED" + IN_PROGRESS = "IN_PROGRESS" + PENDING_APPROVAL = "PENDING_APPROVAL" + CLOSED = "CLOSED" + CANCELED = "CANCELED" diff --git a/nisystemlink/clients/test_plan/models/_test_plan.py b/nisystemlink/clients/test_plan/models/_test_plan.py new file mode 100644 index 00000000..23c9af3b --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_test_plan.py @@ -0,0 +1,102 @@ +from datetime import datetime +from typing import Dict, List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._dashboard import DashboardUrl +from ._execution_definition import ExecutionDefinition +from ._execution_event import ExecutionEvent +from ._state import State + + +class TestPlan(JsonModel): + """Contains information about a test plan.""" + + __test__ = False + + id: str + """The unique identifier of the test plan.""" + + template_id: Optional[str] = None + """The identifier of the template used to create the test plan.""" + + name: Optional[str] = None + """The name of the test plan.""" + + state: Optional[State] = None + """The current state of the test plan.""" + + substate: Optional[str] = None + """The substate of the test plan, if any.""" + + description: Optional[str] = None + """The 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 identifier of the associated work order.""" + + work_order_name: Optional[str] = None + """The name of the associated work order.""" + + workspace: Optional[str] = None + """The workspace to which the test plan belongs.""" + + created_by: Optional[str] = None + """The user who created the test plan.""" + + updated_by: Optional[str] = None + """The user who last updated the test plan.""" + + created_At: Optional[datetime] = None + """The date and time when the test plan was created.""" + + updated_at: Optional[datetime] = None + """The date and time when the test plan was last updated.""" + + properties: Optional[Dict[str, str]] = None + """Additional properties associated with the test plan.""" + + part_number: Optional[str] = None + """The part number associated with the test plan.""" + + dut_id: Optional[str] = None + """The identifier of the device under test (DUT).""" + + test_program: Optional[str] = None + """The test program associated with the test plan.""" + + system_id: Optional[str] = None + """The identifier of the system used for the test plan.""" + + fixture_ids: Optional[List[str]] = None + """The list of fixture identifiers associated with the test plan.""" + + system_filter: Optional[str] = None + """The filter used to select systems for the test plan.""" + + planned_start_date_time: Optional[datetime] = None + """The planned start date and time for the test plan.""" + + estimated_end_date_time: Optional[datetime] = None + """The estimated end date and time for the test plan.""" + + estimated_duration_in_seconds: Optional[float] = None + """The estimated duration of the test plan in seconds.""" + + file_ids_from_template: Optional[List[str]] = None + """The list of file identifiers inherited from the template.""" + + execution_actions: Optional[List[ExecutionDefinition]] = None + """The execution actions defined for the test plan.""" + + execution_history: Optional[List[ExecutionEvent]] = None + """The execution history of the test plan.""" + + dashboard_url: Optional[Dict[str, str]] = None + """The URLs for dashboards related to the test plan.""" + + dashboard: Optional[DashboardUrl] = None + """The dashboard data related to the test plan.""" diff --git a/nisystemlink/clients/test_plan/models/_test_plan_templates.py b/nisystemlink/clients/test_plan/models/_test_plan_templates.py new file mode 100644 index 00000000..cbd37396 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_test_plan_templates.py @@ -0,0 +1,85 @@ +from datetime import datetime +from typing import Dict, List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._dashboard import Dashboard +from ._execution_definition import ExecutionDefinition + + +class TestPlanTemplateBase(JsonModel): + """Contains information about a test plan template.""" + + name: Optional[str] = None + """Name of the test plan template.""" + + template_group: Optional[str] = None + """The template group defined by the user.""" + + product_families: Optional[List[str]] = None + """Array of product families to which the test plan template belongs.""" + + part_numbers: Optional[List[str]] = None + """Array of part numbers of the products to which the test plan template belongs.""" + + summary: Optional[str] = None + """Summary of the test plan template.""" + + description: Optional[str] = None + """Description of the test plan created from this template.""" + + test_program: Optional[str] = None + """Test program name of the test plan created from this template.""" + + estimated_duration_in_seconds: Optional[int] = None + """The estimated time in seconds for executing the test plan created from this template.""" + + system_filter: Optional[str] = None + """The LINQ filter string is used to filter the potential list of + systems capable of executing test plans created from this template. + """ + + execution_actions: Optional[List[ExecutionDefinition]] = None + """Defines the executions that will be used for test plan actions + created from this template. + """ + + file_ids: Optional[List[str]] = None + """Array of file IDs associated with the test plan template.""" + + workspace: Optional[str] = None + """ID of the workspace where the test plan template belongs. + Default workspace will be taken if the value is not given. + """ + + properties: Optional[Dict[str, str]] = None + """Properties of the test plan created from this template as key-value pairs.""" + + dashboard: Optional[Dashboard] = None + """Defines a dashboard reference for a test plan.""" + + +class TestPlanTemplate(TestPlanTemplateBase): + """Contains response information for test plan template.""" + + __test__ = False + + id: Optional[str] = None + """The globally unique id of the test plan template.""" + + name: Optional[str] = None + """Name of the test plan template.""" + + created_by: Optional[str] = None + """ID of the user who created the test plan template.""" + + updated_by: Optional[str] = None + """ID of the user who most recently updated the test plan template.""" + + created_at: Optional[datetime] = None + """ISO-8601 formatted timestamp indicating when the + test plan template was created.""" + + updated_at: Optional[datetime] = None + """ISO-8601 formatted timestamp indicating when the + test plan template was most recently updated.""" diff --git a/nisystemlink/clients/test_plan/models/_update_test_plan_request.py b/nisystemlink/clients/test_plan/models/_update_test_plan_request.py new file mode 100644 index 00000000..7fe1ad25 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_update_test_plan_request.py @@ -0,0 +1,43 @@ +from typing import Dict, List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + + +class UpdateTestPlanRequest(JsonModel): + """Represents the content for updating a single test plan.""" + + id: str + """The unique identifier of the test plan to update.""" + + name: Optional[str] = None + """The new name for the test plan.""" + + state: Optional[str] = None + """The new state of the test plan.""" + + description: Optional[str] = None + """The new description for the test plan.""" + + dut_id: Optional[str] = None + """The device under test (DUT) identifier.""" + + part_number: Optional[str] = None + """The part number associated with the test plan.""" + + assigned_to: Optional[str] = None + """The user or group assigned to the test plan.""" + + test_program: Optional[str] = None + """The test program associated with the test plan.""" + + properties: Optional[Dict[str, str]] = None + """Additional properties for the test plan as key-value pairs.""" + + workspace: Optional[str] = None + """The workspace to which the test plan belongs.""" + + work_order_id: Optional[str] = None + """The work order identifier associated with the test plan.""" + + file_ids_from_template: Optional[List[str]] = None + """List of file IDs from the template to associate with the test plan.""" diff --git a/nisystemlink/clients/test_plan/models/_update_test_plans_request.py b/nisystemlink/clients/test_plan/models/_update_test_plans_request.py new file mode 100644 index 00000000..ff4c820f --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_update_test_plans_request.py @@ -0,0 +1,15 @@ +from typing import List, Optional + +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._update_test_plan_request import UpdateTestPlanRequest + + +class UpdateTestPlansRequest(JsonModel): + """Represents the request body for updating multiple test plans.""" + + test_plans: List[UpdateTestPlanRequest] + """A list of test plans to update.""" + + replace: Optional[bool] = None + """Whether to replace the existing test plans or update them.""" diff --git a/nisystemlink/clients/test_plan/models/_update_test_plans_response.py b/nisystemlink/clients/test_plan/models/_update_test_plans_response.py new file mode 100644 index 00000000..162fe9a6 --- /dev/null +++ b/nisystemlink/clients/test_plan/models/_update_test_plans_response.py @@ -0,0 +1,20 @@ +from typing import List, Optional + +from nisystemlink.clients.core._api_error import ApiError +from nisystemlink.clients.core._uplink._json_model import JsonModel + +from ._test_plan import TestPlan +from ._update_test_plan_request import UpdateTestPlanRequest + + +class UpdateTestPlansResponse(JsonModel): + """Represents the response after attempting to update test plans.""" + + updated_test_plans: Optional[List[TestPlan]] = None + """List of successfully updated test plans.""" + + failed_test_plans: Optional[List[UpdateTestPlanRequest]] = None + """List of test plans that failed to update.""" + + error: Optional[ApiError] = None + """The error that occurred when updating the test plans.""" diff --git a/pyproject.toml b/pyproject.toml index 7f94cc5c..c8c4d317 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,6 +64,7 @@ types = "mypy --config-file mypy.ini nisystemlink examples tests" [tool.pytest.ini_options] addopts = "--strict-markers" +testpaths = ["tests"] markers = [ "unit: mark a test as a unit test", "integration: mark a test as an integration test", diff --git a/tests/integration/assetmanagement/test_asset_management.py b/tests/integration/assetmanagement/test_asset_management.py index 6a5acebe..592fe3fb 100644 --- a/tests/integration/assetmanagement/test_asset_management.py +++ b/tests/integration/assetmanagement/test_asset_management.py @@ -54,6 +54,9 @@ def client(enterprise_config: HttpConfiguration) -> AssetManagementClient: @pytest.mark.integration @pytest.mark.enterprise class TestAssetManagement: + _workspace = "2300760d-38c4-48a1-9acb-800260812337" + """Used the main-test default workspace since the client + for creating a workspace has not been added yet""" _create_assets_request = [ CreateAssetRequest( @@ -77,7 +80,7 @@ class TestAssetManagement: date="2022-06-07T18:58:05.000Z", ), is_NI_asset=True, - workspace="2300760d-38c4-48a1-9acb-800260812337", + workspace=_workspace, location=AssetLocationForCreate( state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT) ), diff --git a/tests/integration/systems/test_systems.py b/tests/integration/systems/test_systems.py index 937439ac..426d0ef6 100644 --- a/tests/integration/systems/test_systems.py +++ b/tests/integration/systems/test_systems.py @@ -27,7 +27,8 @@ def client(enterprise_config: HttpConfiguration) -> SystemsClient: workspace_id = "2300760d-38c4-48a1-9acb-800260812337" -"""Constant represent id of the workspace.""" +"""Used the main-test default workspace since the client +for creating a workspace has not been added yet""" @pytest.fixture diff --git a/tests/integration/test_plan/__init__.py b/tests/integration/test_plan/__init__.py new file mode 100644 index 00000000..9c0fa90a --- /dev/null +++ b/tests/integration/test_plan/__init__.py @@ -0,0 +1 @@ +# flake8: noqa diff --git a/tests/integration/test_plan/test_test_plan_client.py b/tests/integration/test_plan/test_test_plan_client.py new file mode 100644 index 00000000..3b90466b --- /dev/null +++ b/tests/integration/test_plan/test_test_plan_client.py @@ -0,0 +1,462 @@ +from datetime import datetime +from typing import List + +import pytest +from nisystemlink.clients.core._http_configuration import HttpConfiguration +from nisystemlink.clients.test_plan import TestPlanClient +from nisystemlink.clients.test_plan.models import ( + CreateTestPlanRequest, + CreateTestPlansPartialSuccessResponse, + CreateTestPlanTemplatePartialSuccessResponse, + CreateTestPlanTemplateRequest, + Dashboard, + Job, + JobExecution, + ManualExecution, + QueryTestPlansRequest, + QueryTestPlanTemplatesRequest, + ScheduleTestPlanRequest, + ScheduleTestPlansRequest, + State, + TestPlan, + TestPlanField, + TestPlanTemplate, + TestPlanTemplateField, + UpdateTestPlanRequest, + UpdateTestPlansRequest, +) + + +@pytest.fixture(scope="class") +def client(enterprise_config: HttpConfiguration) -> TestPlanClient: + """Fixture to create a TestPlansClient instance""" + return TestPlanClient(enterprise_config) + + +@pytest.fixture +def create_test_plans(client: TestPlanClient): + """Fixture to return a factory that create test plans.""" + responses: List[CreateTestPlansPartialSuccessResponse] = [] + + def _create_test_plans( + new_test_plans: List[CreateTestPlanRequest], + ) -> CreateTestPlansPartialSuccessResponse: + response = client.create_test_plans(test_plans=new_test_plans) + responses.append(response) + return response + + yield _create_test_plans + + created_test_plans: List[TestPlan] = [] + for response in responses: + if response.created_test_plans: + created_test_plans += response.created_test_plans + + client.delete_test_plans( + ids=[ + test_plans.id + for test_plans in created_test_plans + if test_plans.id is not None + ] + ) + + +@pytest.fixture +def create_test_plan_templates(client: TestPlanClient): + """Fixture to return a factory that create test plan templates.""" + responses: List[CreateTestPlanTemplatePartialSuccessResponse] = [] + + def _create_test_plan_templates( + new_test_plan_templates: List[CreateTestPlanTemplateRequest], + ) -> CreateTestPlanTemplatePartialSuccessResponse: + response = client.create_test_plan_templates( + test_plan_templates=new_test_plan_templates + ) + responses.append(response) + return response + + yield _create_test_plan_templates + + created_test_plan_templates: List[TestPlanTemplate] = [] + for response in responses: + if response.created_test_plan_templates: + created_test_plan_templates = ( + created_test_plan_templates + response.created_test_plan_templates + ) + client.delete_test_plan_templates( + ids=[ + test_plan_template.id + for test_plan_template in created_test_plan_templates + if test_plan_template.id is not None + ] + ) + + +@pytest.mark.integration +@pytest.mark.enterprise +class TestTestPlanClient: + + _workspace = "2300760d-38c4-48a1-9acb-800260812337" + """Used the main-test default workspace since the client + for creating a workspace has not been added yet""" + + _dashboard = Dashboard( + id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"} + ) + + _execution_actions = [ + ManualExecution(action="boot", type="MANUAL"), + JobExecution( + action="run", + type="JOB", + jobs=[ + Job( + functions=["run_test_suite"], + arguments=[["test_suite.py"]], + metadata={"env": "staging"}, + ) + ], + systemId="system-001", + ), + ] + + _test_plan_create = [ + CreateTestPlanRequest( + name="Python integration test plan", + state="NEW", + description="Test plan for verifying integration flow", + assigned_to="test.user@example.com", + estimated_duration_in_seconds=86400, + properties={"env": "staging", "priority": "high"}, + part_number="px40482", + dut_id="Sample-Dut_Id", + test_program="TP-Integration-001", + system_filter="os:linux AND arch:x64", + workspace=_workspace, + file_ids_from_template=["file1", "file2"], + dashboard=_dashboard, + execution_actions=_execution_actions, + ) + ] + """create test plan request object.""" + + _create_test_plan_template_request = [ + CreateTestPlanTemplateRequest( + name="Python integration test plan template", + template_group="sample template group", + product_families=["FamilyA", "FamilyB"], + part_numbers=["PN-1001", "PN-1002"], + summary="Template for running integration test plans", + description="This template defines execution steps for integration workflows.", + test_program="TP-INT-002", + estimated_duration_in_seconds=86400, + system_filter="os:linux AND arch:x64", + execution_actions=_execution_actions, + file_ids=["file1", "file2"], + workspace=_workspace, + properties={"env": "staging", "priority": "high"}, + dashboard=_dashboard, + ) + ] + """create test plan template request object.""" + + def test__create_and_delete_test_plan__returns_created_and_deleted_test_plans( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_template_response = client.create_test_plan_templates( + test_plan_templates=self._create_test_plan_template_request + ) + + template_id = ( + create_test_plan_template_response.created_test_plan_templates[0].id + if create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id + else None + ) + + assert template_id is not None + + test_plan_request = self._test_plan_create + test_plan_request[0].template_id = template_id + + create_test_plan_response = create_test_plans(test_plan_request) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + + get_test_plan_response: TestPlan = client.get_test_plan(created_test_plan.id) + + client.delete_test_plan_templates(ids=[template_id]) + + assert created_test_plan is not None + assert created_test_plan.name == "Python integration test plan" + assert created_test_plan.state == State.NEW + assert created_test_plan.part_number == "px40482" + assert get_test_plan_response is not None + assert get_test_plan_response.name == "Python integration test plan" + + def test__get_test_plan__returns_get_test_plan( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_response = create_test_plans(self._test_plan_create) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + + get_test_plan_response: TestPlan = client.get_test_plan(created_test_plan.id) + + assert get_test_plan_response is not None + assert isinstance(get_test_plan_response, TestPlan) + assert get_test_plan_response.id == created_test_plan.id + + def test__update_test_plan__returns_updated_test_plan( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_response = create_test_plans(self._test_plan_create) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + + update_test_plans_request = UpdateTestPlansRequest( + test_plans=[ + UpdateTestPlanRequest( + id=created_test_plan.id, + name="Updated Test Plan", + ) + ] + ) + update_test_plans_response = client.update_test_plans( + update_request=update_test_plans_request + ) + + assert update_test_plans_response.updated_test_plans is not None + + updated_test_plan = update_test_plans_response.updated_test_plans[0] + assert updated_test_plan.id == created_test_plan.id + assert updated_test_plan.name == "Updated Test Plan" + + def test__schedule_test_plan__returns_scheduled_test_plan( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_response = create_test_plans(self._test_plan_create) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + + schedule_test_plans_request = ScheduleTestPlansRequest( + test_plans=[ + ScheduleTestPlanRequest( + id=created_test_plan.id, + planned_start_date_time=datetime.strptime( + "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ), + estimated_end_date_time=datetime.strptime( + "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ), + system_id="fake-system", + ) + ] + ) + schedule_test_plans_response = client.schedule_test_plans( + schedule_request=schedule_test_plans_request + ) + + assert schedule_test_plans_response.scheduled_test_plans is not None + + scheduled_test_plan = schedule_test_plans_response.scheduled_test_plans[0] + assert scheduled_test_plan.id == created_test_plan.id + assert scheduled_test_plan.planned_start_date_time == datetime.strptime( + "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ) + assert scheduled_test_plan.system_id == "fake-system" + + def test__query_test_plans__return_queried_test_plan( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_response = create_test_plans(self._test_plan_create) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + + query_test_plans_request = QueryTestPlansRequest( + filter=f'id = "{created_test_plan.id}"', return_count=True + ) + queried_test_plans_response = client.query_test_plans( + query_request=query_test_plans_request + ) + + assert queried_test_plans_response is not None + assert queried_test_plans_response.test_plans[0].id == created_test_plan.id + assert queried_test_plans_response.total_count is not None + assert queried_test_plans_response.total_count > 0 + + def test__query_test_plans_with_projections__returns_the_test_plans_with_projected_properties( + self, client: TestPlanClient, create_test_plans + ): + create_test_plan_response = create_test_plans(self._test_plan_create) + + assert create_test_plan_response.created_test_plans is not None + + created_test_plan = create_test_plan_response.created_test_plans[0] + query_test_plans_request = QueryTestPlansRequest( + filter=f'id = "{created_test_plan.id}"', + projection=[TestPlanField.ID, TestPlanField.NAME], + take=1, + ) + response = client.query_test_plans(query_request=query_test_plans_request) + + assert response is not None + test_plan = response.test_plans[0] + assert ( + test_plan.id is not None + and test_plan.name is not None + and test_plan.template_id is None + and test_plan.state is None + and test_plan.substate is None + and test_plan.description is None + and test_plan.assigned_to is None + and test_plan.work_order_id is None + and test_plan.work_order_name is None + and test_plan.workspace is None + and test_plan.created_by is None + and test_plan.updated_at is None + and test_plan.created_At is None + and test_plan.updated_by is None + and test_plan.properties is None + and test_plan.part_number is None + and test_plan.dut_id is None + and test_plan.test_program is None + and test_plan.system_filter is None + and test_plan.fixture_ids is None + and test_plan.system_id is None + and test_plan.planned_start_date_time is None + and test_plan.estimated_duration_in_seconds is None + and test_plan.estimated_end_date_time is None + and test_plan.file_ids_from_template is None + and test_plan.execution_actions is None + and test_plan.execution_history is None + and test_plan.dashboard_url is None + and test_plan.dashboard is None + ) + + def test__create_test_plan_template__returns_created_test_plan_template( + self, client: TestPlanClient, create_test_plan_templates + ): + create_test_plan_template_response = create_test_plan_templates( + self._create_test_plan_template_request + ) + + template_id = ( + create_test_plan_template_response.created_test_plan_templates[0].id + if create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id + else None + ) + + assert template_id is not None + assert ( + create_test_plan_template_response.created_test_plan_templates[0].name + == self._create_test_plan_template_request[0].name + ) + + def test__query_test_plan_template__returns_queried_test_plan_template( + self, client: TestPlanClient, create_test_plan_templates + ): + create_test_plan_template_response = create_test_plan_templates( + self._create_test_plan_template_request + ) + + template_id = ( + create_test_plan_template_response.created_test_plan_templates[0].id + if create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id + else None + ) + + assert template_id is not None + + query = QueryTestPlanTemplatesRequest(filter=f'id="{template_id}"', take=1) + + query_test_plan_template_response = client.query_test_plan_templates( + query_test_plan_templates=query + ) + + assert len(query_test_plan_template_response.test_plan_templates) == 1, query + assert ( + query_test_plan_template_response.test_plan_templates[0].id == template_id + ) + + def test__delete_test_plan_template(self, client: TestPlanClient): + create_test_plan_template_response: ( + CreateTestPlanTemplatePartialSuccessResponse + ) = client.create_test_plan_templates( + test_plan_templates=self._create_test_plan_template_request + ) + + template_id = ( + create_test_plan_template_response.created_test_plan_templates[0].id + if create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id + else None + ) + + assert template_id is not None + + client.delete_test_plan_templates(ids=[template_id]) + + query_deleted_test_plan_template_response = client.query_test_plan_templates( + query_test_plan_templates=QueryTestPlanTemplatesRequest( + filter=f'id="{template_id}"', take=1 + ) + ) + + assert len(query_deleted_test_plan_template_response.test_plan_templates) == 0 + + def test__query_test_plan_templates_with_projections__returns_test_plan_templates_with_projected_properties( + self, client: TestPlanClient, create_test_plan_templates + ): + create_test_plan_template_response = create_test_plan_templates( + self._create_test_plan_template_request + ) + + template_id = ( + create_test_plan_template_response.created_test_plan_templates[0].id + if create_test_plan_template_response.created_test_plan_templates + and create_test_plan_template_response.created_test_plan_templates[0].id + else None + ) + + assert template_id is not None + + query = QueryTestPlanTemplatesRequest( + filter=f'id="{template_id}"', + projection=[TestPlanTemplateField.ID, TestPlanTemplateField.NAME], + take=1, + ) + print(query) + response = client.query_test_plan_templates(query_test_plan_templates=query) + + assert response is not None + test_plan_template = response.test_plan_templates[0] + assert ( + test_plan_template.id is not None + and test_plan_template.name is not None + and test_plan_template.template_group is None + and test_plan_template.product_families is None + and test_plan_template.part_numbers is None + and test_plan_template.summary is None + and test_plan_template.description is None + and test_plan_template.test_program is None + and test_plan_template.estimated_duration_in_seconds is None + and test_plan_template.system_filter is None + and test_plan_template.execution_actions is None + and test_plan_template.file_ids is None + and test_plan_template.workspace is None + and test_plan_template.properties is None + and test_plan_template.dashboard is None + )