tcms.testruns.views   F
last analyzed

Complexity

Total Complexity 77

Size/Duplication

Total Lines 727
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 77
eloc 486
dl 0
loc 727
rs 2.24
c 0
b 0
f 0

13 Functions

Rating   Name   Duplication   Size   Complexity  
A search() 0 9 1
C new() 0 88 8
A open_run_get_comments_subtotal() 0 11 1
A _open_run_get_executions() 0 37 2
B get_caseruns_of_runs() 0 26 7
A open_run_get_users() 0 13 4
B clone() 0 41 5
B update_case_run_text() 0 33 5
B edit() 0 46 6
B get() 0 67 5
A change_status() 0 9 1
B cc() 0 28 7
B remove_execution() 0 31 5

5 Methods

Rating   Name   Duplication   Size   Complexity  
A UpdateAssigneeView.post() 0 20 4
A UpdateCaseRunStatusView.post() 0 12 2
B TestRunReportView.get_context_data() 0 75 6
B AddCasesToRunView.post() 0 50 7
A AddCasesToRunView.get() 0 36 1

How to fix   Complexity   

Complexity

Complex classes like tcms.testruns.views often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
# -*- coding: utf-8 -*-
2
3
from datetime import datetime
4
from http import HTTPStatus
5
6
from django.conf import settings
7
from django.contrib import messages
8
from django.contrib.auth.decorators import permission_required
9
from django.contrib.auth import get_user_model
10
from django.contrib.contenttypes.models import ContentType
11
from django.core.exceptions import ObjectDoesNotExist
12
from django.urls import reverse
13
from django.db.models import Count
14
from django.db.models import Q
15
from django.http import HttpResponseRedirect, Http404, JsonResponse
16
from django.shortcuts import get_object_or_404, render
17
from django.utils.decorators import method_decorator
18
from django.utils.translation import ugettext_lazy as _
19
from django.views.decorators.http import require_GET
20
from django.views.decorators.http import require_POST
21
from django.views.generic.base import TemplateView
22
from django.views.generic.base import View
23
24
from django_comments.models import Comment
25
26
from tcms.core.utils import clean_request
27
from tcms.management.models import Priority, Tag
28
from tcms.testcases.models import TestCasePlan, TestCaseStatus, BugSystem
29
from tcms.testcases.views import get_selected_testcases
30
from tcms.testplans.models import TestPlan
31
from tcms.testruns.data import get_run_bug_ids
32
from tcms.testruns.data import TestExecutionDataMixin
33
from tcms.testruns.forms import NewRunForm, SearchRunForm, BaseRunForm
34
from tcms.testruns.models import TestRun, TestExecution, TestExecutionStatus
35
from tcms.issuetracker.types import IssueTrackerType
36
37
38
User = get_user_model()  # pylint: disable=invalid-name
39
40
41
@require_POST
42
@permission_required('testruns.add_testrun')
43
def new(request):
44
    """Display the create test run page."""
45
46
    # If from_plan does not exist will redirect to plans for select a plan
47
    if not request.POST.get('from_plan'):
48
        messages.add_message(request,
49
                             messages.ERROR,
50
                             _('Creating a TestRun requires a TestPlan, select one'))
51
        return HttpResponseRedirect(reverse('plans-search'))
52
53
    plan_id = request.POST.get('from_plan')
54
    # case is required by a test run
55
    # NOTE: currently this is handled in JavaScript but in the TestRun creation
56
    # form cases can be deleted
57
    if not request.POST.get('case'):
58
        messages.add_message(request,
59
                             messages.ERROR,
60
                             _('Creating a TestRun requires at least one TestCase'))
61
        return HttpResponseRedirect(reverse('test_plan_url_short', args=[plan_id]))
62
63
    # Ready to write cases to test plan
64
    test_cases = get_selected_testcases(request)
65
    test_plan = TestPlan.objects.get(plan_id=plan_id)
66
67
    # note: ordered by case_id for test_show_create_new_run_page()
68
    tcs_values = test_cases.select_related('author',
69
                                           'case_status',
70
                                           'category',
71
                                           'priority').order_by('case_id')
