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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# pylint: disable-msg=E0211,E0213
"""Code import interfaces."""
__metaclass__ = type
__all__ = [
'ICodeImport',
'ICodeImportSet',
]
import re
from CVS.protocol import (
CVSRoot,
CvsRootError,
)
from lazr.restful.declarations import (
call_with,
export_as_webservice_entry,
export_write_operation,
exported,
REQUEST_USER,
)
from lazr.restful.fields import ReferenceChoice
from zope.interface import (
Attribute,
Interface,
)
from zope.schema import (
Choice,
Datetime,
Int,
TextLine,
Timedelta,
)
from canonical.launchpad import _
from lp.app.validators import LaunchpadValidationError
from lp.code.enums import (
CodeImportReviewStatus,
RevisionControlSystems,
)
from lp.code.interfaces.branch import IBranch
from lp.services.fields import (
PublicPersonChoice,
URIField,
)
def validate_cvs_root(cvsroot):
try:
root = CVSRoot(cvsroot)
except CvsRootError, e:
raise LaunchpadValidationError(e)
if root.method == 'local':
raise LaunchpadValidationError('Local CVS roots are not allowed.')
if not root.hostname:
raise LaunchpadValidationError('CVS root is invalid.')
if root.hostname.count('.') == 0:
raise LaunchpadValidationError(
'Please use a fully qualified host name.')
return True
def validate_cvs_module(cvsmodule):
valid_module = re.compile('^[a-zA-Z][a-zA-Z0-9_/.+-]*$')
if not valid_module.match(cvsmodule):
raise LaunchpadValidationError(
'The CVS module contains illegal characters.')
if cvsmodule == 'CVS':
raise LaunchpadValidationError(
'A CVS module can not be called "CVS".')
return True
def validate_cvs_branch(branch):
if branch and re.match('^[a-zA-Z][a-zA-Z0-9_-]*$', branch):
return True
else:
raise LaunchpadValidationError('Your CVS branch name is invalid.')
class ICodeImport(Interface):
"""A code import to a Bazaar Branch."""
export_as_webservice_entry()
id = Int(readonly=True, required=True)
date_created = Datetime(
title=_("Date Created"), required=True, readonly=True)
branch = exported(
ReferenceChoice(
title=_('Branch'), required=True, readonly=True,
vocabulary='Branch', schema=IBranch,
description=_("The Bazaar branch produced by the "
"import system.")))
registrant = PublicPersonChoice(
title=_('Registrant'), required=True, readonly=True,
vocabulary='ValidPersonOrTeam',
description=_("The person who initially requested this import."))
review_status = exported(
Choice(
title=_("Review Status"), vocabulary=CodeImportReviewStatus,
default=CodeImportReviewStatus.NEW, readonly=True,
description=_("Before a code import is performed, it is reviewed."
" Only reviewed imports are processed.")))
rcs_type = exported(
Choice(title=_("Type of RCS"), readonly=True,
required=True, vocabulary=RevisionControlSystems,
description=_(
"The version control system to import from. "
"Can be CVS or Subversion.")))
url = exported(
URIField(title=_("URL"), required=False, readonly=True,
description=_("The URL of the VCS branch."),
allowed_schemes=["http", "https", "svn", "git"],
allow_userinfo=True,
allow_port=True,
allow_query=False, # Query makes no sense in Subversion.
allow_fragment=False, # Fragment makes no sense in Subversion.
trailing_slash=False)) # See http://launchpad.net/bugs/56357.
cvs_root = exported(
TextLine(title=_("Repository"), required=False, readonly=True,
constraint=validate_cvs_root,
description=_("The CVSROOT. "
"Example: :pserver:anonymous@anoncvs.gnome.org:/cvs/gnome")))
cvs_module = exported(
TextLine(title=_("Module"), required=False, readonly=True,
constraint=validate_cvs_module,
description=_("The path to import within the repository."
" Usually, it is the name of the project.")))
date_last_successful = exported(
Datetime(title=_("Last successful"), required=False, readonly=True))
update_interval = Timedelta(
title=_("Update interval"), required=False, description=_(
"The user-specified time between automatic updates of this import. "
"If this is unspecified, the effective update interval is a default "
"value selected by Launchpad administrators."))
effective_update_interval = Timedelta(
title=_("Effective update interval"), required=True, readonly=True,
description=_(
"The effective time between automatic updates of this import. "
"If the user did not specify an update interval, this is a default "
"value selected by Launchpad administrators."))
def getImportDetailsForDisplay():
"""Get a one-line summary of the location this import is from."""
import_job = Choice(
title=_("Current job"),
readonly=True, vocabulary='CodeImportJob',
description=_(
"The current job for this import, either pending or running."))
results = Attribute("The results for this code import.")
consecutive_failure_count = Attribute(
"How many times in a row this import has failed.")
def updateFromData(data, user):
"""Modify attributes of the `CodeImport`.
Creates and returns a MODIFY `CodeImportEvent` if changes were made.
This method preserves the invariant that a `CodeImportJob` exists for
a given import if and only if its review_status is REVIEWED, creating
and deleting jobs as necessary.
:param data: dictionary whose keys are attribute names and values are
attribute values.
:param user: user who made the change, to record in the
`CodeImportEvent`. May be ``None``.
:return: The MODIFY `CodeImportEvent`, if any changes were made, or
None if no changes were made.
"""
def tryFailingImportAgain(user):
"""Try a failing import again.
This method sets the review_status back to REVIEWED and requests the
import be attempted as soon as possible.
The import must be in the FAILING state.
:param user: the user who is requesting the import be tried again.
"""
@call_with(requester=REQUEST_USER)
@export_write_operation()
def requestImport(requester, error_if_already_requested=False):
"""Request that an import be tried soon.
This method will schedule an import to happen soon for this branch.
The import must be in the Reviewed state, if not then a
CodeImportNotInReviewedState error will be thrown. If using the
API then a status code of 400 will result.
If the import is already running then a CodeImportAlreadyRunning
error will be thrown. If using the API then a status code of
400 will result.
The two cases can be distinguished over the API by seeing if the
exception names appear in the body of the response.
If used over the API and the request has already been made then this
method will silently do nothing.
If called internally then the error_if_already_requested parameter
controls whether a CodeImportAlreadyRequested exception will be
thrown in that situation.
:return: None
"""
class ICodeImportSet(Interface):
"""Interface representing the set of code imports."""
def new(registrant, target, branch_name, rcs_type, url=None,
cvs_root=None, cvs_module=None, review_status=None,
owner=None):
"""Create a new CodeImport.
:param target: An `IBranchTarget` that the code is associated with.
:param owner: The `IPerson` to set as the owner of the branch, or
None to use registrant. registrant must be a member of owner to
do this.
"""
def get(id):
"""Get a CodeImport by its id.
Raises `NotFoundError` if no such import exists.
"""
def getByBranch(branch):
"""Get the CodeImport, if any, associated to a Branch."""
def getByCVSDetails(cvs_root, cvs_module):
"""Get the CodeImport with the specified CVS details."""
def getByURL(url):
"""Get the CodeImport with the url."""
def delete(id):
"""Delete a CodeImport given its id."""
def search(review_status=None, rcs_type=None):
"""Find the CodeImports of the given status and type.
:param review_status: An entry from the `CodeImportReviewStatus`
schema, or None, which signifies 'any status'.
:param rcs_type: An entry from the `RevisionControlSystems`
schema, or None, which signifies 'any type'.
"""
|