Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Rewrite reportportal test results reporting #3223

Merged
merged 3 commits into from
Oct 1, 2024
Merged

Rewrite reportportal test results reporting #3223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b626b7d
Provide mapping between results and tests for easier reporting
happz Sep 18, 2024
f59bc68
Test results reporting rewrite
kkaarreell Sep 18, 2024
2a030dd
Add a short release note
psss Oct 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/releases.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,10 @@ The :ref:`/plugins/provision/beaker` provision plugin now newly
supports providing a custom :ref:`/spec/plans/provision/kickstart`
configuration.

The :ref:`/plugins/report/reportportal` plugin now uploads the
complete set of discovered tests, including those which have not
been executed. These tests are marked as ``skipped``.


tmt-1.36.1
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Expand Down
28 changes: 28 additions & 0 deletions tmt/steps/execute/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1169,3 +1169,31 @@ def results(self) -> list["tmt.result.Result"]:
https://tmt.readthedocs.io/en/latest/spec/plans.html#execute
"""
return self._results

def results_for_tests(
self,
tests: list['tmt.base.Test']
) -> list[tuple[Optional[Result], Optional['tmt.base.Test']]]:
"""
Collect results and corresponding tests.

:returns: a list of result and test pairs.
* if there is not test found for the result, e.g. when
results were loaded from storage but tests were not,
``None`` represents the missing test: ``(result, None)``.
* if there is no result for a test, e.g. when the test was
not executed, ``None`` represents the missing result:
``(None, test)``.
"""

known_serial_numbers = {test.serial_number: test for test in tests}
referenced_serial_numbers = {result.serial_number for result in self._results}

return [
(result, known_serial_numbers.get(result.serial_number))
for result in self._results
] + [
(None, test)
for test in tests
if test.serial_number not in referenced_serial_numbers
]
199 changes: 99 additions & 100 deletions tmt/steps/report/reportportal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@

import tmt.log
import tmt.steps.report
from tmt.result import Result, ResultOutcome
from tmt.result import ResultOutcome
from tmt.utils import field, yaml_to_dict

if TYPE_CHECKING:
Expand Down Expand Up @@ -511,125 +511,124 @@ def go(self, *, logger: Optional[tmt.log.Logger] = None) -> None:
self.verbose("uuid", suite_uuid, "yellow", shift=1)
self.data.suite_uuid = suite_uuid

# For each test
for test in self.step.plan.discover.tests():

for result, test in self.step.plan.execute.results_for_tests(
self.step.plan.discover.tests()):
test_time = self.time()

results: list[Optional[Result]] = [None]
if executed:
results = [result for result in self.step.plan.execute.results()
if test.serial_number == result.serial_number]

# process even tests not having test results
for result in results:

# TODO: for happz, connect Test to Result if possible
# (but now it is probably too hackish to be fixed)

item_attributes = attributes.copy()
if result:
test_time = result.start_time or self.time()

# for guests, save their primary address
if result.guest.primary_address:
item_attributes.append({
'key': 'guest_primary_address',
'value': result.guest.primary_address})

# for multi-host tests store also guest name and role
if result.guest.name != 'default-0':
item_attributes.append(
{'key': 'guest_name', 'value': result.guest.name})
if result.guest.role:
item_attributes.append(
{'key': 'guest_role', 'value': result.guest.role})

test_name = None
test_description = ''
test_link = None
test_id = None
env_vars = None

item_attributes = attributes.copy()
if result:
serial_number = result.serial_number
test_name = result.name
test_time = result.start_time or self.time()
# for guests, save their primary address
if result.guest.primary_address:
item_attributes.append({
'key': 'guest_primary_address',
'value': result.guest.primary_address})
# for multi-host tests store also provision name and role
if result.guest.name != 'default-0':
item_attributes.append(
{'key': 'guest_name', 'value': result.guest.name})
if result.guest.role:
item_attributes.append(
{'key': 'guest_role', 'value': result.guest.role})

# update RP item with additional attributes if test details are available
if test:
serial_number = test.serial_number
if not test_name:
test_name = test.name
if test.contact:
item_attributes.append({"key": "contact", "value": test.contact[0]})
if test.summary:
test_description = test.summary
if test.web_link():
test_link = test.web_link()
if test.id:
test_id = test.id
env_vars = [
{'key': key, 'value': value}
for key, value in test.environment.items()
if not re.search(envar_pattern, key)]

if create_test:

test_description = test.summary or ""
if ((self.data.upload_to_launch and launch_per_plan)
or self.data.upload_to_suite):
test_description = self.append_description(test_description)
if create_test:
if ((self.data.upload_to_launch and launch_per_plan)
or self.data.upload_to_suite):
test_description = self.append_description(test_description)

# Create a test item
self.info("test", test.name, color="cyan")
# Create a test item
self.info("test", test_name, color="cyan")
response = self.rp_api_post(
session=session,
path=f"item{f'/{suite_uuid}' if suite_uuid else ''}",
json={"name": test_name,
"description": test_description,
"attributes": item_attributes,
"parameters": env_vars,
"codeRef": test_link,
"launchUuid": launch_uuid,
"type": "step",
"testCaseId": test_id,
"startTime": test_time})
item_uuid = yaml_to_dict(response.text).get("id")
assert item_uuid is not None
self.verbose("uuid", item_uuid, "yellow", shift=1)
self.data.test_uuids[serial_number] = item_uuid
else:
item_uuid = self.data.test_uuids[serial_number]
# Support for idle tests
status = "SKIPPED"
if result:
# For each log
for index, log_path in enumerate(result.log):
try:
log = self.step.plan.execute.read(log_path)
except tmt.utils.FileError:
continue

level = "INFO" if log_path == result.log[0] else "TRACE"
status = self.TMT_TO_RP_RESULT_STATUS[result.result]

# Upload log
response = self.rp_api_post(
session=session,
path=f"item{f'/{suite_uuid}' if suite_uuid else ''}",
json={"name": test.name,
"description": test_description,
"attributes": item_attributes,
"parameters": env_vars,
"codeRef": test.web_link() or None,
path="log/entry",
json={"message": _filter_invalid_chars(log),
"itemUuid": item_uuid,
"launchUuid": launch_uuid,
"type": "step",
"testCaseId": test.id or None,
"startTime": test_time})
item_uuid = yaml_to_dict(response.text).get("id")
assert item_uuid is not None
self.verbose("uuid", item_uuid, "yellow", shift=1)
self.data.test_uuids[test.serial_number] = item_uuid
else:
item_uuid = self.data.test_uuids[test.serial_number]

# Support for idle tests
status = "SKIPPED"
if executed and result:
# For each log
for index, log_path in enumerate(result.log):
try:
log = self.step.plan.execute.read(log_path)
except tmt.utils.FileError:
continue

level = "INFO" if log_path == result.log[0] else "TRACE"
status = self.TMT_TO_RP_RESULT_STATUS[result.result]

# Upload log
"level": level,
"time": result.end_time})

# Write out failures
if index == 0 and status == "FAILED":
message = _filter_invalid_chars(result.failures(log))
response = self.rp_api_post(
session=session,
path="log/entry",
json={"message": _filter_invalid_chars(log),
json={"message": message,
"itemUuid": item_uuid,
"launchUuid": launch_uuid,
"level": level,
"level": "ERROR",
"time": result.end_time})

# Write out failures
if index == 0 and status == "FAILED":
message = _filter_invalid_chars(result.failures(log))
response = self.rp_api_post(
session=session,
path="log/entry",
json={"message": message,
"itemUuid": item_uuid,
"launchUuid": launch_uuid,
"level": "ERROR",
"time": result.end_time})

# TODO: Add tmt files as attachments
test_time = result.end_time or self.time()

test_time = result.end_time or self.time()

# Finish the test item
response = self.rp_api_put(
session=session,
path=f"item/{item_uuid}",
json={
"launchUuid": launch_uuid,
"endTime": test_time,
"status": status,
"issue": {
"issueType": self.get_defect_type_locator(session, defect_type)}})
launch_time = test_time
# Finish the test item
response = self.rp_api_put(
session=session,
path=f"item/{item_uuid}",
json={
"launchUuid": launch_uuid,
"endTime": test_time,
"status": status,
"issue": {
"issueType": self.get_defect_type_locator(session, defect_type)}})
launch_time = test_time

if create_suite:
# Finish the test suite
Expand Down
Loading