72
73
    if request.POST.get('POSTING_TO_CREATE'):
74
        form = NewRunForm(request.POST)
75
        form.populate(product_id=test_plan.product_id)
76
77
        if form.is_valid():
78
            # Process the data in form.cleaned_data
79
            default_tester = form.cleaned_data['default_tester']
80
81
            test_run = TestRun.objects.create(
82
                product_version=test_plan.product_version,
83
                stop_date=None,
84
                summary=form.cleaned_data.get('summary'),
85
                notes=form.cleaned_data.get('notes'),
86
                plan=test_plan,
87
                build=form.cleaned_data['build'],
88
                manager=form.cleaned_data['manager'],
89
                default_tester=default_tester,
90
            )
91
92
            try:
93
                assignee_tester = User.objects.get(username=default_tester)
94
            except ObjectDoesNotExist:
95
                assignee_tester = None
96
97
            loop = 1
98
            for case in form.cleaned_data['case']:
99
                try:
100
                    tcp = TestCasePlan.objects.get(plan=test_plan, case=case)
101
                    sortkey = tcp.sortkey
102
                except ObjectDoesNotExist:
103
                    sortkey = loop * 10
104
105
                test_run.add_case_run(case=case,
106
                                      sortkey=sortkey,
107
                                      assignee=assignee_tester)
108
                loop += 1
109
110
            return HttpResponseRedirect(
111
                reverse('testruns-get', args=[test_run.run_id, ])
112
            )
113
114
    else:
115
        form = NewRunForm(initial={
116
            'summary': 'Test run for %s' % test_plan.name,
117
            'manager': test_plan.author.email,
118
            'default_tester': request.user.email,
119
            'notes': '',
120
        })
121
        form.populate(product_id=test_plan.product_id)
122
123
    context_data = {
124
        'test_plan': test_plan,
125
        'test_cases': tcs_values,
126
        'form': form,
127
    }
128
    return render(request, 'testruns/mutable.html', context_data)
129
130
131
@require_GET
132
def search(request):
133
    form = SearchRunForm(request.GET)
134
    form.populate(product_id=request.GET.get('product'))
135
136
    context_data = {
137
        'form': form,
138
    }
139
    return render(request, 'testruns/search.html', context_data)
140
141
142
def _open_run_get_executions(request, run):
143
    """Prepare for executions list in a TestRun page
144
145
    This is an internal method. Do not call this directly.
146
    """
147
148
    executions = run.case_run.select_related(
149
        'run', 'case'
150
    ).only('run__run_id',
151
           'run__plan',
152
           'status',
153
           'assignee',
154
           'tested_by',
155
           'case_text_version',
156
           'sortkey',
157
           'case__summary',
158
           'case__is_automated',
159
           'case__priority',
160
           'case__category__name'
161
           )
162
163
    # Get the bug count for each execution
164
    # 5. have to show the number of bugs of each execution
165
    executions = executions.annotate(num_bug=Count('case_run_bug', distinct=True))
166
167
    # todo: is this last distinct necessary
168
    executions = executions.distinct()
169
170
    # Continue to search the executionss with conditions
171
    # 4. executions preparing for render executions table
172
    executions = executions.filter(**clean_request(request))
173
    order_by = request.GET.get('order_by')
174
    if order_by:
175
        executions = executions.order_by(order_by)
176
    else:
177
        executions = executions.order_by('sortkey', 'pk')
178
    return executions
179
180
181
def open_run_get_comments_subtotal(case_run_ids):
182
    content_type = ContentType.objects.get_for_model(TestExecution)
183
    query_set = Comment.objects.filter(
184
        content_type=content_type,
185
        site_id=settings.SITE_ID,
186
        object_pk__in=case_run_ids,
187
        is_removed=False).values('object_pk').annotate(comment_count=Count('pk')).order_by(
188
            'object_pk')
189
190
    result = ((int(row['object_pk']), row['comment_count']) for row in query_set)
0 ignored issues
show
Comprehensibility Best Practice introduced by Anton Sankov
The variable row does not seem to be defined.
Loading history...
191
    return dict(result)
192
193
194
def open_run_get_users(case_runs):
195
    tester_ids = set()
