Skip to content

Commit

Permalink
feat: add get apis
Browse files Browse the repository at this point in the history
  • Loading branch information
micha91 committed Mar 28, 2024
1 parent 747c883 commit 4147428
Show file tree
Hide file tree
Showing 3 changed files with 240 additions and 19 deletions.
86 changes: 86 additions & 0 deletions polarion_rest_api_client/base_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ class DefaultFields:
_linkedworkitems: str = "id,role,suspect"
_workitem_attachments: str = "@basic"
_documents: str = "@basic"
_testrecords: str = "@basic"
_testruns: str = "@basic"

@property
def workitems(self):
Expand Down Expand Up @@ -55,6 +57,24 @@ def documents(self):
def documents(self, value):
self._documents = value

@property
def testruns(self):
"""Return the fields dict for document."""
return {"testruns": self._testruns}

@testruns.setter
def testruns(self, value):
self._testruns = value

@property
def testrecords(self):
"""Return the fields dict for document."""
return {"testrecords": self._testrecords}

@testrecords.setter
def testrecords(self, value):
self._testrecords = value

@property
def all_types(self):
"""Return all fields dicts merged together."""
Expand All @@ -63,6 +83,8 @@ def all_types(self):
| self.workitems
| self.linkedworkitems
| self.documents
| self.testruns
| self.testrecords
)


Expand Down Expand Up @@ -360,3 +382,67 @@ def delete_work_item_link(self, work_item_link: dm.WorkItemLink):
"""Delete the links between the work items in work_item_link."""
self._set_project(work_item_link)
self._delete_work_item_links([work_item_link])

def get_all_test_runs(
self,
query: str,
fields: dict[str, str] | None = None,
) -> list[dm.TestRun]:
"""Get all test runs matching the given query.
Will handle pagination automatically. Define a fields dictionary
as described in the Polarion API documentation to get certain
fields.
"""
return self._request_all_items(
self.get_test_runs, fields=fields, query=query
)

@abc.abstractmethod
def get_test_runs(
self,
query: str,
fields: dict[str, str] | None = None,
page_size: int = 100,
page_number: int = 1,
retry: bool = True,
) -> tuple[list[dm.TestRun], bool]:
"""Return the test runs on a defined page matching the given query.
In addition, a flag whether a next page is available is
returned. Define a fields dictionary as described in the
Polarion API documentation to get certain fields.
"""
raise NotImplementedError

def get_all_test_records(
self,
test_run_id: str,
fields: dict[str, str] | None = None,
) -> list[dm.TestRecord]:
"""Get all test records matching the given query.
Will handle pagination automatically. Define a fields dictionary
as described in the Polarion API documentation to get certain
fields.
"""
return self._request_all_items(
self.get_test_records, fields=fields, test_run_id=test_run_id
)

