# -*- coding: utf-8 -*-
from datetime import timedelta
from django.conf import settings
from django.db.models import F
from django.db.models.functions import Coalesce
from django.forms.models import model_to_dict
from modernrpc.core import REQUEST_KEY, rpc_method
from tcms.core.contrib.linkreference.models import LinkReference
from tcms.core.helpers import comments
from tcms.rpc.api.forms.testexecution import LinkReferenceForm
from tcms.rpc.api.forms.testrun import UpdateExecutionForm
from tcms.rpc.api.utils import tracker_from_url
from tcms.rpc.decorators import permissions_required
from tcms.testruns.models import TestExecution, TestExecutionProperty
# conditional import b/c this App can be disabled
if "tcms.bugs.apps.AppConfig" in settings.INSTALLED_APPS:
from tcms.issuetracker.kiwitcms import KiwiTCMS
else:
class KiwiTCMS: # pylint: disable=remove-empty-class,nested-class-found,too-few-public-methods
pass
[docs]
@permissions_required("testruns.view_testexecution")
@rpc_method(name="TestExecution.filter")
def filter(query): # pylint: disable=redefined-builtin
"""
.. function:: RPC TestExecution.filter(query)
Perform a search and return the resulting list of test case executions.
:param query: Field lookups for :class:`tcms.testruns.models.TestExecution`
:type query: dict
:return: List of serialized :class:`tcms.testruns.models.TestExecution` objects
:rtype: list(dict)
"""
return list(
TestExecution.objects.annotate(
expected_duration=(
Coalesce("case__setup_duration", timedelta(0))
+ Coalesce("case__testing_duration", timedelta(0))
),
actual_duration=F("stop_date") - F("start_date"),
)
.filter(**query)
.values(
"id",
"assignee",
"assignee__username",
"tested_by",
"tested_by__username",
"case_text_version",
"start_date",
"stop_date",
"sortkey",
"run",
"case",
"case__summary",
"build",
"build__name",
"status",
"status__name",
"status__icon",
"status__color",
"expected_duration",
"actual_duration",
)
.distinct()
)
[docs]
@permissions_required("testruns.view_testexecution")
@rpc_method(name="TestExecution.history")
def history(execution_id):
"""
.. function:: RPC TestExecution.history(execution_id)
Return the history for the selected test execution.
:param execution_id: PK of a TestExecution object
:type execution_id: int
:return: List of serialized :class:`tcms.core.history.KiwiHistoricalRecords` objects
:rtype: list(dict)
:raises PermissionDenied: if missing *testruns.view_testexecution* permission
"""
execution = TestExecution.objects.get(pk=execution_id)
execution_history = (
execution.history.all()
.order_by("-history_date")
.values(
"history_user__username",
"history_change_reason",
"history_date",
"history_change_reason",
)
)
return list(execution_history)
[docs]
@permissions_required("testruns.change_testexecution")
@rpc_method(name="TestExecution.update")
def update(execution_id, values, **kwargs):
"""
.. function:: RPC TestExecution.update(execution_id, values)
Update the selected TestExecution
:param execution_id: PK of TestExecution to modify
:type execution_id: int
:param values: Field values for :class:`tcms.testruns.models.TestExecution`
:type values: dict
:param \\**kwargs: Dict providing access to the current request, protocol,
entry point name and handler instance from the rpc method
:return: Serialized :class:`tcms.testruns.models.TestExecution` object
:rtype: dict
:raises ValueError: if data validations fail
:raises PermissionDenied: if missing *testruns.change_testexecution* permission
"""
test_execution = TestExecution.objects.get(pk=execution_id)
if values.get("case_text_version") == "latest":
values["case_text_version"] = test_execution.case.history.latest().history_id
if values.get("status") and not values.get("tested_by"):
values["tested_by"] = kwargs.get(REQUEST_KEY).user.id
if values.get("status") and not values.get("build"):
values["build"] = test_execution.run.build.pk
form = UpdateExecutionForm(values, instance=test_execution)
if form.is_valid():
test_execution = form.save()
else:
raise ValueError(list(form.errors.items()))
result = model_to_dict(test_execution)
# augment result with additional information
result["assignee__username"] = (
test_execution.assignee.username if test_execution.assignee else None
)
result["tested_by__username"] = (
test_execution.tested_by.username if test_execution.tested_by else None
)
result["case__summary"] = test_execution.case.summary
result["build__name"] = test_execution.build.name
result["status__name"] = test_execution.status.name
return result
[docs]
@permissions_required("linkreference.add_linkreference")
@rpc_method(name="TestExecution.add_link")
def add_link(values, update_tracker=False, **kwargs):
"""
.. function:: RPC TestExecution.add_link(values)
Add new URL link to a TestExecution
:param values: Field values for
:class:`tcms.core.contrib.linkreference.models.LinkReference`
:type values: dict
:param update_tracker: Automatically update Issue Tracker by placing a comment
linking back to the failed TestExecution.
:type update_tracker: bool, default=False
:param \\**kwargs: Dict providing access to the current request, protocol,
entry point name and handler instance from the rpc method
:return: Serialized
:class:`tcms.core.contrib.linkreference.models.LinkReference` object
:rtype: dict
:raises RuntimeError: if operation not successfull
:raises ValueError: if input validation fails
.. note::
Always 'link' with IT instance if URL is from Kiwi TCMS own bug tracker!
"""
# for backwards compatibility
if "execution_id" in values:
values["execution"] = values["execution_id"]
del values["execution_id"]
form = LinkReferenceForm(values)
if form.is_valid():
link = form.save()
else:
raise ValueError(list(form.errors.items()))
request = kwargs.get(REQUEST_KEY)
tracker = tracker_from_url(link.url, request)
if (
link.is_defect
and tracker is not None
and update_tracker
and not tracker.is_adding_testcase_to_issue_disabled()
) or isinstance(tracker, KiwiTCMS):
tracker.add_testexecution_to_issue([link.execution], link.url)
return model_to_dict(link)
[docs]
@permissions_required("linkreference.delete_linkreference")
@rpc_method(name="TestExecution.remove_link")
def remove_link(query):
"""
.. function:: RPC TestExecution.remove_link(query)
Remove URL link from TestExecution
:param query: Field lookups for
:class:`tcms.core.contrib.linkreference.models.LinkReference`
:type query: dict
"""
LinkReference.objects.filter(**query).delete()
[docs]
@permissions_required("linkreference.view_linkreference")
@rpc_method(name="TestExecution.get_links")
def get_links(query):
"""
.. function:: RPC TestExecution.get_links(query)
Get URL links for the specified TestExecution
:param query: Field lookups for
:class:`tcms.core.contrib.linkreference.models.LinkReference`
:type query: dict
:return: Serialized list of :class:`tcms.core.contrib.linkreference.models.LinkReference`
objects
:rtype: dict
"""
return list(
LinkReference.objects.filter(**query).values(
"id",
"name",
"url",
"execution",
"created_on",
"is_defect",
)
)
[docs]
@permissions_required("testruns.view_testexecutionproperty")
@rpc_method(name="TestExecution.properties")
def properties(query):
"""
.. function:: RPC TestExecution.properties(query)
Return properties for a TestExecution
:param query: Field lookups for
:class:`tcms.testruns.models.TestExecutionProperty`
:type query: dict
:return: Serialized list of :class:`tcms.testruns.models.TestExecutionProperty`
objects
:rtype: dict
"""
return list(
TestExecutionProperty.objects.filter(**query).values(
"id",
"name",
"value",
"execution",
)
)
[docs]
@permissions_required("testruns.delete_testexecution")
@rpc_method(name="TestExecution.remove")
def remove(query):
"""
.. function:: RPC TestExecution.remove(query)
Remove a TestExecution.
:param query: Field lookups for :class:`tcms.testruns.models.TestExecution`
:type query: dict
:raises PermissionDenied: if missing *testruns.delete_testexecution* permission
"""
TestExecution.objects.filter(**query).delete()