196
    assignee_ids = set()
197
    for case_run in case_runs:
198
        if case_run.tested_by_id:
199
            tester_ids.add(case_run.tested_by_id)
200
        if case_run.assignee_id:
201
            assignee_ids.add(case_run.assignee_id)
202
    testers = User.objects.filter(
203
        pk__in=tester_ids).values_list('pk', 'username')
204
    assignees = User.objects.filter(
205
        pk__in=assignee_ids).values_list('pk', 'username')
206
    return (dict(testers.iterator()), dict(assignees.iterator()))
207
208
209
@require_GET
210
def get(request, run_id, template_name='run/get.html'):
211
    """Display testrun's detail"""
212
213
    # Get the test run
214
    try:
215
        # 1. get test run itself
216
        test_run = TestRun.objects.select_related().get(run_id=run_id)
217
    except ObjectDoesNotExist:
218
        raise Http404
219
220
    # Get the test executions that belong to the run
221
    # 2. get test run's all executions
222
    test_executions = _open_run_get_executions(request, test_run)
223
224
    status = TestExecutionStatus.objects.only('pk', 'name').order_by('pk')
225
226
    # Count the status
227
    # 3. calculate number of executions of each status
228
    status_stats_result = test_run.stats_executions_status(status)
229
230
    # Get the test execution bugs summary
231
    # 6. get the number of bugs of this run
232
    test_case_run_bugs_count = test_run.get_bug_count()
233
234
    # Get tag list of testcases
235
    # 7. get tags
236
    # Get the list of testcases belong to the run
237
    test_cases = []
238
    for test_execution in test_executions:
239
        test_cases.append(test_execution.case_id)
240
    tags = Tag.objects.filter(case__in=test_cases).values_list('name', flat=True)
241
    tags = list(set(tags))
242
    tags.sort()
243
244
    def walk_executions():
245
        """Walking executions for helping rendering executions table"""
246
247
        priorities = dict(Priority.objects.values_list('pk', 'value'))
248
        testers, assignees = open_run_get_users(test_executions)
249
        test_case_run_pks = []
250
        for test_case_run in test_executions:
251
            test_case_run_pks.append(test_case_run.pk)
252
        comments_subtotal = open_run_get_comments_subtotal(test_case_run_pks)
253
        status = TestExecutionStatus.get_names()
254
255
        for case_run in test_executions:
256
            yield (case_run,
257
                   testers.get(case_run.tested_by_id, None),
258
                   assignees.get(case_run.assignee_id, None),
259
                   priorities.get(case_run.case.priority_id),
260
                   status[case_run.status_id],
261
                   comments_subtotal.get(case_run.pk, 0))
262
263
    context_data = {
264
        'test_run': test_run,
265
        'from_plan': request.GET.get('from_plan', False),
266
        'test_case_runs': walk_executions(),
267
        'test_case_runs_count': len(test_executions),
268
        'status_stats': status_stats_result,
269
        'test_case_run_bugs_count': test_case_run_bugs_count,
270
        'test_status': status,
271
        'priorities': Priority.objects.filter(is_active=True),
272
        'case_own_tags': tags,
273
        'bug_trackers': BugSystem.objects.all(),
274
    }
275
    return render(request, template_name, context_data)
276
277
278
@permission_required('testruns.change_testrun')
279
def edit(request, run_id):
280
    """Edit test plan view"""
281
282
    try:
283
        test_run = TestRun.objects.select_related().get(run_id=run_id)
284
    except ObjectDoesNotExist:
285
        raise Http404
286
287
    # If the form is submitted
288
    if request.method == "POST":
289
        form = BaseRunForm(request.POST)
290
        form.populate(product_id=test_run.plan.product_id)
291
292
        # FIXME: Error handler
293
        if form.is_valid():
294
            test_run.summary = form.cleaned_data['summary']
295
            # Permission hack
296
            if test_run.manager == request.user or test_run.plan.author == request.user:
297
                test_run.manager = form.cleaned_data['manager']
298
            test_run.default_tester = form.cleaned_data['default_tester']
299
            test_run.build = form.cleaned_data['build']
300
            test_run.product_version = test_run.plan.product_version
