@@ -249,48 +249,51 @@ class MercurialRepository(BaseRepository
ignorews=ignore_whitespace,
context=context)))
def _check_url(self, url):
"""
Function will check given url and try to verify if it's a valid
link. Sometimes it may happened that mercurial will issue basic
auth request that can cause whole API to hang when used from python
or other external calls.
On failures it'll raise urllib2.HTTPError, return code 200 if url
is valid or True if it's a local path
from mercurial.util import url as Url
# those authnadlers are patched for python 2.6.5 bug an
# infinit looping when given invalid resources
from mercurial.url import httpbasicauthhandler, httpdigestauthhandler
# check first if it's not an local url
if os.path.isdir(url) or url.startswith('file:'):
return True
if('+' in url[:url.find('://')]):
url = url[url.find('+')+1:]
handlers = []
test_uri, authinfo = Url(url).authinfo()
if authinfo:
#create a password manager
passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
passmgr.add_password(*authinfo)
handlers.extend((httpbasicauthhandler(passmgr),
httpdigestauthhandler(passmgr)))
o = urllib2.build_opener(*handlers)
o.addheaders = [('Content-Type', 'application/mercurial-0.1'),
('Accept', 'application/mercurial-0.1')]
q = {"cmd": 'between'}
q.update({'pairs': "%s-%s" % ('0' * 40, '0' * 40)})
qs = '?%s' % urllib.urlencode(q)
cu = "%s%s" % (test_uri, qs)
req = urllib2.Request(cu, None, {})
try:
resp = o.open(req)
return resp.code == 200
@@ -349,87 +349,84 @@ def ValidRepoName(edit=False, old_data={
)
return value
return _validator
def ValidForkName(*args, **kwargs):
return ValidRepoName(*args, **kwargs)
def SlugifyName():
class _validator(formencode.validators.FancyValidator):
def _to_python(self, value, state):
return repo_name_slug(value)
def validate_python(self, value, state):
pass
def ValidCloneUri():
from rhodecode.lib.utils import make_ui
def url_handler(repo_type, url, proto, ui=None):
def url_handler(repo_type, url, ui=None):
if repo_type == 'hg':
from mercurial.httprepo import httprepository, httpsrepository
if proto == 'https':
if url.startswith('https'):
httpsrepository(make_ui('db'), url).capabilities
elif proto == 'http':
elif url.startswith('http'):
httprepository(make_ui('db'), url).capabilities
elif url.startswith('svn+http'):
from hgsubversion.svnrepo import svnremoterepo
svnremoterepo(make_ui('db'), url).capabilities
elif repo_type == 'git':
#TODO: write a git url validator
messages = {
'clone_uri': _(u'invalid clone url'),
'invalid_clone_uri': _(u'Invalid clone url, provide a '
'valid clone http\s url')
'valid clone http(s)/svn+http(s) url')
}
repo_type = value.get('repo_type')
url = value.get('clone_uri')
if not url:
elif url.startswith('https') or url.startswith('http'):
_type = 'https' if url.startswith('https') else 'http'
else:
url_handler(repo_type, url, _type, make_ui('db'))
url_handler(repo_type, url, make_ui('db'))
except Exception:
log.exception('Url validation failed')
msg = M(self, 'clone_uri')
raise formencode.Invalid(msg, value, state,
error_dict=dict(clone_uri=msg)
msg = M(self, 'invalid_clone_uri', state)
def ValidForkType(old_data={}):
'invalid_fork_type': _(u'Fork have to be the same type as parent')
if old_data['repo_type'] != value:
msg = M(self, 'invalid_fork_type', state)
error_dict=dict(repo_type=msg)
def ValidPerms(type_='repo'):
if type_ == 'group':
EMPTY_PERM = 'group.none'
elif type_ == 'repo':
EMPTY_PERM = 'repository.none'
Status change: