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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""FTPMaster base classes.
PackageLocation and SoyuzScript.
"""
__metaclass__ = type
__all__ = [
'SoyuzScriptError',
'SoyuzScript',
]
from zope.component import getUtility
from lp.app.errors import NotFoundError
from lp.services.scripts.base import (
LaunchpadScript,
LaunchpadScriptFailure,
)
from lp.soyuz.adapters.packagelocation import build_package_location
from lp.soyuz.enums import ArchivePurpose
from lp.soyuz.interfaces.component import IComponentSet
class SoyuzScriptError(Exception):
"""Raised when a soyuz script failed.
The textual content should explain the error.
"""
class SoyuzScript(LaunchpadScript):
"""`LaunchpadScript` extended for Soyuz related use.
Possible exceptions raised are:
* `PackageLocationError`: specified package or distro does not exist
* `LaunchpadScriptError`: only raised if entering via main(), ie this
code is running as a genuine script. In this case, this is
also the _only_ exception to be raised.
The test harness doesn't enter via main(), it calls mainTask(), so
it does not see LaunchpadScriptError.
Each script can extend:
* `usage`: string describing the expected command-line format;
* `description`: string describing the tool;
* `success_message`: string to be presented on successful runs;
* `mainTask`: a method to actually perform a specific task.
See `add_my_options` for the default `SoyuzScript` command-line options.
"""
location = None
success_message = "Done."
def add_my_options(self):
"""Adds SoyuzScript default options.
Any subclass may override this method and call the add_*_options
individually to reduce the number of available options as necessary.
"""
self.add_transaction_options()
self.add_distro_options()
self.add_package_location_options()
self.add_archive_options()
def add_transaction_options(self):
"""Add SoyuzScript transaction-related options."""
self.parser.add_option(
'-n', '--dry-run', dest='dryrun', default=False,
action='store_true', help='Do not commit changes.')
self.parser.add_option(
'-y', '--confirm-all', dest='confirm_all',
default=False, action='store_true',
help='Do not prompt the user for confirmation.')
def add_distro_options(self):
"""Add SoyuzScript distro-related options."""
self.parser.add_option(
'-d', '--distribution', dest='distribution_name',
default='ubuntu', action='store',
help='Distribution name.')
self.parser.add_option(
'-s', '--suite', dest='suite', default=None,
action='store', help='Suite name.')
def add_package_location_options(self):
"""Add SoyuzScript package location-related options."""
self.parser.add_option(
"-a", "--architecture", dest="architecture", default=None,
help="Architecture tag.")
self.parser.add_option(
'-e', '--version', dest='version', default=None,
action='store',
help='Optional package version, defaults to the current version.')
self.parser.add_option(
"-c", "--component", dest="component", default=None,
help="Component name.")
def add_archive_options(self):
"""Add SoyuzScript archive-related options."""
self.parser.add_option(
'-p', '--ppa', dest='archive_owner_name', action='store',
help='Archive owner name in case of PPA operations')
self.parser.add_option(
'--ppa-name', dest='archive_name', action='store', default="ppa",
help='PPA name in case of PPA operations')
self.parser.add_option(
'-j', '--partner', dest='partner_archive', default=False,
action='store_true',
help='Specify partner archive')
def _validatePublishing(self, currently_published):
"""Validate the given publishing record.
Check if it matches the desired 'pocket' and 'component'.
"""
if not self.options.component:
return
try:
desired_component = getUtility(IComponentSet)[
self.options.component]
except NotFoundError, err:
raise SoyuzScriptError(err)
if currently_published.component != desired_component:
raise SoyuzScriptError(
"%s was skipped because it is not in %s component" % (
currently_published.displayname,
desired_component.name.upper()))
def findLatestPublishedSource(self, name):
"""Return a suitable `SourcePackagePublishingHistory`."""
assert self.location is not None, 'Undefined location.'
# Avoiding circular imports.
from lp.soyuz.interfaces.publishing import active_publishing_status
published_sources = self.location.archive.getPublishedSources(
name=name, version=self.options.version,
status=active_publishing_status,
distroseries=self.location.distroseries,
pocket=self.location.pocket,
exact_match=True)
try:
latest_source = published_sources[0]
except IndexError:
raise SoyuzScriptError(
"Could not find source '%s/%s' in %s" % (
name, self.options.version, self.location))
self._validatePublishing(latest_source)
return latest_source
def findLatestPublishedBinaries(self, name):
"""Build a list of suitable `BinaryPackagePublishingHistory`.
Try to find a group of binary package release matching the current
context. 'architecture' or 'version', if passed via command-line,
will restrict the lookup accordingly.
"""
assert self.location is not None, 'Undefined location.'
# Avoiding circular imports.
from lp.soyuz.interfaces.publishing import active_publishing_status
target_binaries = []
if self.options.architecture is None:
architectures = self.location.distroseries.architectures
else:
try:
architectures = [
self.location.distroseries[self.options.architecture]]
except NotFoundError, err:
raise SoyuzScriptError(err)
for architecture in architectures:
binaries = self.location.archive.getAllPublishedBinaries(
name=name, version=self.options.version,
status=active_publishing_status,
distroarchseries=architecture,
pocket=self.location.pocket,
exact_match=True)
if not binaries:
continue
binary = binaries[0]
try:
self._validatePublishing(binary)
except SoyuzScriptError, err:
self.logger.warn(err)
else:
target_binaries.append(binary)
if not target_binaries:
raise SoyuzScriptError(
"Could not find binaries for '%s/%s' in %s" % (
name, self.options.version, self.location))
return target_binaries
def _getUserConfirmation(self, full_question=None, valid_answers=None):
"""Use raw_input to collect user feedback.
Return True if the user typed the first value of the given
'valid_answers' (defaults to 'yes') or False otherwise.
"""
if valid_answers is None:
valid_answers = ['yes', 'no']
display_answers = '[%s]' % (', '.join(valid_answers))
if full_question is None:
full_question = 'Confirm this transaction? %s ' % display_answers
else:
full_question = '%s %s' % (full_question, display_answers)
answer = None
while answer not in valid_answers:
answer = raw_input(full_question)
return answer == valid_answers[0]
def waitForUserConfirmation(self):
"""Blocks the script flow waiting for a user confirmation.
Return True immediately if options.confirm_all was passed or after
getting a valid confirmation, False otherwise.
"""
if not self.options.confirm_all and not self._getUserConfirmation():
return False
return True
def setupLocation(self):
"""Setup `PackageLocation` for context distribution and suite."""
# These can raise PackageLocationError, but we're happy to pass
# it upwards.
if getattr(self.options, 'partner_archive', ''):
self.location = build_package_location(
self.options.distribution_name,
self.options.suite,
ArchivePurpose.PARTNER)
elif getattr(self.options, 'archive_owner_name', ''):
self.location = build_package_location(
self.options.distribution_name,
self.options.suite,
ArchivePurpose.PPA,
self.options.archive_owner_name,
self.options.archive_name)
else:
self.location = build_package_location(
self.options.distribution_name,
self.options.suite)
def finishProcedure(self):
"""Script finalization procedure.
'dry-run' command-line option will case the transaction to be
immediatelly aborted.
In normal mode it will ask for user confirmation (see
`waitForUserConfirmation`) and will commit the transaction or abort
it according to the user answer.
Returns True if the transaction was committed, False otherwise.
"""
if self.options.dryrun:
self.logger.info('Dry run, so nothing to commit.')
self.txn.abort()
return False
confirmed = self.waitForUserConfirmation()
if confirmed:
self.txn.commit()
self.logger.info('Transaction committed.')
self.logger.info(self.success_message)
return True
else:
self.logger.info("Ok, see you later")
self.txn.abort()
return False
def main(self):
"""LaunchpadScript entry point.
Can only raise LaunchpadScriptFailure - other exceptions are
absorbed into that.
"""
try:
self.setupLocation()
self.mainTask()
except SoyuzScriptError, err:
raise LaunchpadScriptFailure(err)
self.finishProcedure()
def mainTask(self):
"""Main task to be performed by the script"""
raise NotImplementedError
|