301
            test_run.notes = form.cleaned_data['notes']
302
            test_run.save()
303
304
            return HttpResponseRedirect(reverse('testruns-get', args=[run_id, ]))
305
    else:
306
        # Generate a blank form
307
        form = BaseRunForm(initial={
308
            'summary': test_run.summary,
309
            'manager': test_run.manager.email,
310
            'default_tester': (test_run.default_tester and
311
                               test_run.default_tester.email or None),
312
            'version': test_run.product_version_id,
313
            'build': test_run.build_id,
314
            'notes': test_run.notes,
315
        })
316
        form.populate(test_run.plan.product_id)
317
318
    context_data = {
319
        'test_run': test_run,
320
        'test_plan': test_run.plan,
321
        'form': form,
322
    }
323
    return render(request, 'testruns/mutable.html', context_data)
324
325
326
class TestRunReportView(TemplateView, TestExecutionDataMixin):
327
    """Test Run report"""
328
329
    template_name = 'run/report.html'
330
331
    def get_context_data(self, **kwargs):
332
        """Generate report for specific TestRun
333
334
        There are four data source to generate this report.
335
        1. TestRun
336
        2. Test executions included in the TestRun
337
        3. Comments associated with each test execution
338
        4. Statistics
339
        5. bugs
340
        """
341
        run = TestRun.objects.select_related('manager', 'plan').get(pk=kwargs['run_id'])
342
343
        case_runs = TestExecution.objects.filter(
344
            run=run
345
        ).select_related(
346
            'status', 'case', 'tested_by'
347
        ).only(
348
            'close_date',
349
            'status__name',
350
            'case__category__name',
351
            'case__summary', 'case__is_automated',
352
            'tested_by__username'
353
        )
354
        mode_stats = self.stats_mode_executions(case_runs)
355
        summary_stats = self.get_summary_stats(case_runs)
356
357
        test_case_run_bugs = []
358
        bug_system_types = {}
359
        for _bug in get_run_bug_ids(run.pk):
360
            # format the bug URLs based on DB settings
361
            test_case_run_bugs.append((
362
                _bug['bug_id'],
363
                _bug['bug_system__url_reg_exp'] % _bug['bug_id'],
364
            ))
365
            # find out all unique bug tracking systems which were used to record
366
            # bugs in this particular test run. we use this data for reporting
367
            if _bug['bug_system'] not in bug_system_types:
368
                # store a tracker type object for producing the report URL
369
                tracker_class = IssueTrackerType.from_name(_bug['bug_system__tracker_type'])
370
                bug_system = BugSystem.objects.get(pk=_bug['bug_system'])
371
                tracker = tracker_class(bug_system)
372
                bug_system_types[_bug['bug_system']] = (tracker, [])
373
374
            # store the list of bugs as well
375
            bug_system_types[_bug['bug_system']][1].append(_bug['bug_id'])
376
377
        # list of URLs which opens all bugs reported to every different
378
        # issue tracker used in this test run
379
        report_urls = []
380
        for (issue_tracker, ids) in bug_system_types.values():
381
            report_url = issue_tracker.all_issues_link(ids)
382
            # if IT doesn't support this feature or report url is not configured
383
            # the above method will return None
384
            if report_url:
385
                report_urls.append((issue_tracker.tracker.name, report_url))
386
387
        case_run_bugs = self.get_execution_bugs(run.pk)
388
        comments = self.get_execution_comments(run.pk)
389
390
        for case_run in case_runs:
391
            case_run.bugs = case_run_bugs.get(case_run.pk, ())
392
            case_run.user_comments = comments.get(case_run.pk, [])
393
394
        context = super().get_context_data(**kwargs)
395
        context.update({
396
            'test_run': run,
397
            'test_case_runs': case_runs,
398
            'test_case_runs_count': len(case_runs),
399
            'test_case_run_bugs': test_case_run_bugs,
400
            'mode_stats': mode_stats,
401
            'summary_stats': summary_stats,
402
            'report_urls': report_urls,
403
        })
404
405
        return context
406
407
408
@require_POST
409
def clone(request, run_id):
410
    """Clone cases from filter caserun"""
