merge: Merge pull request #61 from DSD-DBS/test-parameters

feat: Implement testparameter endpoints

Author: micha91

polarion_rest_api_client/client.py

@@ -26,6 +26,7 @@ class DefaultFields:
     _testrecords: str = "@basic"
     _testruns: str = "@basic"
     _teststeps: str = "@basic"
+    _testparameters: str = "@all"
 
     @property
     def workitems(self) -> dict[str, str]:
@@ -65,7 +66,7 @@ def documents(self, value: str):
 
     @property
     def testruns(self) -> dict[str, str]:
-        """Return the fields dict for document."""
+        """Return the fields dict for testruns."""
         return {"testruns": self._testruns}
 
     @testruns.setter
@@ -74,7 +75,7 @@ def testruns(self, value: str):
 
     @property
     def testrecords(self) -> dict[str, str]:
-        """Return the fields dict for document."""
+        """Return the fields dict for testrecords."""
         return {"testrecords": self._testrecords}
 
     @testrecords.setter
@@ -83,13 +84,22 @@ def testrecords(self, value: str):
 
     @property
     def teststeps(self) -> dict[str, str]:
-        """Return the fields dict for document."""
+        """Return the fields dict for teststeps."""
         return {"teststeps": self._teststeps}
 
     @teststeps.setter
     def teststeps(self, value: str):
         self._teststeps = value
 
+    @property
+    def testparameters(self) -> dict[str, str]:
+        """Return the fields dict for testparameters."""
+        return {"testparameters": self._testparameters}
+
+    @testparameters.setter
+    def testparameters(self, value: str):
+        self._testparameters = value
+
     @property
     def all_types(self) -> dict[str, str]:
         """Return all fields dicts merged together."""
@@ -100,6 +110,7 @@ def all_types(self) -> dict[str, str]:
             | self.documents
             | self.testruns
             | self.testrecords
+            | self.testparameters
         )
 
 

polarion_rest_api_client/clients/test_parameters.py

