tcms.rpc.api.testexecution.update() - Code Metrics - kiwitcms/Kiwi - Measure and Improve Code Quality continuously with Scrutinizer

tcms.rpc.api.testexecution.update() F
last analyzed 2022-04-14 09:41 UTC

↳ Parent: tcms.rpc.api.testexecution

Complexity

Conditions

Size

Total Lines	71
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	38
dl	0
loc	71
rs	2.9998
c	0
b	0
f	0
cc	15
nop	3

How to fix Long Method Complexity

# -*- coding: utf-8 -*-

from django.conf import settings
from django.forms.models import model_to_dict
from django.utils import timezone
from modernrpc.core import REQUEST_KEY, rpc_method

from tcms.core.contrib.linkreference.models import LinkReference
from tcms.core.helpers import comments
from tcms.core.utils import form_errors_to_list
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


__all__ = (
    "update",
    "filter",
    "history",
    "add_comment",
    "remove_comment",
    "add_link",
    "get_links",
    "remove_link",
    "properties",
)


@permissions_required("django_comments.add_comment")
@rpc_method(name="TestExecution.add_comment")
def add_comment(execution_id, comment, **kwargs):
    """
    .. function:: RPC TestExecution.add_comment(execution_id, comment)

        Add comment to selected test execution.

        :param execution_id: PK of a TestExecution object
        :type execution_id: int
        :param comment: The text to add as a comment
        :type comment: str
        :param \\**kwargs: Dict providing access to the current request, protocol,
                entry point name and handler instance from the rpc method
        :return: Serialized :class:`django_comments.models.Comment` object
        :rtype: dict
        :raises PermissionDenied: if missing *django_comments.add_comment* permission
    """
    execution = TestExecution.objects.get(pk=execution_id)
    created = comments.add_comment([execution], comment, kwargs.get(REQUEST_KEY).user)
    # we always create only one comment
    return model_to_dict(created[0])


@permissions_required("django_comments.delete_comment")
@rpc_method(name="TestExecution.remove_comment")
def remove_comment(execution_id, comment_id=None):
    """
    .. function:: RPC TestExecution.remove_comment(execution_id, comment_id)

        Remove all or specified comment(s) from selected test execution.

        :param execution_id: PK of a TestExecution object
        :type execution_id: int
        :param comment_id: PK of a Comment object or None
        :type comment_id: int
        :raises PermissionDenied: if missing *django_comments.delete_comment* permission
    """
    execution = TestExecution.objects.get(pk=execution_id)
    to_be_deleted = comments.get_comments(execution)
    if comment_id:
        to_be_deleted = to_be_deleted.filter(pk=comment_id)

    to_be_deleted.delete()


@permissions_required("django_comments.view_comment")
@rpc_method(name="TestExecution.get_comments")
def get_comments(execution_id):
    """
    .. function:: RPC TestExecution.get_comments(execution_id)

        Get all comments for selected test execution.

        :param execution_id: PK of a TestExecution object
        :type execution_id: int
        :return: Serialized :class:`django_comments.models.Comment` object
        :rtype: dict
        :raises PermissionDenied: if missing *django_comments.view_comment* permission
    """
    execution = TestExecution.objects.get(pk=execution_id)
    execution_comments = comments.get_comments(execution).values()
    return list(execution_comments)


@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.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",
        )
        .distinct()
    )


@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)


@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(form_errors_to_list(form))

    # if this call updated TE.status then adjust timestamps
    if values.get("status"):
        now = timezone.now()
        if test_execution.status.weight != 0:
            test_execution.stop_date = now
        else:
            test_execution.stop_date = None
        test_execution.save()

        all_executions = TestExecution.objects.filter(run=test_execution.run)
        if (
            test_execution.status.weight != 0
            and not all_executions.filter(status__weight=0).exists()
        ):
            test_execution.run.stop_date = now
            test_execution.run.save()
        elif test_execution.status.weight == 0 and test_execution.run.stop_date:
            test_execution.run.stop_date = None
            test_execution.run.save()

    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


@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(form_errors_to_list(form))

    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)