411
412
    test_run = get_object_or_404(TestRun, run_id=run_id)
413
    confirmed_case_status = TestCaseStatus.get_confirmed()
414
    disabled_cases = 0
415
416
    if request.POST.get('case_run'):
417
        test_cases = []
418
        for test_case_run in test_run.case_run.filter(pk__in=request.POST.getlist('case_run')):
419
            if test_case_run.case.case_status == confirmed_case_status:
420
                test_cases.append(test_case_run.case)
421
            else:
422
                disabled_cases += 1
423
    else:
424
        test_cases = None
425
426
    if not test_cases:
427
        messages.add_message(request,
428
                             messages.ERROR,
429
                             _('At least one TestCase is required'))
430
        return HttpResponseRedirect(reverse('testruns-get', args=[run_id]))
431
432
    form = NewRunForm(initial={
433
        'summary': _('Clone of ') + test_run.summary,
434
        'notes': test_run.notes,
435
        'manager': test_run.manager,
436
        'build': test_run.build_id,
437
        'default_tester': test_run.default_tester,
438
    })
439
    form.populate(product_id=test_run.plan.product_id)
440
441
    context_data = {
442
        'is_cloning': True,
443
        'disabled_cases': disabled_cases,
444
        'test_plan': test_run.plan,
445
        'test_cases': test_cases,
446
        'form': form,
447
    }
448
    return render(request, 'testruns/mutable.html', context_data)
449
450
451
@permission_required('testruns.change_testrun')
452
def change_status(request, run_id):
453
    """Change test run finished or running"""
454
    test_run = get_object_or_404(TestRun, run_id=run_id)
455
456
    test_run.update_completion_status(request.GET.get('finished') == '1')
457
    test_run.save()
458
459
    return HttpResponseRedirect(reverse('testruns-get', args=[run_id, ]))
460
461
462
@require_POST
463
@permission_required('testruns.delete_testexecution')
464
def remove_execution(request, run_id):
465
    """Remove specific execution from the run"""
466
467
    # Ignore invalid execution ids
468
    execution_ids = []
469
    for item in request.POST.getlist('case_run'):
470
        try:
471
            execution_ids.append(int(item))
472
        except (ValueError, TypeError):
473
            pass
474
475
    # If no execution to remove, no further operation is required, just return
476
    # back to run page immediately.
477
    if not execution_ids:
478
        return HttpResponseRedirect(reverse('testruns-get',
479
                                            args=[run_id, ]))
480
481
    run = get_object_or_404(TestRun.objects.only('pk'), pk=run_id)
482
483
    # Restrict to delete those executions that belongs to run
484
    TestExecution.objects.filter(run_id=run.pk, pk__in=execution_ids).delete()
485
486
    execution_exist = TestExecution.objects.filter(run_id=run.pk).exists()
487
    if execution_exist:
488
        redirect_to = 'testruns-get'
489
    else:
490
        redirect_to = 'add-cases-to-run'
491
492
    return HttpResponseRedirect(reverse(redirect_to, args=[run_id, ]))
493
494
495
@method_decorator(permission_required('testruns.add_testexecution'), name='dispatch')
496
class AddCasesToRunView(View):
497
    """Add cases to a TestRun"""
498
499
    def post(self, request, run_id):
500
        # Selected cases' ids to add to run
501
        test_cases_ids = request.POST.getlist('case')
502
        if not test_cases_ids:
503
            # user clicked Update button without selecting new Test Cases
504
            # to be dded to TestRun
505
            messages.add_message(request,
506
                                 messages.ERROR,
507
                                 _('At least one TestCase is required'))
508
            return HttpResponseRedirect(reverse('add-cases-to-run', args=[run_id]))
509
510
        try:
511
            test_cases_ids = list(map(int, test_cases_ids))
512
        except (ValueError, TypeError):
513
            # this will happen only on malicious requests
514
            messages.add_message(request,
515
                                 messages.ERROR,
516
                                 _('TestCase ID is not a valid integer'))
517
            return HttpResponseRedirect(reverse('add-cases-to-run', args=[run_id]))
518
519
        try:
