kallithea Changeset - 02e0d2d469bf

Changeset - 02e0d2d469bf

Parent rev.

Child rev.

[Not reviewed]

stable

0 2 0

Mads Kiilerich - 8 years ago 2018-05-07 11:38:13
mads@kiilerich.com

utils: move clone URI validator function to more generic utils.is_valid_repo_uri

No changes to the functionality, even though the API and implementation could
use some clean-up ...

2 files changed with 32 insertions and 30 deletions:

kallithea/lib/utils.py

kallithea/model/validators.py

0 comments (0 inline, 0 general)

kallithea/lib/utils.py

➞

Show inline comments

@@ @@ -244,48 +244,78 @@ def get_filesystem_repos(path): @@
                 if not os.access(cur_path, os.R_OK) or not os.access(cur_path, os.X_OK):
                     log.warning('ignoring repo path without access: %s', cur_path)
                     continue
                 if not os.access(cur_path, os.W_OK):
                     log.warning('repo path without write access: %s', cur_path)
                 try:
                     scm_info = get_scm(cur_path)
                     assert cur_path.startswith(path)
                     repo_path = cur_path[len(path) + 1:]
                     yield repo_path, scm_info
                     continue # no recursion
                 except VCSError:
                     # We should perhaps ignore such broken repos, but especially
                     # the bare git detection is unreliable so we dive into it
                     pass
             recurse_dirs.append(subdir)
         dirs[:] = recurse_dirs
 def is_valid_repo_uri(repo_type, url, ui):
     """Check if the url seems like a valid remote repo location - raise an Exception if any problems"""
     if repo_type == 'hg':
         from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
         if url.startswith('http') or url.startswith('ssh'):
             # initially check if it's at least the proper URL
             # or does it pass basic auth
             MercurialRepository._check_url(url, ui)
         elif url.startswith('svn+http'):
             from hgsubversion.svnrepo import svnremoterepo
             svnremoterepo(ui, url).svn.uuid
         elif url.startswith('git+http'):
             raise NotImplementedError()
         else:
             raise Exception('URI %s not allowed' % (url,))
     elif repo_type == 'git':
         from kallithea.lib.vcs.backends.git.repository import GitRepository
         if url.startswith('http') or url.startswith('git'):
             # initially check if it's at least the proper URL
             # or does it pass basic auth
             GitRepository._check_url(url)
         elif url.startswith('svn+http'):
             raise NotImplementedError()
         elif url.startswith('hg+http'):
             raise NotImplementedError()
         else:
             raise Exception('URI %s not allowed' % (url))
 def is_valid_repo(repo_name, base_path, scm=None):
     """
     Returns True if given path is a valid repository False otherwise.
     If scm param is given also compare if given scm is the same as expected
     from scm parameter
     :param repo_name:
     :param base_path:
     :param scm:
     :return True: if given path is a valid repository
     """
     full_path = os.path.join(safe_str(base_path), safe_str(repo_name))
     try:
         scm_ = get_scm(full_path)
         if scm:
             return scm_[0] == scm
         return True
     except VCSError:
         return False
 def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):

kallithea/model/validators.py

➞

Show inline comments

@@ @@ -9,49 +9,49 @@ @@
 # 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/>.
 """
 Set of generic validators
 """
 import os
 import re
 import formencode
 import logging
 from collections import defaultdict
 from pylons.i18n.translation import _
 from webhelpers.pylonslib.secure_form import authentication_token
 import sqlalchemy
 from formencode.validators import (
     UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,
     NotEmpty, IPAddress, CIDR, String, FancyValidator
+)
 from kallithea.lib.compat import OrderedSet
 from kallithea.lib import ipaddr
 from kallithea.lib.utils import repo_name_slug
+from kallithea.lib.utils import repo_name_slug, is_valid_repo_uri
 from kallithea.lib.utils2 import str2bool, aslist
 from kallithea.model.db import RepoGroup, Repository, UserGroup, User
 from kallithea.lib.exceptions import LdapImportError
 from kallithea.config.routing import ADMIN_PREFIX
 from kallithea.lib.auth import HasRepoGroupPermissionAny, HasPermissionAny
 # silence warnings and pylint
 UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \
     NotEmpty, IPAddress, CIDR, String, FancyValidator
 log = logging.getLogger(__name__)
 class StateObj(object):
     """
     this is needed to translate the messages using _() in validators
     """
     _ = staticmethod(_)
 def M(self, key, state=None, **kwargs):
     """
     returns string from self.message based on given key,
     passed kw params are used to substitute %(named)s params inside
@@ @@ -411,90 +411,62 @@ 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
     return _validator
 def ValidCloneUri():
     from kallithea.lib.utils import make_ui
     def url_handler(repo_type, url, ui):
         if repo_type == 'hg':
             from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
             if url.startswith('http') or url.startswith('ssh'):
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 MercurialRepository._check_url(url, ui)
             elif url.startswith('svn+http'):
                 from hgsubversion.svnrepo import svnremoterepo
                 svnremoterepo(ui, url).svn.uuid
             elif url.startswith('git+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url,))
         elif repo_type == 'git':
             from kallithea.lib.vcs.backends.git.repository import GitRepository
             if url.startswith('http') or url.startswith('git'):
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 GitRepository._check_url(url)
             elif url.startswith('svn+http'):
                 raise NotImplementedError()
             elif url.startswith('hg+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url))
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'clone_uri': _('Invalid repository URL'),
             'invalid_clone_uri': _('Invalid repository URL. It must be a '
                                    'valid http, https, ssh, svn+http or svn+https URL'),
+        }
         def validate_python(self, value, state):
             repo_type = value.get('repo_type')
             url = value.get('clone_uri')
             if url and url != value.get('clone_uri_hidden'):
                 try:
-                    url_handler(repo_type, url, make_ui('db', clear_session=False))
+                    is_valid_repo_uri(repo_type, url, make_ui('db', clear_session=False))
                 except Exception:
                     log.exception('URL validation failed')
                     msg = M(self, 'clone_uri')
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(clone_uri=msg)
+                    )
     return _validator
 def ValidForkType(old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_fork_type': _('Fork has to be the same type as parent')
+        }
         def validate_python(self, value, state):
             if old_data['repo_type'] != value:
                 msg = M(self, 'invalid_fork_type', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_type=msg)
+                )
     return _validator

0 comments (0 inline, 0 general)