@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()


@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",
        )
    )


@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",
        )
    )


1			# -- coding: utf-8 --
2
3			from django.conf import settings
4			from django.forms.models import model_to_dict
5			from django.utils import timezone
6			from modernrpc.core import REQUEST_KEY, rpc_method
7
8			from tcms.core.contrib.linkreference.models import LinkReference
9			from tcms.core.helpers import comments
10			from tcms.core.utils import form_errors_to_list
11			from tcms.rpc.api.forms.testexecution import LinkReferenceForm
12			from tcms.rpc.api.forms.testrun import UpdateExecutionForm
13			from tcms.rpc.api.utils import tracker_from_url
14			from tcms.rpc.decorators import permissions_required
15			from tcms.testruns.models import TestExecution, TestExecutionProperty
16
17			# conditional import b/c this App can be disabled
18			if "tcms.bugs.apps.AppConfig" in settings.INSTALLED_APPS:
19			from tcms.issuetracker.kiwitcms import KiwiTCMS
20			else:
21
22			class KiwiTCMS: # pylint: disable=remove-empty-class,nested-class-found,too-few-public-methods
23			pass
24
25
26			__all__ = (
27			"update",
28			"filter",
29			"history",
30			"add_comment",
31			"remove_comment",
32			"add_link",
33			"get_links",
34			"remove_link",
35			"properties",
36			)
37
38
39			@permissions_required("django_comments.add_comment")
40			@rpc_method(name="TestExecution.add_comment")
41			def add_comment(execution_id, comment, **kwargs):
42			"""
43			.. function:: RPC TestExecution.add_comment(execution_id, comment)
44
45			Add comment to selected test execution.
46
47			:param execution_id: PK of a TestExecution object
48			:type execution_id: int
49			:param comment: The text to add as a comment
50			:type comment: str
51			:param \\**kwargs: Dict providing access to the current request, protocol,
52			entry point name and handler instance from the rpc method
53			:return: Serialized :class:`django_comments.models.Comment` object
54			:rtype: dict
55			:raises PermissionDenied: if missing django_comments.add_comment permission
56			"""
57			execution = TestExecution.objects.get(pk=execution_id)
58			created = comments.add_comment([execution], comment, kwargs.get(REQUEST_KEY).user)
59			# we always create only one comment
60			return model_to_dict(created[0])
61
62
63			@permissions_required("django_comments.delete_comment")
64			@rpc_method(name="TestExecution.remove_comment")
65			def remove_comment(execution_id, comment_id=None):
66			"""
67			.. function:: RPC TestExecution.remove_comment(execution_id, comment_id)
68
69			Remove all or specified comment(s) from selected test execution.
70
71			:param execution_id: PK of a TestExecution object
72			:type execution_id: int
73			:param comment_id: PK of a Comment object or None
74			:type comment_id: int
75			:raises PermissionDenied: if missing django_comments.delete_comment permission
76			"""
77			execution = TestExecution.objects.get(pk=execution_id)
78			to_be_deleted = comments.get_comments(execution)
79			if comment_id:
80			to_be_deleted = to_be_deleted.filter(pk=comment_id)
81
82			to_be_deleted.delete()
83
84
85			@permissions_required("django_comments.view_comment")
86			@rpc_method(name="TestExecution.get_comments")
87			def get_comments(execution_id):
88			"""
89			.. function:: RPC TestExecution.get_comments(execution_id)
90
91			Get all comments for selected test execution.
92
93			:param execution_id: PK of a TestExecution object
94			:type execution_id: int
95			:return: Serialized :class:`django_comments.models.Comment` object
96			:rtype: dict
97			:raises PermissionDenied: if missing django_comments.view_comment permission
98			"""
99			execution = TestExecution.objects.get(pk=execution_id)
100			execution_comments = comments.get_comments(execution).values()
101			return list(execution_comments)
102
103
104			@permissions_required("testruns.view_testexecution")
105			@rpc_method(name="TestExecution.filter")
106			def filter(query): # pylint: disable=redefined-builtin
107			"""
108			.. function:: RPC TestExecution.filter(query)
109
110			Perform a search and return the resulting list of test case executions.
111
112			:param query: Field lookups for :class:`tcms.testruns.models.TestExecution`
113			:type query: dict
114			:return: List of serialized :class:`tcms.testruns.models.TestExecution` objects
115			:rtype: list(dict)
116			"""
117			return list(
118			TestExecution.objects.filter(**query)
119			.values(
120			"id",
121			"assignee",
122			"assignee__username",
123			"tested_by",
124			"tested_by__username",
125			"case_text_version",
126			"start_date",
127			"stop_date",
128			"sortkey",
129			"run",
130			"case",
131			"case__summary",
132			"build",
133			"build__name",
134			"status",
135			"status__name",
136			)
137			.distinct()
138			)
139
140
141			@permissions_required("testruns.view_testexecution")
142			@rpc_method(name="TestExecution.history")
143			def history(execution_id):
144			"""
145			.. function:: RPC TestExecution.history(execution_id)
146
147			Return the history for the selected test execution.
148
149			:param execution_id: PK of a TestExecution object
150			:type execution_id: int
151			:return: List of serialized :class:`tcms.core.history.KiwiHistoricalRecords` objects
152			:rtype: list(dict)
153			:raises PermissionDenied: if missing testruns.view_testexecution permission
154			"""
155			execution = TestExecution.objects.get(pk=execution_id)
156			execution_history = (
157			execution.history.all()
158			.order_by("-history_date")
159			.values(
160			"history_user__username",
161			"history_change_reason",
162			"history_date",
163			"history_change_reason",
164			)
165			)
166			return list(execution_history)
167
168
169			@permissions_required("testruns.change_testexecution")
170			@rpc_method(name="TestExecution.update")
171			def update(execution_id, values, **kwargs):
172			"""
173			.. function:: RPC TestExecution.update(execution_id, values)
174
175			Update the selected TestExecution
176
177			:param execution_id: PK of TestExecution to modify
178			:type execution_id: int
179			:param values: Field values for :class:`tcms.testruns.models.TestExecution`
180			:type values: dict
181			:param \\**kwargs: Dict providing access to the current request, protocol,
182			entry point name and handler instance from the rpc method
183			:return: Serialized :class:`tcms.testruns.models.TestExecution` object
184			:rtype: dict
185			:raises ValueError: if data validations fail
186			:raises PermissionDenied: if missing testruns.change_testexecution permission
187			"""
188			test_execution = TestExecution.objects.get(pk=execution_id)
189
190			if values.get("case_text_version") == "latest":
191			values["case_text_version"] = test_execution.case.history.latest().history_id
192
193			if values.get("status") and not values.get("tested_by"):
194			values["tested_by"] = kwargs.get(REQUEST_KEY).user.id
195
196			if values.get("status") and not values.get("build"):
197			values["build"] = test_execution.run.build.pk
198
199			form = UpdateExecutionForm(values, instance=test_execution)
200
201			if form.is_valid():
202			test_execution = form.save()
203			else:
204			raise ValueError(form_errors_to_list(form))
205
206			# if this call updated TE.status then adjust timestamps
207			if values.get("status"):
208			now = timezone.now()
209			if test_execution.status.weight != 0:
210			test_execution.stop_date = now
211			else:
212			test_execution.stop_date = None
213			test_execution.save()
214
215			all_executions = TestExecution.objects.filter(run=test_execution.run)
216			if (
217			test_execution.status.weight != 0
218			and not all_executions.filter(status__weight=0).exists()
219			):
220			test_execution.run.stop_date = now
221			test_execution.run.save()
222			elif test_execution.status.weight == 0 and test_execution.run.stop_date:
223			test_execution.run.stop_date = None
224			test_execution.run.save()
225
226			result = model_to_dict(test_execution)
227
228			# augment result with additional information
229			result["assignee__username"] = (
230			test_execution.assignee.username if test_execution.assignee else None
231			)
232			result["tested_by__username"] = (
233			test_execution.tested_by.username if test_execution.tested_by else None
234			)
235			result["case__summary"] = test_execution.case.summary
236			result["build__name"] = test_execution.build.name
237			result["status__name"] = test_execution.status.name
238
239			return result
240
241
242			@permissions_required("linkreference.add_linkreference")
243			@rpc_method(name="TestExecution.add_link")
244			def add_link(values, update_tracker=False, **kwargs):
245			"""
246			.. function:: RPC TestExecution.add_link(values)
247
248			Add new URL link to a TestExecution
249
250			:param values: Field values for
251			:class:`tcms.core.contrib.linkreference.models.LinkReference`
252			:type values: dict
253			:param update_tracker: Automatically update Issue Tracker by placing a comment
254			linking back to the failed TestExecution.
255			:type update_tracker: bool, default=False
256			:param \\**kwargs: Dict providing access to the current request, protocol,
257			entry point name and handler instance from the rpc method
258			:return: Serialized
259			:class:`tcms.core.contrib.linkreference.models.LinkReference` object
260			:rtype: dict
261			:raises RuntimeError: if operation not successfull
262			:raises ValueError: if input validation fails
263
264			.. note::
265
266			Always 'link' with IT instance if URL is from Kiwi TCMS own bug tracker!
267			"""
268			# for backwards compatibility
269			if "execution_id" in values:
270			values["execution"] = values["execution_id"]
271			del values["execution_id"]
272
273			form = LinkReferenceForm(values)
274
275			if form.is_valid():
276			link = form.save()
277			else:
278			raise ValueError(form_errors_to_list(form))
279
280			request = kwargs.get(REQUEST_KEY)
281			tracker = tracker_from_url(link.url, request)
282
283			if (
284			link.is_defect
285			and tracker is not None
286			and update_tracker
287			and not tracker.is_adding_testcase_to_issue_disabled()
288			) or isinstance(tracker, KiwiTCMS):
289			tracker.add_testexecution_to_issue([link.execution], link.url)
290
291			return model_to_dict(link)
292
293
294			@permissions_required("linkreference.delete_linkreference")
295			@rpc_method(name="TestExecution.remove_link")
296			def remove_link(query):
297			"""
298			.. function:: RPC TestExecution.remove_link(query)
299
300			Remove URL link from TestExecution
301
302			:param query: Field lookups for
303			:class:`tcms.core.contrib.linkreference.models.LinkReference`
304			:type query: dict
305			"""
306			LinkReference.objects.filter(**query).delete()
307
308
309			@permissions_required("linkreference.view_linkreference")
310			@rpc_method(name="TestExecution.get_links")
311			def get_links(query):
312			"""
313			.. function:: RPC TestExecution.get_links(query)
314
315			Get URL links for the specified TestExecution
316
317			:param query: Field lookups for
318			:class:`tcms.core.contrib.linkreference.models.LinkReference`
319			:type query: dict
320			:return: Serialized list of :class:`tcms.core.contrib.linkreference.models.LinkReference`
321			objects
322			:rtype: dict
323			"""
324			return list(
325			LinkReference.objects.filter(**query).values(
326			"id",
327			"name",
328			"url",
329			"execution",
330			"created_on",
331			"is_defect",
332			)
333			)
334
335
336			@permissions_required("testruns.view_testexecutionproperty")
337			@rpc_method(name="TestExecution.properties")
338			def properties(query):
339			"""
340			.. function:: RPC TestExecution.properties(query)
341
342			Return properties for a TestExecution
343
344			:param query: Field lookups for
345			:class:`tcms.testruns.models.TestExecutionProperty`
346			:type query: dict
347			:return: Serialized list of :class:`tcms.testruns.models.TestExecutionProperty`
348			objects
349			:rtype: dict
350			"""
351			return list(
352			TestExecutionProperty.objects.filter(**query).values(
353			"id",
354			"name",
355			"value",
356			"execution",
357			)
358			)
359

kiwitcms / Kiwi

tcms.rpc.api.testexecution.update() F last analyzed 2022-04-14 09:41 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

tcms.rpc.api.testexecution.update() F
last analyzed 2022-04-14 09:41 UTC