520
            test_run = TestRun.objects.select_related('plan').only('plan__plan_id').get(
521
                run_id=run_id)
522
        except ObjectDoesNotExist:
523
            raise Http404
524
525
        test_case_runs_ids = test_run.case_run.values_list('case', flat=True)
526
527
        # avoid add cases that are already in current run with pk run_id
528
        test_cases_ids = set(test_cases_ids) - set(test_case_runs_ids)
529
530
        test_plan = test_run.plan
531
        test_cases = test_run.plan.case.filter(case_status__name='CONFIRMED').select_related(
532
            'default_tester').only('default_tester__id').filter(
533
                case_id__in=test_cases_ids)
534
535
        if request.POST.get('_use_plan_sortkey'):
536
            test_case_pks = (test_case.pk for test_case in test_cases)
0 ignored issues
show
Comprehensibility Best Practice introduced by Anton Sankov
The variable test_case does not seem to be defined.
Loading history...
537
            query_set = TestCasePlan.objects.filter(
538
                plan=test_plan, case__in=test_case_pks).values('case', 'sortkey')
539
            sort_keys_in_plan = dict((row['case'], row['sortkey']) for row in query_set.iterator())
540
            for test_case in test_cases:
541
                sort_key = sort_keys_in_plan.get(test_case.pk, 0)
542
                test_run.add_case_run(case=test_case, sortkey=sort_key)
543
        else:
544
            for test_case in test_cases:
545
                test_run.add_case_run(case=test_case)
546
547
        return HttpResponseRedirect(reverse('testruns-get',
548
                                            args=[test_run.run_id, ]))
549
550
    def get(self, request, run_id):
551
        # information about TestRun, used in the page header
552
        test_run = TestRun.objects.select_related(
553
            'plan', 'manager', 'build'
554
        ).only(
555
            'plan', 'plan__name',
556
            'manager__email', 'build__name'
557
        ).get(run_id=run_id)
558
559
        # select all CONFIRMED cases from the TestPlan that is a parent
560
        # of this particular TestRun
561
        rows = TestCasePlan.objects.values(
562
            'case',
563
            'case__create_date', 'case__summary',
564
            'case__category__name',
565
            'case__priority__value',
566
            'case__author__username'
567
        ).filter(
568
            plan_id=test_run.plan,
569
            case__case_status=TestCaseStatus.objects.filter(name='CONFIRMED').first().pk
570
        ).order_by('case')  # order b/c of PostgreSQL
571
572
        # also grab a list of all TestCase IDs which are already present in the
573
        # current TestRun so we can mark them as disabled and not allow them to
574
        # be selected
575
        test_case_runs = TestExecution.objects.filter(run=run_id).values_list('case', flat=True)
576
577
        data = {
578
            'test_run': test_run,
579
            'confirmed_cases': rows,
580
            'confirmed_cases_count': rows.count(),
581
            'test_case_runs_count': len(test_case_runs),
582
            'exist_case_run_ids': test_case_runs,
583
        }
584
585
        return render(request, 'run/assign_case.html', data)
586
587
588
@require_GET
589
def cc(request, run_id):  # pylint: disable=invalid-name
590
    """Add or remove cc from a test run"""
591
592
    test_run = get_object_or_404(TestRun, run_id=run_id)
593
    action = request.GET.get('do')
594
    username_or_email = request.GET.get('user')
595
    context_data = {'test_run': test_run, 'is_ajax': True}
596
597
    if action:
598
        if not username_or_email:
599
            context_data['message'] = 'User name or email is required by this operation'
600
        else:
601
            try:
602
                user = User.objects.get(
603
                    Q(username=username_or_email) |
604
                    Q(email=username_or_email)
605
                )
606
            except ObjectDoesNotExist:
607
                context_data['message'] = 'The user you typed does not exist in database'
608
            else:
609
                if action == 'add':
610
                    test_run.add_cc(user=user)
611
612
                if action == 'remove':
613
                    test_run.remove_cc(user=user)
614
615
    return render(request, 'run/get_cc.html', context_data)
616
617
618
@require_POST
619
@permission_required('testruns.change_testexecution')
620
def update_case_run_text(request, run_id):
621
    """Update the IDLE cases to newest text"""