@@ -0,0 +1,264 @@
+# Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+"""Clients to handle test parameters."""
+
+import itertools
+import typing as t
+
+from polarion_rest_api_client import data_models as dm
+from polarion_rest_api_client.open_api_client import models as api_models
+from polarion_rest_api_client.open_api_client.api.test_records import (
+    delete_test_record_test_parameter,
+    get_test_record_test_parameters,
+    post_test_record_test_parameters,
+)
+from polarion_rest_api_client.open_api_client.api.test_runs import (
+    delete_test_run_test_parameters,
+    get_test_run_test_parameters,
+    post_test_run_test_parameters,
+)
+
+from . import base_classes as bc
+
+
+class TestRunParameters(bc.ItemsClient[dm.TestRunParameter]):
+    """Client to handle TestParameters of a TestRun."""
+
+    def get(self, *args, **kwargs) -> dm.TestRunParameter:
+        raise NotImplementedError
+
+    def get_multi(  # type: ignore[override]
+        self,
+        test_run_id: str,
+        *,
+        page_size: int = 100,
+        page_number: int = 1,
+        fields: dict[str, str] | None = None,
+    ) -> tuple[list[dm.TestRunParameter], bool]:
+        if fields is None:
+            fields = self._client.default_fields.testparameters
+
+        sparse_fields = self._build_sparse_fields(fields)
+        response = get_test_run_test_parameters.sync_detailed(
+            self._project_id,
+            test_run_id,
+            client=self._client.client,
+            fields=sparse_fields,
+            pagenumber=page_number,
+            pagesize=page_size,
+        )
+
+        self._raise_on_error(response)
+        return _parse_get_response(response.parsed, test_run_id)
+
+    def _split_into_batches(
+        self, items: list[dm.TestRunParameter]
+    ) -> t.Generator[list[dm.TestRunParameter], None, None]:
+        for _, group in itertools.groupby(
+            sorted(
+                items,
+                key=lambda x: x.test_run_id,
+            ),
+            lambda x: x.test_run_id,
+        ):
+            yield list(group)
+
+    def _create(self, items: list[dm.TestRunParameter]):
+        """Call only with items with common test_run_id."""
+        test_run_id = items[0].test_run_id
+        body = _build_post_body(items)
+
+        response = post_test_run_test_parameters.sync_detailed(
+            self._project_id,
+            test_run_id,
+            client=self._client.client,
+            body=body,
+        )
+        self._raise_on_error(response)
+
+        assert (
+            isinstance(
+                response.parsed, api_models.TestparametersListPostResponse
+            )
+            and response.parsed.data
+        )
+
+    def _delete(self, items: list[dm.TestRunParameter]):
+        """Call only with items with common test_run_id."""
+        test_run_id = items[0].test_run_id
+        body = api_models.TestparametersListDeleteRequest(
+            data=[
+                api_models.TestparametersListDeleteRequestDataItem(
+                    type_=api_models.TestparametersListDeleteRequestDataItemType.TESTPARAMETERS,
+                    id=f"{self._project_id}/{item.test_run_id}/{item.name}",
+                )
+                for item in items
+            ]
+        )
+
+        response = delete_test_run_test_parameters.sync_detailed(
+            self._project_id,
+            test_run_id,
+            client=self._client.client,
+            body=body,
+        )
+        self._raise_on_error(response)
+
+
+class TestRecordParameters(bc.ItemsClient[dm.TestRecordParameter]):
+    """Clients to handle TestParameters of a TestRecord."""
+
+    def get(self, *args, **kwargs) -> dm.TestRecordParameter:
+        raise NotImplementedError
+
+    def get_multi(  # type: ignore[override]
+        self,
+        test_record: dm.TestRecord,
+        *,
+        page_size: int = 100,
+        page_number: int = 1,
+        fields: dict[str, str] | None = None,
+    ) -> tuple[list[dm.TestRecordParameter], bool]:
+        if fields is None:
+            fields = self._client.default_fields.testparameters
+
+        sparse_fields = self._build_sparse_fields(fields)
+        response = get_test_record_test_parameters.sync_detailed(
+            self._project_id,
+            test_record.test_run_id,
+            test_record.work_item_project_id,
+            test_record.work_item_id,
+            str(test_record.iteration),
+            client=self._client.client,
+            fields=sparse_fields,
+            pagenumber=page_number,
+            pagesize=page_size,
+        )
+
+        self._raise_on_error(response)
+
+        return _parse_get_response(response.parsed, test_record)
+
+    def _split_into_batches(
+        self, items: list[dm.TestRecordParameter]
+    ) -> t.Generator[list[dm.TestRecordParameter], None, None]:
+        for _, group in itertools.groupby(
+            sorted(
+                items,
+                key=lambda x: (
+                    f"{x.test_record.test_run_id}/{x.test_record.work_item_project_id}/{x.test_record.work_item_id}/{x.test_record.iteration}"
+                ),
+            ),
+            lambda x: x.test_record,
+        ):
+            yield list(group)
+
+    def _create(self, items: list[dm.TestRecordParameter]):
+        """Call only with items with common test_records."""
+        test_record = items[0].test_record
+        body = _build_post_body(items)
+
+        response = post_test_record_test_parameters.sync_detailed(
+            self._project_id,
+            test_record.test_run_id,
+            test_record.work_item_project_id,
+            test_record.work_item_id,
+            str(test_record.iteration),
+            client=self._client.client,
+            body=body,
+        )
+        self._raise_on_error(response)
+
+    def _delete(self, items: list[dm.TestRecordParameter]):
+        """We expect only one TestRecordParameter for deletion."""
+        assert len(items) == 1
+        test_record = items[0].test_record
+
+        response = delete_test_record_test_parameter.sync_detailed(
+            self._project_id,
+            test_record.test_run_id,
+            test_record.work_item_project_id,
+            test_record.work_item_id,
+            str(test_record.iteration),
+            items[0].name,
+            client=self._client.client,
+        )
+        self._raise_on_error(response)
+
+    def delete(
+        self, items: dm.TestRecordParameter | list[dm.TestRecordParameter]
+    ):
+        if not isinstance(items, list):
+            items = [items]
+
+        for item in items:
+            self._delete([item])
+
+
+@t.overload
+def _parse_get_response(
+    parsed_response: (
+        api_models.TestparametersListGetResponse | api_models.Errors | None
+    ),
+    scope: dm.TestRecord,
+) -> tuple[list[dm.TestRecordParameter], bool]: ...
+
+
+@t.overload
+def _parse_get_response(
+    parsed_response: (
+        api_models.TestparametersListGetResponse | api_models.Errors | None
+    ),
+    scope: str,
+) -> tuple[list[dm.TestRunParameter], bool]: ...
+
+
+def _parse_get_response(
+    parsed_response: (
+        api_models.TestparametersListGetResponse | api_models.Errors | None
+    ),
+    scope: str | dm.TestRecord,
+) -> tuple[list[dm.TestRecordParameter] | list[dm.TestRunParameter], bool]:
+    assert isinstance(
+        parsed_response, api_models.TestparametersListGetResponse
+    )
+    next_page = isinstance(
+        parsed_response.links, api_models.TestparametersListGetResponseLinks
+    ) and bool(parsed_response.links.next_)
+    if isinstance(scope, str):
+        return [
+            dm.TestRunParameter(
+                test_run_id=scope,
+                name=data.attributes.name or "",
+                value=data.attributes.value or "",
+            )
+            for data in parsed_response.data or []
+            if data.attributes
+        ], next_page
+
+    return [
+        dm.TestRecordParameter(
+            test_record=scope,
+            name=data.attributes.name or "",
+            value=data.attributes.value or "",
+        )
+        for data in parsed_response.data or []
+        if data.attributes
+    ], next_page
+
+
+def _build_post_body(
+    items: list[dm.TestRunParameter] | list[dm.TestRecordParameter],
+) -> api_models.TestparametersListPostRequest:
+    body = api_models.TestparametersListPostRequest(
+        data=[
+            api_models.TestparametersListPostRequestDataItem(
+                type_=api_models.TestparametersListPostRequestDataItemType.TESTPARAMETERS,
+                attributes=api_models.TestparametersListPostRequestDataItemAttributes(
+                    name=item.name, value=item.value
+                ),
+            )
+            for item in items
+        ]
+    )
+    return body

