diff --git a/nisystemlink/clients/assetmanagement/_asset_management_client.py b/nisystemlink/clients/assetmanagement/_asset_management_client.py index 1ea9b61c..a3f0e707 100644 --- a/nisystemlink/clients/assetmanagement/_asset_management_client.py +++ b/nisystemlink/clients/assetmanagement/_asset_management_client.py @@ -13,6 +13,7 @@ from uplink import Field, Path, retry from . import models +from ..core.helpers._partial_success import unwrap_single_item_partial_success @retry( @@ -56,6 +57,30 @@ def create_assets( """ ... + def create_asset(self, asset: models.CreateAssetRequest) -> models.Asset: + """Create a single asset. + + Args: + asset: The asset to create. + + Returns: + The created asset. + + Raises: + ApiException: if the asset could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_assets([asset]) + + return unwrap_single_item_partial_success( + response=response, + items=response.assets, + failed=response.failed, + error=response.error, + failure_message="Failed to create asset.", + empty_message="Server returned no created assets.", + ) + @post("query-assets") def __query_assets( self, query: models._QueryAssetsRequest diff --git a/nisystemlink/clients/core/helpers/_partial_success.py b/nisystemlink/clients/core/helpers/_partial_success.py new file mode 100644 index 00000000..ee39aa3f --- /dev/null +++ b/nisystemlink/clients/core/helpers/_partial_success.py @@ -0,0 +1,47 @@ +from typing import Any, Sequence, TypeVar + +from nisystemlink.clients import core + +_ItemT = TypeVar("_ItemT") + + +def unwrap_single_item_partial_success( + *, + response: Any | None, + items: Sequence[_ItemT] | None, + failed: Sequence[Any] | None, + error: core.ApiError | None, + failure_message: str, + empty_message: str, +) -> _ItemT: + """Return the first successful item from a partial-success response. + + Raises: + ApiException: if the response reports a failure or contains no successful item. + """ + response_data = ( + response.model_dump(mode="json", by_alias=True) + if response is not None + else None + ) + + if failed or error: + raise core.ApiException( + failure_message, + error=error, + response_data=response_data, + ) + + if not items: + raise core.ApiException( + empty_message, + response_data=response_data, + ) + + if len(items) != 1: + raise core.ApiException( + f"Expected exactly one successful item but received {len(items)}.", + response_data=response_data, + ) + + return items[0] diff --git a/nisystemlink/clients/product/_product_client.py b/nisystemlink/clients/product/_product_client.py index 453b2c69..89346543 100644 --- a/nisystemlink/clients/product/_product_client.py +++ b/nisystemlink/clients/product/_product_client.py @@ -8,6 +8,7 @@ from uplink import Field, Query, retry, returns from . import models +from ..core.helpers._partial_success import unwrap_single_item_partial_success @retry( @@ -50,6 +51,30 @@ def create_products( """ ... + def create_product(self, product: models.CreateProductRequest) -> models.Product: + """Creates a single product. + + Args: + product: The product to create. + + Returns: + The created product. + + Raises: + ApiException: if the product could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_products([product]) + + return unwrap_single_item_partial_success( + response=response, + items=response.products, + failed=response.failed, + error=response.error, + failure_message="Failed to create product.", + empty_message="Server returned no created products.", + ) + @get( "products", args=[Query("continuationToken"), Query("take"), Query("returnCount")], @@ -153,6 +178,33 @@ def update_products( """ ... + def update_product( + self, product: models.UpdateProductRequest, replace: bool = False + ) -> models.Product: + """Updates a single product. + + Args: + product: The product to update. + replace: Replace the existing fields instead of merging them. + + Returns: + The updated product. + + Raises: + ApiException: if the product could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_products([product], replace=replace) + + return unwrap_single_item_partial_success( + response=response, + items=response.products, + failed=response.failed, + error=response.error, + failure_message="Failed to update product.", + empty_message="Server returned no updated products.", + ) + @delete("products/{id}") def delete_product(self, id: str) -> None: """Deletes a single product by id. diff --git a/nisystemlink/clients/spec/_spec_client.py b/nisystemlink/clients/spec/_spec_client.py index d7d8912a..95b24a40 100644 --- a/nisystemlink/clients/spec/_spec_client.py +++ b/nisystemlink/clients/spec/_spec_client.py @@ -8,6 +8,7 @@ from uplink import Field, retry from . import models +from ..core.helpers._partial_success import unwrap_single_item_partial_success @retry( @@ -63,6 +64,32 @@ def create_specs( """ ... + def create_spec( + self, spec: models.CreateSpecificationsRequestObject + ) -> models.CreatedSpecification: + """Creates a single specification. + + Args: + spec: The specification to create. + + Returns: + The created specification. + + Raises: + ApiException: if the specification could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_specs(models.CreateSpecificationsRequest(specs=[spec])) + + return unwrap_single_item_partial_success( + response=response, + items=response.created_specs, + failed=response.failed_specs, + error=response.error, + failure_message="Failed to create spec.", + empty_message="Server returned no created specs.", + ) + @post("delete-specs", args=[Field("ids")]) def delete_specs( self, ids: List[str] @@ -129,3 +156,29 @@ def update_specs( with error messages for updates that failed. """ ... + + def update_spec( + self, spec: models.UpdateSpecificationsRequestObject + ) -> models.UpdatedSpecification: + """Updates a single specification. + + Args: + spec: The specification to update. + + Returns: + The updated specification. + + Raises: + ApiException: if the specification could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_specs(models.UpdateSpecificationsRequest(specs=[spec])) + + return unwrap_single_item_partial_success( + response=response, + items=response.updated_specs if response is not None else None, + failed=response.failed_specs if response is not None else None, + error=response.error if response is not None else None, + failure_message="Failed to update spec.", + empty_message="Server returned no updated specs.", + ) diff --git a/nisystemlink/clients/test_plan/_test_plan_client.py b/nisystemlink/clients/test_plan/_test_plan_client.py index a730d9e5..33f97ecb 100644 --- a/nisystemlink/clients/test_plan/_test_plan_client.py +++ b/nisystemlink/clients/test_plan/_test_plan_client.py @@ -8,6 +8,8 @@ from nisystemlink.clients.test_plan import models from uplink import Field, retry +from ..core.helpers._partial_success import unwrap_single_item_partial_success + @retry( when=retry.when.status(408, 429, 502, 503, 504), @@ -66,6 +68,32 @@ def create_test_plans( """ ... + def create_test_plan( + self, test_plan: models.CreateTestPlanRequest + ) -> models.TestPlan: + """Create a single test plan. + + Args: + test_plan: The test plan to create. + + Returns: + The created test plan. + + Raises: + ApiException: if the test plan could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_test_plans([test_plan]) + + return unwrap_single_item_partial_success( + response=response, + items=response.created_test_plans, + failed=response.failed_test_plans, + error=response.error, + failure_message="Failed to create test plan.", + empty_message="Server returned no created test plans.", + ) + @post("delete-testplans", args=[Field("ids")]) def delete_test_plans(self, ids: List[str]) -> None: """Delete test plans by IDs. @@ -106,6 +134,37 @@ def schedule_test_plans( """ ... + def schedule_test_plan( + self, + test_plan: models.ScheduleTestPlanRequest, + replace: bool | None = None, + ) -> models.TestPlan: + """Schedule a single test plan. + + Args: + test_plan: The test plan schedule request. + replace: Whether to replace the existing scheduled test plan. + + Returns: + The scheduled test plan. + + Raises: + ApiException: if the test plan could not be scheduled or the service returns an + unexpected partial-success payload. + """ + response = self.schedule_test_plans( + models.ScheduleTestPlansRequest(test_plans=[test_plan], replace=replace) + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.scheduled_test_plans, + failed=response.failed_test_plans, + error=response.error, + failure_message="Failed to schedule test plan.", + empty_message="Server returned no scheduled test plans.", + ) + @post("update-testplans") def update_test_plans( self, update_request: models.UpdateTestPlansRequest @@ -120,6 +179,35 @@ def update_test_plans( """ ... + def update_test_plan( + self, test_plan: models.UpdateTestPlanRequest, replace: bool | None = None + ) -> models.TestPlan: + """Update a single test plan. + + Args: + test_plan: The test plan to update. + replace: Whether to replace the existing test plan instead of merging updates. + + Returns: + The updated test plan. + + Raises: + ApiException: if the test plan could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_test_plans( + models.UpdateTestPlansRequest(test_plans=[test_plan], replace=replace) + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.updated_test_plans, + failed=response.failed_test_plans, + error=response.error, + failure_message="Failed to update test plan.", + empty_message="Server returned no updated test plans.", + ) + @post("testplan-templates", args=[Field("testPlanTemplates")]) def create_test_plan_templates( self, test_plan_templates: List[models.CreateTestPlanTemplateRequest] @@ -137,6 +225,32 @@ def create_test_plan_templates( """ ... + def create_test_plan_template( + self, test_plan_template: models.CreateTestPlanTemplateRequest + ) -> models.TestPlanTemplate: + """Creates a single test plan template. + + Args: + test_plan_template: The test plan template to create. + + Returns: + The created test plan template. + + Raises: + ApiException: if the test plan template could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_test_plan_templates([test_plan_template]) + + return unwrap_single_item_partial_success( + response=response, + items=response.created_test_plan_templates, + failed=response.failed_test_plan_templates, + error=response.error, + failure_message="Failed to create test plan template.", + empty_message="Server returned no created test plan templates.", + ) + @post("query-testplan-templates") def query_test_plan_templates( self, query_test_plan_templates: models.QueryTestPlanTemplatesRequest diff --git a/nisystemlink/clients/testmonitor/_test_monitor_client.py b/nisystemlink/clients/testmonitor/_test_monitor_client.py index 416d1a93..c64eba99 100644 --- a/nisystemlink/clients/testmonitor/_test_monitor_client.py +++ b/nisystemlink/clients/testmonitor/_test_monitor_client.py @@ -12,6 +12,7 @@ from uplink import Field, Path, Query, retry, returns from . import models +from ..core.helpers._partial_success import unwrap_single_item_partial_success @retry( @@ -69,6 +70,30 @@ def create_results( """ ... + def create_result(self, result: CreateResultRequest) -> models.Result: + """Creates a single result. + + Args: + result: The result to create. + + Returns: + The created result. + + Raises: + ApiException: if the result could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_results([result]) + + return unwrap_single_item_partial_success( + response=response, + items=response.results, + failed=response.failed, + error=response.error, + failure_message="Failed to create result.", + empty_message="Server returned no created results.", + ) + @get( "results", args=[Query("continuationToken"), Query("take"), Query("returnCount")], @@ -168,6 +193,33 @@ def update_results( """ ... + def update_result( + self, result: UpdateResultRequest, replace: bool = False + ) -> models.Result: + """Updates a single result. + + Args: + result: The result to update. + replace: Replace the existing fields instead of merging them. + + Returns: + The updated result. + + Raises: + ApiException: if the result could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_results([result], replace=replace) + + return unwrap_single_item_partial_success( + response=response, + items=response.results, + failed=response.failed, + error=response.error, + failure_message="Failed to update result.", + empty_message="Server returned no updated results.", + ) + @delete("results/{id}") def delete_result(self, id: str) -> None: """Deletes a single result by id. @@ -225,6 +277,38 @@ def create_steps( """ ... + def create_step( + self, + step: models.CreateStepRequest, + update_result_total_time: bool = False, + ) -> models.Step: + """Creates a single step. + + Args: + step: The step to create. + update_result_total_time: Determine test result total time from the step total times. + + Returns: + The created step. + + Raises: + ApiException: if the step could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_steps( + [step], + update_result_total_time=update_result_total_time, + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.steps, + failed=response.failed, + error=response.error, + failure_message="Failed to create step.", + empty_message="Server returned no created steps.", + ) + @post("delete-steps", args=[Field("steps")]) def delete_steps( self, steps: List[models.StepIdResultIdPair] @@ -327,6 +411,44 @@ def update_steps( """ ... + def update_step( + self, + step: models.UpdateStepRequest, + update_result_total_time: bool = False, + replace_keywords: bool = False, + replace_properties: bool = False, + ) -> models.Step: + """Updates a single step. + + Args: + step: The step to update. + update_result_total_time: Determine test result total time from the step total times. + replace_keywords: Replace existing keywords instead of merging them. + replace_properties: Replace existing properties instead of merging them. + + Returns: + The updated step. + + Raises: + ApiException: if the step could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_steps( + [step], + update_result_total_time=update_result_total_time, + replace_keywords=replace_keywords, + replace_properties=replace_properties, + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.steps, + failed=response.failed, + error=response.error, + failure_message="Failed to update step.", + empty_message="Server returned no updated steps.", + ) + @get( "steps", args=[Query("continuationToken"), Query("take"), Query("returnCount")], diff --git a/nisystemlink/clients/work_item/_work_item_client.py b/nisystemlink/clients/work_item/_work_item_client.py index 4c1c54ba..376b754c 100644 --- a/nisystemlink/clients/work_item/_work_item_client.py +++ b/nisystemlink/clients/work_item/_work_item_client.py @@ -7,6 +7,8 @@ from nisystemlink.clients.work_item import models from uplink import Field, Path, retry +from ..core.helpers._partial_success import unwrap_single_item_partial_success + class WorkItemExecuteApiException(core.ApiException): """Raised when the execute work item API returns an error with a structured response body. @@ -87,6 +89,32 @@ def create_work_items( """ ... + def create_work_item( + self, work_item: models.CreateWorkItemRequest + ) -> models.WorkItem: + """Creates a single work item. + + Args: + work_item: The work item to create. + + Returns: + The created work item. + + Raises: + ApiException: if the work item could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_work_items([work_item]) + + return unwrap_single_item_partial_success( + response=response, + items=response.created_work_items, + failed=response.failed_work_items, + error=response.error, + failure_message="Failed to create work item.", + empty_message="Server returned no created work items.", + ) + @post("query-workitems") def query_work_items( self, query_work_items: models.QueryWorkItemsRequest @@ -121,6 +149,37 @@ def schedule_work_items( """ ... + def schedule_work_item( + self, + work_item: models.ScheduleWorkItemRequest, + replace: bool | None = None, + ) -> models.WorkItem: + """Schedules a single work item. + + Args: + work_item: The work item schedule request. + replace: When true, existing array fields are replaced instead of merged. + + Returns: + The scheduled work item. + + Raises: + ApiException: if the work item could not be scheduled or the service returns an + unexpected partial-success payload. + """ + response = self.schedule_work_items( + models.ScheduleWorkItemsRequest(work_items=[work_item], replace=replace) + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.scheduled_work_items, + failed=response.failed_work_items, + error=response.error, + failure_message="Failed to schedule work item.", + empty_message="Server returned no scheduled work items.", + ) + @post("update-workitems") def update_work_items( self, update_work_items: models.UpdateWorkItemsRequest @@ -138,6 +197,37 @@ def update_work_items( """ ... + def update_work_item( + self, + work_item: models.UpdateWorkItemRequest, + replace: bool | None = None, + ) -> models.WorkItem: + """Updates a single work item. + + Args: + work_item: The work item to update. + replace: When true, existing array and key-value pair fields are replaced instead of merged. + + Returns: + The updated work item. + + Raises: + ApiException: if the work item could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_work_items( + models.UpdateWorkItemsRequest(work_items=[work_item], replace=replace) + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.updated_work_items, + failed=response.failed_work_items, + error=response.error, + failure_message="Failed to update work item.", + empty_message="Server returned no updated work items.", + ) + @post("delete-workitems", args=[Field("ids")]) def delete_work_items( self, ids: List[str] @@ -221,6 +311,32 @@ def create_work_item_templates( """ ... + def create_work_item_template( + self, work_item_template: models.CreateWorkItemTemplateRequest + ) -> models.WorkItemTemplate: + """Creates a single work item template. + + Args: + work_item_template: The work item template to create. + + Returns: + The created work item template. + + Raises: + ApiException: if the work item template could not be created or the service returns an + unexpected partial-success payload. + """ + response = self.create_work_item_templates([work_item_template]) + + return unwrap_single_item_partial_success( + response=response, + items=response.created_work_item_templates, + failed=response.failed_work_item_templates, + error=response.error, + failure_message="Failed to create work item template.", + empty_message="Server returned no created work item templates.", + ) + @post("query-workitem-templates") def query_work_item_templates( self, query_work_item_templates: models.QueryWorkItemTemplatesRequest @@ -255,6 +371,40 @@ def update_work_item_templates( """ ... + def update_work_item_template( + self, + work_item_template: models.UpdateWorkItemTemplateRequest, + replace: bool | None = None, + ) -> models.WorkItemTemplate: + """Updates a single work item template. + + Args: + work_item_template: The work item template to update. + replace: When true, existing key-value pair fields are replaced instead of merged. + + Returns: + The updated work item template. + + Raises: + ApiException: if the work item template could not be updated or the service returns an + unexpected partial-success payload. + """ + response = self.update_work_item_templates( + models.UpdateWorkItemTemplatesRequest( + work_item_templates=[work_item_template], + replace=replace, + ) + ) + + return unwrap_single_item_partial_success( + response=response, + items=response.updated_work_item_templates, + failed=response.failed_work_item_templates, + error=response.error, + failure_message="Failed to update work item template.", + empty_message="Server returned no updated work item templates.", + ) + @post("delete-workitem-templates", args=[Field("ids")]) def delete_work_item_templates( self, ids: List[str] diff --git a/tests/core/test_partial_success.py b/tests/core/test_partial_success.py new file mode 100644 index 00000000..7e64da62 --- /dev/null +++ b/tests/core/test_partial_success.py @@ -0,0 +1,100 @@ +import pytest +from nisystemlink.clients.core import ApiError, ApiException +from nisystemlink.clients.core.helpers._partial_success import ( + unwrap_single_item_partial_success, +) + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def model_dump(self, *, mode, by_alias): + assert mode == "json" + assert by_alias is True + return self._payload + + +def test__unwrap_single_item_partial_success__returns_first_item(): + """Return the first item when the partial-success response is successful.""" + response = _FakeResponse({"items": ["created-item"]}) + + created_item = unwrap_single_item_partial_success( + response=response, + items=["created-item"], + failed=None, + error=None, + failure_message="Failed to create item.", + empty_message="Server returned no created items.", + ) + + assert created_item == "created-item" + + +def test__unwrap_single_item_partial_success__raises_on_partial_failure(): + """Raise ApiException when the response reports a structured partial failure.""" + response = _FakeResponse( + { + "items": [], + "failed": [{"id": "request-id"}], + "error": {"message": "Create failed"}, + } + ) + error = ApiError(message="Create failed") + + with pytest.raises(ApiException) as exc_info: + unwrap_single_item_partial_success( + response=response, + items=[], + failed=[{"id": "request-id"}], + error=error, + failure_message="Failed to create item.", + empty_message="Server returned no created items.", + ) + + assert exc_info.value.error == error + assert exc_info.value.response_data == response.model_dump( + mode="json", by_alias=True + ) + + +def test__unwrap_single_item_partial_success__raises_on_empty_success_payload(): + """Raise ApiException when the response succeeds but contains no created item.""" + response = _FakeResponse({"items": []}) + + with pytest.raises( + ApiException, match="Server returned no created items" + ) as exc_info: + unwrap_single_item_partial_success( + response=response, + items=[], + failed=None, + error=None, + failure_message="Failed to create item.", + empty_message="Server returned no created items.", + ) + + assert exc_info.value.response_data == response.model_dump( + mode="json", by_alias=True + ) + + +def test__unwrap_single_item_partial_success__raises_on_multiple_success_items(): + """Raise ApiException when the response unexpectedly contains multiple items.""" + response = _FakeResponse({"items": ["created-item", "extra-item"]}) + + with pytest.raises( + ApiException, match="Expected exactly one successful item but received 2" + ) as exc_info: + unwrap_single_item_partial_success( + response=response, + items=["created-item", "extra-item"], + failed=None, + error=None, + failure_message="Failed to create item.", + empty_message="Server returned no created items.", + ) + + assert exc_info.value.response_data == response.model_dump( + mode="json", by_alias=True + ) diff --git a/tests/test_single_item_client_convenience_methods.py b/tests/test_single_item_client_convenience_methods.py new file mode 100644 index 00000000..0765365b --- /dev/null +++ b/tests/test_single_item_client_convenience_methods.py @@ -0,0 +1,420 @@ +import pytest +from nisystemlink.clients.assetmanagement import AssetManagementClient +from nisystemlink.clients.assetmanagement import models as asset_models +from nisystemlink.clients.core import ApiException +from nisystemlink.clients.product import models as product_models +from nisystemlink.clients.product import ProductClient +from nisystemlink.clients.spec import models as spec_models +from nisystemlink.clients.spec import SpecClient +from nisystemlink.clients.test_plan import models as test_plan_models +from nisystemlink.clients.test_plan import TestPlanClient +from nisystemlink.clients.testmonitor import models as testmonitor_models +from nisystemlink.clients.testmonitor import TestMonitorClient +from nisystemlink.clients.work_item import models as work_item_models +from nisystemlink.clients.work_item import WorkItemClient + + +class TestProductClientSingleItemConvenience: + def test__create_product__wraps_bulk_create(self): + request = product_models.CreateProductRequest(part_number="part-number") + created_product = product_models.Product.model_construct(id="product-id") + captured_products = [] + client = object.__new__(ProductClient) + + def fake_create_products(products): + captured_products.append(products) + return product_models.CreateProductsPartialSuccess( + products=[created_product] + ) + + setattr(client, "create_products", fake_create_products) + + response = client.create_product(request) + + assert response == created_product + assert captured_products == [[request]] + + def test__update_product__wraps_bulk_update(self): + request = product_models.UpdateProductRequest(id="product-id") + updated_product = product_models.Product.model_construct(id="product-id") + captured_calls = [] + client = object.__new__(ProductClient) + + def fake_update_products(products, replace=False): + captured_calls.append((products, replace)) + return product_models.CreateProductsPartialSuccess( + products=[updated_product] + ) + + setattr(client, "update_products", fake_update_products) + + response = client.update_product(request, replace=True) + + assert response == updated_product + assert captured_calls == [([request], True)] + + +class TestSpecClientSingleItemConvenience: + def test__create_spec__wraps_bulk_create(self): + request = spec_models.CreateSpecificationsRequestObject.model_construct( + product_id="product-id", + spec_id="spec-id", + ) + created_spec = spec_models.CreatedSpecification.model_construct(id="spec-id") + captured_requests = [] + client = object.__new__(SpecClient) + + def fake_create_specs(specs): + captured_requests.append(specs) + return spec_models.CreateSpecificationsPartialSuccess( + created_specs=[created_spec] + ) + + setattr(client, "create_specs", fake_create_specs) + + response = client.create_spec(request) + + assert response == created_spec + assert captured_requests[0].specs == [request] + + def test__update_spec__wraps_bulk_update(self): + request = spec_models.UpdateSpecificationsRequestObject.model_construct( + id="spec-id", + product_id="product-id", + spec_id="spec-id", + ) + updated_spec = spec_models.UpdatedSpecification.model_construct(id="spec-id") + captured_requests = [] + client = object.__new__(SpecClient) + + def fake_update_specs(specs): + captured_requests.append(specs) + return spec_models.UpdateSpecificationsPartialSuccess( + updated_specs=[updated_spec] + ) + + setattr(client, "update_specs", fake_update_specs) + + response = client.update_spec(request) + + assert response == updated_spec + assert captured_requests[0].specs == [request] + + def test__update_spec__raises_on_missing_success_payload(self): + request = spec_models.UpdateSpecificationsRequestObject.model_construct( + id="spec-id", + product_id="product-id", + spec_id="spec-id", + ) + client = object.__new__(SpecClient) + + def fake_update_specs(_specs): + return None + + setattr(client, "update_specs", fake_update_specs) + + with pytest.raises( + ApiException, match="Server returned no updated specs" + ) as exc_info: + client.update_spec(request) + + assert exc_info.value.response_data is None + + +class TestAssetManagementClientSingleItemConvenience: + def test__create_asset__wraps_bulk_create(self): + request = asset_models.CreateAssetRequest.model_construct( + location=asset_models.AssetLocationForCreate.model_construct() + ) + created_asset = asset_models.Asset.model_construct(id="asset-id") + captured_assets = [] + client = object.__new__(AssetManagementClient) + + def fake_create_assets(assets): + captured_assets.append(assets) + return asset_models.CreateAssetsPartialSuccessResponse( + assets=[created_asset] + ) + + setattr(client, "create_assets", fake_create_assets) + + response = client.create_asset(request) + + assert response == created_asset + assert captured_assets == [[request]] + + +class TestTestPlanClientSingleItemConvenience: + def test__create_test_plan__wraps_bulk_create(self): + request = test_plan_models.CreateTestPlanRequest(name="test-plan") + created_test_plan = test_plan_models.TestPlan.model_construct(id="test-plan-id") + captured_test_plans = [] + client = object.__new__(TestPlanClient) + + def fake_create_test_plans(test_plans): + captured_test_plans.append(test_plans) + return test_plan_models.CreateTestPlansPartialSuccessResponse( + created_test_plans=[created_test_plan] + ) + + setattr(client, "create_test_plans", fake_create_test_plans) + + response = client.create_test_plan(request) + + assert response == created_test_plan + assert captured_test_plans == [[request]] + + def test__update_test_plan__wraps_bulk_update(self): + request = test_plan_models.UpdateTestPlanRequest(id="test-plan-id") + updated_test_plan = test_plan_models.TestPlan.model_construct(id="test-plan-id") + captured_requests = [] + client = object.__new__(TestPlanClient) + + def fake_update_test_plans(update_request): + captured_requests.append(update_request) + return test_plan_models.UpdateTestPlansResponse( + updated_test_plans=[updated_test_plan] + ) + + setattr(client, "update_test_plans", fake_update_test_plans) + + response = client.update_test_plan(request, replace=True) + + assert response == updated_test_plan + assert captured_requests[0].test_plans == [request] + assert captured_requests[0].replace is True + + def test__schedule_test_plan__wraps_bulk_schedule(self): + request = test_plan_models.ScheduleTestPlanRequest(id="test-plan-id") + scheduled_test_plan = test_plan_models.TestPlan.model_construct( + id="test-plan-id" + ) + captured_requests = [] + client = object.__new__(TestPlanClient) + + def fake_schedule_test_plans(schedule_request): + captured_requests.append(schedule_request) + return test_plan_models.ScheduleTestPlansResponse( + scheduled_test_plans=[scheduled_test_plan] + ) + + setattr(client, "schedule_test_plans", fake_schedule_test_plans) + + response = client.schedule_test_plan(request, replace=True) + + assert response == scheduled_test_plan + assert captured_requests[0].test_plans == [request] + assert captured_requests[0].replace is True + + def test__create_test_plan_template__wraps_bulk_create(self): + request = test_plan_models.CreateTestPlanTemplateRequest.model_construct( + name="template", + template_group="group", + ) + created_template = test_plan_models.TestPlanTemplate.model_construct( + id="template-id" + ) + captured_templates = [] + client = object.__new__(TestPlanClient) + + def fake_create_test_plan_templates(test_plan_templates): + captured_templates.append(test_plan_templates) + return test_plan_models.CreateTestPlanTemplatePartialSuccessResponse( + created_test_plan_templates=[created_template] + ) + + setattr(client, "create_test_plan_templates", fake_create_test_plan_templates) + + response = client.create_test_plan_template(request) + + assert response == created_template + assert captured_templates == [[request]] + + +class TestWorkItemClientSingleItemConvenience: + def test__create_work_item__wraps_bulk_create(self): + request = work_item_models.CreateWorkItemRequest(name="work-item") + created_work_item = work_item_models.WorkItem.model_construct(id="work-item-id") + captured_work_items = [] + client = object.__new__(WorkItemClient) + + def fake_create_work_items(work_items): + captured_work_items.append(work_items) + return work_item_models.CreateWorkItemsPartialSuccessResponse( + created_work_items=[created_work_item] + ) + + setattr(client, "create_work_items", fake_create_work_items) + + response = client.create_work_item(request) + + assert response == created_work_item + assert captured_work_items == [[request]] + + def test__update_work_item__wraps_bulk_update(self): + request = work_item_models.UpdateWorkItemRequest(id="work-item-id") + updated_work_item = work_item_models.WorkItem.model_construct(id="work-item-id") + captured_requests = [] + client = object.__new__(WorkItemClient) + + def fake_update_work_items(update_work_items): + captured_requests.append(update_work_items) + return work_item_models.UpdateWorkItemsPartialSuccessResponse( + updated_work_items=[updated_work_item] + ) + + setattr(client, "update_work_items", fake_update_work_items) + + response = client.update_work_item(request, replace=True) + + assert response == updated_work_item + assert captured_requests[0].work_items == [request] + assert captured_requests[0].replace is True + + def test__schedule_work_item__wraps_bulk_schedule(self): + request = work_item_models.ScheduleWorkItemRequest(id="work-item-id") + scheduled_work_item = work_item_models.WorkItem.model_construct( + id="work-item-id" + ) + captured_requests = [] + client = object.__new__(WorkItemClient) + + def fake_schedule_work_items(schedule_work_items): + captured_requests.append(schedule_work_items) + return work_item_models.ScheduleWorkItemsPartialSuccessResponse( + scheduled_work_items=[scheduled_work_item] + ) + + setattr(client, "schedule_work_items", fake_schedule_work_items) + + response = client.schedule_work_item(request, replace=True) + + assert response == scheduled_work_item + assert captured_requests[0].work_items == [request] + assert captured_requests[0].replace is True + + def test__create_work_item_template__wraps_bulk_create(self): + request = work_item_models.CreateWorkItemTemplateRequest( + name="template", + template_group="group", + type="type", + ) + created_template = work_item_models.WorkItemTemplate.model_construct( + id="template-id" + ) + captured_templates = [] + client = object.__new__(WorkItemClient) + + def fake_create_work_item_templates(work_item_templates): + captured_templates.append(work_item_templates) + return work_item_models.CreateWorkItemTemplatesPartialSuccessResponse( + created_work_item_templates=[created_template] + ) + + setattr(client, "create_work_item_templates", fake_create_work_item_templates) + + response = client.create_work_item_template(request) + + assert response == created_template + assert captured_templates == [[request]] + + def test__update_work_item_template__wraps_bulk_update(self): + request = work_item_models.UpdateWorkItemTemplateRequest(id="template-id") + updated_template = work_item_models.WorkItemTemplate.model_construct( + id="template-id" + ) + captured_requests = [] + client = object.__new__(WorkItemClient) + + def fake_update_work_item_templates(update_work_item_templates): + captured_requests.append(update_work_item_templates) + return work_item_models.UpdateWorkItemTemplatesPartialSuccessResponse( + updated_work_item_templates=[updated_template] + ) + + setattr(client, "update_work_item_templates", fake_update_work_item_templates) + + response = client.update_work_item_template(request, replace=True) + + assert response == updated_template + assert captured_requests[0].work_item_templates == [request] + assert captured_requests[0].replace is True + + +class TestTestMonitorClientSingleItemConvenience: + def test__update_result__wraps_bulk_update(self): + request = testmonitor_models.UpdateResultRequest(id="result-id") + updated_result = testmonitor_models.Result.model_construct(id="result-id") + captured_calls = [] + client = object.__new__(TestMonitorClient) + + def fake_update_results(results, replace=False): + captured_calls.append((results, replace)) + return testmonitor_models.UpdateResultsPartialSuccess( + results=[updated_result] + ) + + setattr(client, "update_results", fake_update_results) + + response = client.update_result(request, replace=True) + + assert response == updated_result + assert captured_calls == [([request], True)] + + def test__create_step__wraps_bulk_create(self): + request = testmonitor_models.CreateStepRequest( + name="step", + result_id="result-id", + step_id="step-id", + ) + created_step = testmonitor_models.Step.model_construct(id="step-id") + captured_calls = [] + client = object.__new__(TestMonitorClient) + + def fake_create_steps(steps, update_result_total_time=False): + captured_calls.append((steps, update_result_total_time)) + return testmonitor_models.CreateStepsPartialSuccess(steps=[created_step]) + + setattr(client, "create_steps", fake_create_steps) + + response = client.create_step(request, update_result_total_time=True) + + assert response == created_step + assert captured_calls == [([request], True)] + + def test__update_step__wraps_bulk_update(self): + request = testmonitor_models.UpdateStepRequest( + result_id="result-id", + step_id="step-id", + ) + updated_step = testmonitor_models.Step.model_construct(id="step-id") + captured_calls = [] + client = object.__new__(TestMonitorClient) + + def fake_update_steps( + steps, + update_result_total_time=False, + replace_keywords=False, + replace_properties=False, + ): + captured_calls.append( + ( + steps, + update_result_total_time, + replace_keywords, + replace_properties, + ) + ) + return testmonitor_models.UpdateStepsPartialSuccess(steps=[updated_step]) + + setattr(client, "update_steps", fake_update_steps) + + response = client.update_step( + request, + update_result_total_time=True, + replace_keywords=True, + replace_properties=True, + ) + + assert response == updated_step + assert captured_calls == [([request], True, True, True)] diff --git a/tests/testmonitor/test_testmonitor_client.py b/tests/testmonitor/test_testmonitor_client.py new file mode 100644 index 00000000..cc490326 --- /dev/null +++ b/tests/testmonitor/test_testmonitor_client.py @@ -0,0 +1,88 @@ +"""Tests for TestMonitorClient convenience methods.""" + +import pytest +from nisystemlink.clients.core import ApiError, ApiException +from nisystemlink.clients.testmonitor import TestMonitorClient +from nisystemlink.clients.testmonitor.models import ( + CreateResultRequest, + CreateResultsPartialSuccess, + Result, + Status, +) + + +class TestTestMonitorClient: + """Test cases for TestMonitorClient convenience methods.""" + + def test__create_result__returns_created_result(self): + """Test that create_result returns the created result on success.""" + request = CreateResultRequest( + part_number="Part Number", + program_name="Program Name", + status=Status.PASSED(), + ) + created_result = Result( + id="result-id", + part_number=request.part_number, + program_name=request.program_name, + status=request.status, + ) + captured_results = [] + client = object.__new__(TestMonitorClient) + + def fake_create_results(results): + captured_results.append(results) + return CreateResultsPartialSuccess(results=[created_result]) + + client.create_results = fake_create_results # type: ignore[method-assign] + + response = client.create_result(request) + + assert response == created_result + assert captured_results == [[request]] + + def test__create_result__raises_api_exception_on_partial_failure(self): + """Test that create_result raises with the structured error on failure.""" + request = CreateResultRequest( + part_number="Part Number", + program_name="Program Name", + status=Status.PASSED(), + ) + error = ApiError(message="Create failed") + client = object.__new__(TestMonitorClient) + + def fake_create_results(results): + return CreateResultsPartialSuccess( + results=[], + failed=results, + error=error, + ) + + client.create_results = fake_create_results # type: ignore[method-assign] + + with pytest.raises(ApiException) as exc_info: + client.create_result(request) + + assert exc_info.value.error == error + assert exc_info.value.response_data == { + "results": [], + "failed": [request.model_dump(mode="json", by_alias=True)], + "error": error.model_dump(mode="json", by_alias=True), + } + + def test__create_result__raises_on_missing_created_result(self): + """Test that create_result rejects an unexpected empty success payload.""" + request = CreateResultRequest( + part_number="Part Number", + program_name="Program Name", + status=Status.PASSED(), + ) + client = object.__new__(TestMonitorClient) + + def fake_create_results(_results): + return CreateResultsPartialSuccess(results=[]) + + client.create_results = fake_create_results # type: ignore[method-assign] + + with pytest.raises(ApiException, match="Server returned no created results"): + client.create_result(request)