622
623
    test_run = get_object_or_404(TestRun, run_id=run_id)
624
625
    if request.POST.get('case_run'):
626
        test_case_runs = test_run.case_run.filter(pk__in=request.POST.getlist('case_run'))
627
    else:
628
        test_case_runs = test_run.case_run.all()
629
630
    test_case_runs = test_case_runs.filter(status__name='IDLE')
631
632
    count = 0
633
    updated_test_case_runs = ''
634
    for test_case_run in test_case_runs:
635
        latest_version = test_case_run.case.history.latest().history_id
636
        if test_case_run.case_text_version != latest_version:
637
            count += 1
638
            updated_test_case_runs += '<li>%s: %s -> %s</li>' % (
639
                test_case_run.case.summary, test_case_run.case_text_version, latest_version
640
            )
641
            test_case_run.case_text_version = latest_version
642
            test_case_run.save()
643
644
    info = "<p>%s</p><ul>%s</ul>" % (_("%d CaseRun(s) updated:") % count, updated_test_case_runs)
645
    message_level = messages.INFO
646
    if count:
647
        message_level = messages.SUCCESS
648
649
    messages.add_message(request, message_level, info)
650
    return HttpResponseRedirect(reverse('testruns-get', args=[run_id]))
651
652
653
def get_caseruns_of_runs(runs, kwargs=None):
654
    """
655
    Filtering argument -
656
        priority
657
        tester
658
        plan tag
659
    """
660
661
    if kwargs is None:
662
        kwargs = {}
663
    plan_tag = kwargs.get('plan_tag', None)
664
    if plan_tag:
665
        runs = runs.filter(plan__tag__name=plan_tag)
666
    caseruns = TestExecution.objects.filter(run__in=runs)
667
    priority = kwargs.get('priority', None)
668
    if priority:
669
        caseruns = caseruns.filter(case__priority__pk=priority)
670
    tester = kwargs.get('tester', None)
671
    if not tester:
672
        caseruns = caseruns.filter(tested_by=None)
673
    if tester:
674
        caseruns = caseruns.filter(tested_by__pk=tester)
675
    status = kwargs.get('status', None)
676
    if status:
677
        caseruns = caseruns.filter(status__name__iexact=status)
678
    return caseruns
679
680
681
@method_decorator(permission_required('testruns.change_testexecution'), name='dispatch')
682
class UpdateAssigneeView(View):
683
    """Updates TestExecution.assignee. Called from the front-end."""
684
685
    http_method_names = ['post']
686
687
    def post(self, request):
688
        assignee = request.POST.get('assignee')
689
        try:
690
            user = User.objects.get(username=assignee)
691
        except User.DoesNotExist:
692
            try:
693
                user = User.objects.get(email=assignee)
694
            except User.DoesNotExist:
695
                return JsonResponse({'rc': 1,
696
                                     'response': _('User %s not found!' % assignee)},
697
                                    status=HTTPStatus.NOT_FOUND)
698
699
        object_ids = request.POST.getlist('ids[]')
700
701
        for caserun_pk in object_ids:
702
            test_case_run = get_object_or_404(TestExecution, pk=int(caserun_pk))
703
            test_case_run.assignee = user
704
            test_case_run.save()
705
706
        return JsonResponse({'rc': 0, 'response': 'ok'})
707
708
709
@method_decorator(permission_required('testruns.change_testexecution'), name='dispatch')
710
class UpdateCaseRunStatusView(View):
711
    """Updates TestExecution.status_id. Called from the front-end."""
712
713
    http_method_names = ['post']
714
715
    def post(self, request):
716
        status_id = int(request.POST.get('status_id'))
717
        object_ids = request.POST.getlist('object_pk[]')
718
719
        for caserun_pk in object_ids:
720
            test_case_run = get_object_or_404(TestExecution, pk=int(caserun_pk))
721
            test_case_run.status_id = status_id
722
            test_case_run.tested_by = request.user
723
            test_case_run.close_date = datetime.now()
724
            test_case_run.save()
725
726
        return JsonResponse({'rc': 0, 'response': 'ok'})
727