# # bugtask.py - Internal helper class for sponsor-patch # # Copyright (C) 2010-2011, Benjamin Drung # # 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(" ")