@abc.abstractmethod
def get_test_records(
self,
test_run_id: str,
fields: dict[str, str] | None = None,
page_size: int = 100,
page_number: int = 1,
retry: bool = True,
) -> tuple[list[dm.TestRecord], bool]:
"""Return the test records on a defined page matching the given query.
In addition, a flag whether a next page is available is
returned. Define a fields dictionary as described in the
Polarion API documentation to get certain fields.
"""
raise NotImplementedError
162 changes: 146 additions & 16 deletions polarion_rest_api_client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,14 @@
post_linked_work_items,
)
from polarion_rest_api_client.open_api_client.api.projects import get_project
from polarion_rest_api_client.open_api_client.api.test_records import (
get_test_records,
post_test_records,
)
from polarion_rest_api_client.open_api_client.api.test_runs import (
get_test_runs,
post_test_runs,
)
from polarion_rest_api_client.open_api_client.api.work_item_attachments import ( # pylint: disable=line-too-long
delete_work_item_attachment,
get_work_item_attachments,
Expand Down Expand Up @@ -740,7 +748,7 @@ def get_document(
if not getattr(data.meta, "errors", []):
assert (attributes := data.attributes)
assert isinstance(data.id, str)
home_page_content = self._handle_home_page_content(
home_page_content = self._handle_text_content(
attributes.home_page_content
)

Expand All @@ -754,27 +762,19 @@ def get_document(
)
return None

def _handle_home_page_content(
def _handle_text_content(
self,
home_page_content: api_models.DocumentsSingleGetResponseDataAttributesHomePageContent
polarion_content: api_models.DocumentsSingleGetResponseDataAttributesHomePageContent
| api_models.TestrecordsListGetResponseDataItemAttributesComment
| api_models.TestrunsListGetResponseDataItemAttributesHomePageContent
| oa_types.Unset,
) -> dm.TextContent | None:
if isinstance(home_page_content, oa_types.Unset):
if not polarion_content:
return None

home_page_content_type = None
home_page_content_value = None

if isinstance(
home_page_content.type,
api_models.DocumentsSingleGetResponseDataAttributesHomePageContentType,
):
home_page_content_type = str(home_page_content.type)
if isinstance(home_page_content.value, str):
home_page_content_value = home_page_content.value
return dm.TextContent(
type=home_page_content_type,
value=home_page_content_value,
type=str(polarion_content.type) if polarion_content.type else None,
value=polarion_content.value or None,
)

def create_work_items(self, work_items: list[base_client.WorkItemType]):
Expand Down Expand Up @@ -1015,3 +1015,133 @@ def _delete_work_item_links(
if not self._check_response(response, not retry) and retry:
sleep_random_time()
self._delete_work_item_links(work_item_links, False)

def get_test_records(
self,
test_run_id: str,
fields: dict[str, str] | None = None,
page_size: int = 100,
page_number: int = 1,
retry: bool = True,
) -> tuple[list[dm.TestRecord], bool]:
"""Return the test records on a defined page matching the given query.
In addition, a flag whether a next page is available is
returned. Define a fields dictionary as described in the
Polarion API documentation to get certain fields.
"""
if fields is None:
fields = self.default_fields.testrecords

sparse_fields = _build_sparse_fields(fields)
response = get_test_records.sync_detailed(
self.project_id,
test_run_id,
client=self.client,
fields=sparse_fields,
pagenumber=page_number,
pagesize=page_size,
)

if not self._check_response(response, not retry) and retry:
sleep_random_time()
return self.get_test_records(
test_run_id, fields, page_size, page_number, False
)

parsed_response = response.parsed
assert parsed_response

test_records = []
for data in parsed_response.data or []:
assert isinstance(data.id, str)
assert isinstance(
data.attributes,
api_models.TestrecordsListGetResponseDataItemAttributes,
)
_, _, project_id, work_item, iteration = data.id.split("/")
test_records.append(
dm.TestRecord(
project_id,
work_item,
data.attributes.test_case_revision or None,
int(iteration),
data.attributes.duration or -1,
data.attributes.result or None,
self._handle_text_content(data.attributes.comment),
data.additional_properties or {},
)
)
next_page = isinstance(
parsed_response.links,
api_models.TestrecordsListGetResponseLinks,
) and bool(parsed_response.links.next_)

return test_records, next_page

def get_test_runs(
self,
query: str,
fields: dict[str, str] | None = None,
page_size: int = 100,
page_number: int = 1,
retry: bool = True,
) -> tuple[list[dm.TestRun], bool]:
"""Return the test runs on a defined page matching the given query.
In addition, a flag whether a next page is available is
returned. Define a fields dictionary as described in the
Polarion API documentation to get certain fields.
"""
if fields is None:
fields = self.default_fields.testruns

sparse_fields = _build_sparse_fields(fields)
response = get_test_runs.sync_detailed(
self.project_id,
client=self.client,
query=query,
fields=sparse_fields,
pagenumber=page_number,
pagesize=page_size,
)

if not self._check_response(response, not retry) and retry:
sleep_random_time()
return self.get_test_runs(
query, fields, page_size, page_number, False
)

parsed_response = response.parsed
assert parsed_response

test_runs = []
for data in parsed_response.data or []:
assert isinstance(data.id, str)
assert isinstance(
data.attributes,
api_models.TestrunsListGetResponseDataItemAttributes,
)
test_runs.append(
dm.TestRun(
data.id.split("/")[-1],
data.attributes.type or None,
data.attributes.status or None,
data.attributes.title or None,
self._handle_text_content(
data.attributes.home_page_content
),
dm.SelectTestCasesBy(
str(data.attributes.select_test_cases_by)
)
if data.attributes.select_test_cases_by
else None,
data.attributes.additional_properties or {},
)
)
next_page = isinstance(
parsed_response.links,
api_models.TestrunsListGetResponseLinks,
) and bool(parsed_response.links.next_)

return test_runs, next_page
11 changes: 8 additions & 3 deletions polarion_rest_api_client/data_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -229,25 +229,30 @@ def __init__(
self.home_page_content = home_page_content


@dataclasses.dataclass
class TestRun(BaseItem):
"""A data class for all data of a test run."""

title: str | None = None
home_page_content: TextContent | None = None
select_test_cases_by: SelectTestCasesBy | None = None
additional_attributes: dict[str, t.Any] = dataclasses.field(
default_factory=dict
)


@dataclasses.dataclass
class TestRecord:
"""A data class for test record data."""

work_item_project_id: str
work_item_id: str
work_item_revision: str | None = None
iteration: int = 0
duration: int = 0
duration: float = 0
result: str | None = None
test_case_revision: str | None = None
comment: TextContent | None = None
additional_properties: dict[str, t.Any] = dataclasses.field(
additional_attributes: dict[str, t.Any] = dataclasses.field(
default_factory=dict
)

Expand Down

0 comments on commit 4147428

Please sign in to comment.