1# -*- coding: utf-8 -*-
3# Copyright 2013-2018 The Distro Tracker Developers
4# See the COPYRIGHT file at the top-level directory of this distribution and
5# at https://deb.li/DTAuthors
6#
7# This file is part of Distro Tracker. It is subject to the license terms
8# in the LICENSE file found in the top-level directory of this
9# distribution and at https://deb.li/DTLicense. No part of Distro Tracker,
10# including this file, may be copied, modified, propagated, or distributed
11# except according to the terms contained in the LICENSE file.
12"""
13Distro Tracker tasks for the :mod:`distro_tracker.stdver_warnings` app.
14"""
16from debian.debian_support import version_compare
18from django.db import transaction
19from django.db.models import Prefetch
21from distro_tracker.core.models import (
22 ActionItem,
23 ActionItemType,
24 SourcePackageName
25)
26from distro_tracker.core.tasks import BaseTask
27from distro_tracker.core.tasks.mixins import (
28 ProcessSrcRepoEntryInDefaultRepository
29)
30from distro_tracker.core.tasks.schedulers import IntervalScheduler
31from distro_tracker.core.utils import get_or_none
34class UpdateStandardsVersionWarnings(BaseTask,
35 ProcessSrcRepoEntryInDefaultRepository):
36 """
37 The task updates warnings for packages which have an outdated
38 Standards-Version.
39 """
41 class Scheduler(IntervalScheduler):
42 interval = 3600
44 ACTION_ITEM_TYPE = 'debian-std-ver-outdated'
45 FULL_DESCRIPTION_TEMPLATE = \
46 'stdver_warnings/standards-version-action-item.html'
47 ITEM_DESCRIPTION = "Standards version of the package is outdated."
49 def initialize(self, *args, **kwargs):
50 super().initialize(*args, **kwargs)
51 self.action_type = ActionItemType.objects.create_or_update(
52 type_name=self.ACTION_ITEM_TYPE,
53 full_description_template=self.FULL_DESCRIPTION_TEMPLATE)
55 def items_extend_queryset(self, queryset):
56 queryset = super().items_extend_queryset(queryset)
57 base_qs = ActionItem.objects.filter(item_type=self.action_type)
58 return queryset.prefetch_related(
59 Prefetch('source_package__source_package_name__action_items',
60 queryset=base_qs, to_attr='stdver_action_items')
61 )
63 def get_policy_version(self):
64 """
65 :returns: The latest version of the ``debian-policy`` package.
66 """
67 debian_policy = get_or_none(SourcePackageName, name='debian-policy')
68 if not debian_policy: 68 ↛ 69line 68 didn't jump to line 69, because the condition on line 68 was never true
69 return
70 policy_version = debian_policy.main_version.version
71 # Minor patch level should be disregarded for the comparison
72 if policy_version.count('.') == 3: 72 ↛ 75line 72 didn't jump to line 75, because the condition on line 72 was never false
73 policy_version, _ = policy_version.rsplit('.', 1)
75 return policy_version
77 def check_if_full_update_is_required(self, policy_version):
78 last_policy_version = self.data.get('policy_version')
79 if last_policy_version != policy_version: 79 ↛ exitline 79 didn't return from function 'check_if_full_update_is_required', because the condition on line 79 was never false
80 # Force a full update when a new policy version is released
81 self.force_update = True
82 self.data['policy_version'] = policy_version
83 self.data_mark_modified()
85 @transaction.atomic
86 def execute_main(self):
87 # Get the current policy version
88 policy_version = self.get_policy_version()
89 if policy_version is None: 89 ↛ 91line 89 didn't jump to line 91, because the condition on line 89 was never true
90 # Nothing to do if there is no ``debian-policy``
91 return
93 self.check_if_full_update_is_required(policy_version)
95 seen_packages = {}
96 for entry in self.items_to_process():
97 try:
98 package = entry.source_package.source_package_name
99 standards_version = entry.source_package.standards_version
100 try:
101 if package.name in seen_packages:
102 seen_version = seen_packages[package.name]
103 version = entry.source_package.version
104 if version_compare(version, seen_version) < 0: 104 ↛ 109line 104 didn't jump to line 109, because the condition on line 104 was never false
105 # This version is older, skip it
106 continue
107 # If already seen, then the cached action item
108 # is no longer reliable, retrieve it from the db
109 action_item = get_or_none(ActionItem, package=package,
110 item_type=self.action_type)
111 else:
112 action_item = package.stdver_action_items[0]
113 except IndexError:
114 action_item = None
115 seen_packages[package.name] = entry.source_package.version
117 if standards_version.startswith(policy_version):
118 # The std-ver of the package is up to date.
119 # Remove any possibly existing action item.
120 if action_item is not None:
121 action_item.delete()
122 continue
124 major_policy_version_number, _ = policy_version.split('.', 1)
125 severely_outdated = not standards_version.startswith(
126 major_policy_version_number)
128 if action_item is None:
129 action_item = ActionItem(
130 package=package,
131 item_type=self.action_type)
133 if severely_outdated:
134 action_item.severity = ActionItem.SEVERITY_HIGH
135 else:
136 action_item.severity = ActionItem.SEVERITY_WISHLIST
138 action_item.short_description = self.ITEM_DESCRIPTION
139 action_item.extra_data = {
140 'lastsv': policy_version,
141 'lastsv_dashes': policy_version.replace('.', '-'),
142 'standards_version': standards_version,
143 'standards_version_dashes':
144 standards_version.replace('.', '-'),
145 'severely_outdated': severely_outdated,
146 }
147 action_item.save()
148 finally:
149 self.item_mark_processed(entry)
151 # Remove action items for packages that disappeared from the default
152 # repository
153 ActionItem.objects.delete_obsolete_items(
154 [self.action_type],
155 self.items_all().values_list(
156 'source_package__source_package_name__name', flat=True)
157 )