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
|
# Copyright 2012 Canonical Ltd.
# Author: Colin Watson <cjwatson@ubuntu.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Launchpad API utility functions."""
from operator import attrgetter
from debian import debian_support
from lazr.restfulclient.resource import Entry
class PackageMissing(Exception):
"Generic exception generated by `lputils`."
def __init__(self, message=None):
Exception.__init__(self, message)
self.message = message
known_pockets = (
"Security",
"Updates",
"Proposed",
"Backports",
)
ARCHIVE_REFERENCE_DESCRIPTION = (
'ARCHIVE can take one of four forms: "ubuntu" for a primary archive, '
'"~canonical-kernel-team/ubuntu/ppa" or '
'"ppa:canonical-kernel-team/ubuntu/ppa" for a PPA, or '
'"ubuntu/partner" for a partner or copy archive.')
def setup_location(options, default_pocket="Release"):
archive = None
if getattr(options, "archive", False):
# Try parsing an archive reference first.
archive = options.launchpad.archives.getByReference(
reference=options.archive)
if archive is None:
raise AssertionError("No such archive: %s" % options.archive)
else:
# Otherwise derive the archive from the deprecated
# -d/--ppa/--ppa-name/--partner options.
if isinstance(options.distribution, Entry):
distro = options.distribution
else:
distro = options.launchpad.distributions[options.distribution]
if getattr(options, "partner", False):
archive = [
archive for archive in distro.archives
if archive.name == "partner"][0]
elif getattr(options, "ppa", None):
archive = options.launchpad.people[options.ppa].getPPAByName(
distribution=distro, name=options.ppa_name)
else:
archive = distro.main_archive
options.archive = archive
options.distribution = archive.distribution
if options.suite:
if "-" in options.suite:
options.series, options.pocket = options.suite.rsplit("-", 1)
options.pocket = options.pocket.title()
if options.pocket not in known_pockets:
options.series = options.suite
options.pocket = "Release"
else:
options.series = options.suite
options.pocket = "Release"
options.series = options.distribution.getSeries(
name_or_version=options.series)
else:
options.series = options.distribution.current_series
options.pocket = default_pocket
if options.pocket == "Release":
options.suite = options.series.name
else:
options.suite = "%s-%s" % (
options.series.name, options.pocket.lower())
if getattr(options, "architecture", None) is not None:
options.architectures = [options.series.getDistroArchSeries(
archtag=options.architecture)]
elif getattr(options, "architectures", None) is not None:
options.architectures = sorted(
[a for a in options.series.architectures
if a.architecture_tag in options.architectures],
key=attrgetter("architecture_tag"))
else:
options.architectures = sorted(
options.series.architectures, key=attrgetter("architecture_tag"))
def find_newest_publication(method, version_attr, **kwargs):
"""Hack around being unable to pass status=("Published", "Pending")."""
published_pubs = method(status="Published", **kwargs)
pending_pubs = method(status="Pending", **kwargs)
try:
newest_published = published_pubs[0]
newest_published_ver = getattr(newest_published, version_attr)
except IndexError:
try:
return pending_pubs[0]
except IndexError:
if kwargs["version"] is not None:
try:
return method(**kwargs)[0]
except IndexError:
return None
else:
return None
try:
newest_pending = pending_pubs[0]
newest_pending_ver = getattr(newest_pending, version_attr)
except IndexError:
return newest_published
if debian_support.version_compare(
newest_published_ver, newest_pending_ver) > 0:
return newest_published
else:
return newest_pending
def find_latest_published_binaries(options, package):
target_binaries = []
for architecture in options.architectures:
binary = find_newest_publication(
options.archive.getPublishedBinaries, "binary_package_version",
binary_name=package, version=options.version,
distro_arch_series=architecture, pocket=options.pocket,
exact_match=True)
if binary is not None:
target_binaries.append(binary)
if not target_binaries:
raise PackageMissing(
"Could not find binaries for '%s/%s' in %s" %
(package, options.version, options.suite))
return target_binaries
def find_latest_published_source(options, package):
source = find_newest_publication(
options.archive.getPublishedSources, "source_package_version",
source_name=package, version=options.version,
distro_series=options.series, pocket=options.pocket, exact_match=True)
if source is None:
raise PackageMissing(
"Could not find source '%s/%s' in %s" %
(package, options.version, options.suite))
return source
|