polarion_rest_api_client/clients/test_records.py

@@ -1,5 +1,6 @@
 # Copyright DB InfraGO AG and contributors
 # SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
 
 import itertools
 import typing as t
@@ -14,12 +15,24 @@
 )
 
 from . import base_classes as bc
+from . import test_parameters
+
+if t.TYPE_CHECKING:
+    from polarion_rest_api_client import client as polarion_client
 
 
 class TestRecords(
     bc.SingleUpdatableItemsMixin[dm.TestRecord],
     bc.UpdatableItemsClient[dm.TestRecord],
 ):
+    def __init__(
+        self, project_id: str, client: polarion_client.PolarionClient
+    ):
+        super().__init__(project_id, client)
+        self.parameters = test_parameters.TestRecordParameters(
+            project_id, client
+        )
+
     def get(self, *args, **kwargs) -> dm.TestRecord:
         raise NotImplementedError
 

polarion_rest_api_client/clients/test_runs.py

@@ -12,7 +12,7 @@
 )
 
 from . import base_classes as bc
-from . import test_records
+from . import test_parameters, test_records
 
 if t.TYPE_CHECKING:
     from polarion_rest_api_client import client as polarion_client
@@ -30,6 +30,7 @@ def __init__(
     ):
         super().__init__(project_id, client)
         self.records = test_records.TestRecords(project_id, client)
+        self.parameters = test_parameters.TestRunParameters(project_id, client)
 
     def _update(self, to_update: list[dm.TestRun] | dm.TestRun):
         """Create the given list of test runs."""