1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
#
# bugtask.py - Internal helper class for sponsor-patch
#
# Copyright (C) 2010-2011, Benjamin Drung <bdrung@ubuntu.com>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
import logging
import os
import re
from urllib.parse import unquote
from urllib.request import urlretrieve
import distro_info
import httplib2
from ubuntutools.version import Version
Logger = logging.getLogger(__name__)
def is_sync(bug):
"""Checks if a Launchpad bug is a sync request.
Either the title contains the word sync or the bug contains the tag sync."""
return "sync" in bug.title.lower().split(" ") or "sync" in bug.tags
class BugTask:
def __init__(self, bug_task, launchpad):
self.bug_task = bug_task
self.launchpad = launchpad
components = re.split(r" \(| ", self.bug_task.bug_target_name.strip(")"))
assert len(components) >= 1 and len(components) <= 3
if len(components) == 1:
self.package = None
self.project = components[0]
self.series = None
elif len(components) == 2:
self.package = components[0]
self.project = components[1].lower()
self.series = None
elif len(components) == 3:
self.package = components[0]
self.project = components[1].lower()
self.series = components[2].lower()
if self.package is None:
title_re = r"^Sync ([a-z0-9+.-]+) [a-z0-9.+:~-]+ \([a-z]+\) from.*"
match = re.match(title_re, self.get_bug_title(), re.U | re.I)
if match is not None:
self.package = match.group(1)
def download_source(self):
source_files = self.get_source().sourceFileUrls()
dsc_file = ""
for url in source_files:
filename = unquote(os.path.basename(url))
Logger.debug("Downloading %s...", filename)
# HttpLib2 isn't suitable for large files (it reads into memory),
# but we want its https certificate validation on the .dsc
if url.endswith(".dsc"):
response, data = httplib2.Http().request(url)
assert response.status == 200
with open(filename, "wb") as f:
f.write(data)
dsc_file = os.path.join(os.getcwd(), filename)
else:
urlretrieve(url, filename)
assert os.path.isfile(dsc_file), f"{dsc_file} does not exist."
return dsc_file
def get_branch_link(self):
return "lp:" + self.project + "/" + self.get_series() + "/" + self.package
def get_bug_title(self):
"""Returns the title of the related bug."""
return self.bug_task.bug.title
def get_long_info(self):
return (
"Bug task: "
+ str(self.bug_task)
+ "\n"
+ "Package: "
+ str(self.package)
+ "\n"
+ "Project: "
+ str(self.project)
+ "\n"
+ "Series: "
+ str(self.series)
)
def get_lp_task(self):
"""Returns the Launchpad bug task object."""
return self.bug_task
def get_package_and_series(self):
result = self.package
if self.series:
result += " (" + self.series + ")"
return result
def get_previous_version(self):
if self.is_derived_from_debian():
previous_version = self.get_latest_released_version()
else:
previous_version = self.get_version()
return previous_version
def get_series(self, latest_release=False):
if self.series is None or latest_release:
dist = self.launchpad.distributions[self.project]
return dist.current_series.name
return self.series
def get_short_info(self):
return self.bug_task.bug_target_name + ": " + self.bug_task.status
def get_source(self, latest_release=False):
assert self.package is not None
if self.is_derived_from_debian() and not latest_release:
project = "debian"
series = self.get_debian_source_series()
else:
project = self.project
series = self.get_series(latest_release)
dist = self.launchpad.distributions[project]
archive = dist.getArchive(name="primary")
distro_series = dist.getSeries(name_or_version=series)
published = archive.getPublishedSources(
source_name=self.package,
distro_series=distro_series,
status="Published",
exact_match=True,
)
latest_source = None
for source in published:
if source.pocket in ("Release", "Security", "Updates", "Proposed"):
latest_source = source
break
return latest_source
def get_version(self):
source_package_version = self.get_source().source_package_version
return Version(source_package_version)
def get_latest_released_version(self):
source = self.get_source(True)
if source is None: # Not currently published in Ubuntu
version = "~"
else:
version = source.source_package_version
return Version(version)
def get_debian_source_series(self):
title = self.bug_task.bug.title.lower().split()
if "experimental" in title:
series = "experimental"
elif "testing" in title:
series = distro_info.DebianDistroInfo().testing()
else:
series = distro_info.DebianDistroInfo().devel()
return series
def is_complete(self):
return self.bug_task.is_complete
def is_derived_from_debian(self):
"""Checks if this task get's the source from Debian."""
return self.is_merge() or self.is_sync()
def is_merge(self):
bug = self.bug_task.bug
return "merge" in bug.title.lower().split(" ") or "merge" in bug.tags
def is_sync(self):
return is_sync(self.bug_task.bug)
def is_ubuntu_task(self):
return self.project == "ubuntu"
def title_contains(self, word):
"""Checks if the bug title contains the given word."""
return word in self.bug_task.bug.title.split(" ")
|
