py3: migrate from urllib2 to urllib

Based on 2to3 urllib, but with unnecessary imports removed.

kallithea/bin/base.py

@@ -29,7 +29,7 b' import os'
 import pprint
 import random
 import sys
-import urllib2
+import urllib.request
 
 from kallithea.lib import ext_json
 from kallithea.lib.utils2 import ascii_bytes
@@ -68,10 +68,10 b' def api_call(apikey, apihost, method=Non'
         raise Exception('please specify method name !')
     apihost = apihost.rstrip('/')
     id_ = random.randrange(1, 9999)
-    req = urllib2.Request('%s/_admin/api' % apihost,
+    req = urllib.request.Request('%s/_admin/api' % apihost,
                       data=ascii_bytes(ext_json.dumps(_build_data(id_))),
                       headers={'content-type': 'text/plain'})
-    ret = urllib2.urlopen(req)
+    ret = urllib.request.urlopen(req)
     raw_json = ret.read()
     json_data = ext_json.loads(raw_json)
     id_ret = json_data['id']

kallithea/bin/ldap_sync.py

@@ -27,7 +27,7 b' Original author and date, and relevant c'
 
 from __future__ import print_function
 
-import urllib2
+import urllib.request
 import uuid
 from configparser import ConfigParser
 
@@ -83,9 +83,9 b' class API(object):'
 
         data = ascii_bytes(ext_json.dumps(data))
         headers = {'content-type': 'text/plain'}
-        req = urllib2.Request(self.url, data, headers)
+        req = urllib.request.Request(self.url, data, headers)
 
-        response = urllib2.urlopen(req)
+        response = urllib.request.urlopen(req)
         response = ext_json.load(response)
 
         if uid != response["id"]:

kallithea/lib/auth_modules/auth_crowd.py

@@ -28,7 +28,8 b' Original author and date, and relevant c'
 
 import base64
 import logging
-import urllib2
+import urllib.parse
+import urllib.request
 
 from kallithea.lib import auth_modules, ext_json
 from kallithea.lib.compat import hybrid_property
@@ -72,10 +73,10 b' class CrowdServer(object):'
         self._make_opener()
 
     def _make_opener(self):
-        mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
+        mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
         mgr.add_password(None, self._uri, self.user, self.passwd)
-        handler = urllib2.HTTPBasicAuthHandler(mgr)
-        self.opener = urllib2.build_opener(handler)
+        handler = urllib.request.HTTPBasicAuthHandler(mgr)
+        self.opener = urllib.request.build_opener(handler)
 
     def _request(self, url, body=None, headers=None,
                  method=None, noformat=False,
@@ -88,7 +89,7 b' class CrowdServer(object):'
         if headers:
             _headers.update(headers)
         log.debug("Sent to crowd at %s:\nHeaders: %s\nBody:\n%s", url, _headers, body)
-        req = urllib2.Request(url, body, _headers)
+        req = urllib.request.Request(url, body, _headers)
         if method:
             req.get_method = lambda: method
 
@@ -119,14 +120,14 b' class CrowdServer(object):'
         """Authenticate a user against crowd. Returns brief information about
         the user."""
         url = ("%s/rest/usermanagement/%s/authentication?username=%s"
-               % (self._uri, self._version, urllib2.quote(username)))
+               % (self._uri, self._version, urllib.parse.quote(username)))
         body = ascii_bytes(ext_json.dumps({"value": password}))
         return self._request(url, body)
 
     def user_groups(self, username):
         """Retrieve a list of groups to which this user belongs."""
         url = ("%s/rest/usermanagement/%s/user/group/nested?username=%s"
-               % (self._uri, self._version, urllib2.quote(username)))
+               % (self._uri, self._version, urllib.parse.quote(username)))
         return self._request(url)
 
 

kallithea/lib/middleware/simplehg.py

@@ -30,7 +30,7 b' Original author and date, and relevant c'
 
 import logging
 import os
-import urllib
+import urllib.parse
 
 import mercurial.hgweb
 
@@ -121,7 +121,7 b' class SimpleHg(BaseVCSController):'
                             break
                         action = 'pull'
                         for cmd_arg in hgarg[5:].split(';'):
-                            cmd, _args = urllib.unquote_plus(cmd_arg).split(' ', 1)
+                            cmd, _args = urllib.parse.unquote_plus(cmd_arg).split(' ', 1)
                             op = cmd_mapping.get(cmd, 'push')
                             if op != 'pull':
                                 assert op == 'push'

kallithea/lib/recaptcha.py

@@ -1,7 +1,7 b''
 # -*- coding: utf-8 -*-
 import json
-import urllib
-import urllib2
+import urllib.parse
+import urllib.request
 
 
 class RecaptchaResponse(object):
@@ -30,13 +30,13 b' def submit(g_recaptcha_response, private'
             return s.encode('utf-8')
         return s
 
-    params = urllib.urlencode({
+    params = urllib.parse.urlencode({
         'secret': encode_if_necessary(private_key),
         'remoteip': encode_if_necessary(remoteip),
         'response': encode_if_necessary(g_recaptcha_response),
     }).encode('ascii')
 
-    req = urllib2.Request(
+    req = urllib.request.Request(
         url="https://www.google.com/recaptcha/api/siteverify",
         data=params,
         headers={
@@ -45,7 +45,7 b' def submit(g_recaptcha_response, private'
         }
     )
 
-    httpresp = urllib2.urlopen(req)
+    httpresp = urllib.request.urlopen(req)
     return_values = json.loads(httpresp.read())
     httpresp.close()
 

kallithea/lib/utils2.py

@@ -36,7 +36,7 b' import os'
 import pwd
 import re
 import time
-import urllib
+import urllib.parse
 
 import urlobject
 from tg.i18n import ugettext as _
@@ -322,14 +322,14 b' def credentials_filter(uri):'
 
 def get_clone_url(clone_uri_tmpl, prefix_url, repo_name, repo_id, username=None):
     parsed_url = urlobject.URLObject(prefix_url)
-    prefix = safe_unicode(urllib.unquote(parsed_url.path.rstrip('/')))
+    prefix = safe_unicode(urllib.parse.unquote(parsed_url.path.rstrip('/')))
     try:
         system_user = pwd.getpwuid(os.getuid()).pw_name
     except Exception: # TODO: support all systems - especially Windows
         system_user = 'kallithea' # hardcoded default value ...
     args = {
         'scheme': parsed_url.scheme,
-        'user': safe_unicode(urllib.quote(safe_str(username or ''))),
+        'user': safe_unicode(urllib.parse.quote(safe_str(username or ''))),
         'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")
         'prefix': prefix, # undocumented, empty or starting with /
         'repo': repo_name,

kallithea/lib/vcs/backends/git/repository.py

@@ -14,8 +14,9 b' import logging'
 import os
 import re
 import time
-import urllib
-import urllib2
+import urllib.error
+import urllib.parse
+import urllib.request
 from collections import OrderedDict
 
 import mercurial.url  # import httpbasicauthhandler, httpdigestauthhandler
@@ -178,19 +179,19 b' class GitRepository(BaseRepository):'
 
         if authinfo:
             # create a password manager
-            passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
+            passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
 
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
 
-        o = urllib2.build_opener(*handlers)
+        o = urllib.request.build_opener(*handlers)
         o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git
 
-        req = urllib2.Request(
+        req = urllib.request.Request(
             "%s?%s" % (
                 test_uri,
-                urllib.urlencode({"service": 'git-upload-pack'})
+                urllib.parse.urlencode({"service": 'git-upload-pack'})
             ))
 
         try:
@@ -199,12 +200,12 b' class GitRepository(BaseRepository):'
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
-            raise urllib2.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
+            raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
 
         # now detect if it's proper git repo
         gitdata = resp.read()
         if 'service=git-upload-pack' not in gitdata:
-            raise urllib2.URLError(
+            raise urllib.error.URLError(
                 "url [%s] does not look like an git" % cleaned_uri)
 
         return True

kallithea/lib/vcs/backends/hg/repository.py

@@ -13,8 +13,9 b' import datetime'
 import logging
 import os
 import time
-import urllib
-import urllib2
+import urllib.error
+import urllib.parse
+import urllib.request
 from collections import OrderedDict
 
 import mercurial.commands
@@ -314,20 +315,20 b' class MercurialRepository(BaseRepository'
 
         if authinfo:
             # create a password manager
-            passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
+            passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
 
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
 
-        o = urllib2.build_opener(*handlers)
+        o = urllib.request.build_opener(*handlers)
         o.addheaders = [('Content-Type', 'application/mercurial-0.1'),
                         ('Accept', 'application/mercurial-0.1')]
 
-        req = urllib2.Request(
+        req = urllib.request.Request(
             "%s?%s" % (
                 test_uri,
-                urllib.urlencode({
+                urllib.parse.urlencode({
                     'cmd': 'between',
                     'pairs': "%s-%s" % ('0' * 40, '0' * 40),
                 })
@@ -339,14 +340,14 b' class MercurialRepository(BaseRepository'
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
-            raise urllib2.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
+            raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
 
         if not url_prefix: # skip svn+http://... (and git+... too)
             # now check if it's a proper hg repo
             try:
                 mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             except Exception as e:
-                raise urllib2.URLError(
+                raise urllib.error.URLError(
                     "url [%s] does not look like an hg repo org_exc: %s"
                     % (cleaned_uri, e))
 
@@ -490,7 +491,7 b' class MercurialRepository(BaseRepository'
         """
         url = safe_str(url)
         if url != 'default' and '://' not in url:
-            url = "file:" + urllib.pathname2url(url)
+            url = "file:" + urllib.request.pathname2url(url)
         return url
 
     def get_changeset(self, revision=None):

kallithea/tests/functional/test_admin_repos.py

@@ -1,7 +1,7 b''
 # -*- coding: utf-8 -*-
 
 import os
-import urllib
+import urllib.parse
 
 import mock
 import pytest
@@ -410,7 +410,7 b' class _BaseTestCase(base.TestController)'
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
-                               % (urllib.quote(repo_name), repo_name_unicode))
+                               % (urllib.parse.quote(repo_name), repo_name_unicode))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).one()

kallithea/tests/functional/test_forks.py

@@ -1,6 +1,6 b''
 # -*- coding: utf-8 -*-
 
-import urllib
+import urllib.parse
 
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, User
@@ -159,7 +159,7 b' class _BaseTestCase(base.TestController)'
         response = self.app.get(base.url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain(
-            """<a href="/%s">%s</a>""" % (urllib.quote(fork_name), fork_name)
+            """<a href="/%s">%s</a>""" % (urllib.parse.quote(fork_name), fork_name)
         )
         fork_repo = Repository.get_by_repo_name(safe_unicode(fork_name))
         assert fork_repo
@@ -180,7 +180,7 b' class _BaseTestCase(base.TestController)'
         response = self.app.get(base.url(controller='forks', action='forks',
                                     repo_name=fork_name))
         response.mustcontain(
-            """<a href="/%s">%s</a>""" % (urllib.quote(fork_name_2), fork_name_2)
+            """<a href="/%s">%s</a>""" % (urllib.parse.quote(fork_name_2), fork_name_2)
         )
 
         # remove these forks

kallithea/tests/other/test_vcs_operations.py

@@ -32,7 +32,7 b' import os'
 import re
 import tempfile
 import time
-import urllib2
+import urllib.request
 from subprocess import PIPE, Popen
 from tempfile import _RandomNameSequence
 
@@ -329,11 +329,11 b' class TestVCSOperations(base.TestControl'
                          owner=base.TEST_USER_ADMIN_LOGIN,
                          repo_type=vt.repo_type),
         }
-        req = urllib2.Request(
+        req = urllib.request.Request(
             'http://%s:%s/_admin/api' % webserver.server_address,
             data=ascii_bytes(json.dumps(params)),
             headers={'content-type': 'application/json'})
-        response = urllib2.urlopen(req)
+        response = urllib.request.urlopen(req)
         result = json.loads(response.read())
         # Expect something like:
         # {u'result': {u'msg': u'Created new repository `new_XXX`', u'task': None, u'success': True}, u'id': 7, u'error': None}

kallithea/tests/scripts/manual_test_crawler.py

@@ -37,8 +37,8 b' import os'
 import sys
 import tempfile
 import time
-import urllib
-import urllib2
+import urllib.parse
+import urllib.request
 from os.path import dirname
 
 from kallithea.lib import vcs
@@ -73,13 +73,13 b' PROJECTS = ['
 
 
 cj = http.cookiejar.FileCookieJar(os.path.join(tempfile.gettempdir(), 'rc_test_cookie.txt'))
-o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
+o = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
 o.addheaders = [
     ('User-agent', 'kallithea-crawler'),
     ('Accept-Language', 'en - us, en;q = 0.5')
 ]
 
-urllib2.install_opener(o)
+urllib.request.install_opener(o)
 
 
 def _get_repo(proj):
@@ -101,7 +101,7 b' def test_changelog_walk(proj, pages=100)'
 
         page = '/'.join((proj, 'changelog',))
 
-        full_uri = (BASE_URI % page) + '?' + urllib.urlencode({'page': i})
+        full_uri = (BASE_URI % page) + '?' + urllib.parse.urlencode({'page': i})
         s = time.time()
         f = o.open(full_uri)
 

kallithea/tests/vcs/test_git.py

@@ -1,7 +1,7 b''
 import datetime
 import os
 import sys
-import urllib2
+import urllib.error
 
 import mock
 import pytest
@@ -32,7 +32,7 b' class TestGitRepository(object):'
 
     def test_git_cmd_injection(self):
         repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'
-        with pytest.raises(urllib2.URLError):
+        with pytest.raises(urllib.error.URLError):
             # Should fail because URL will contain the parts after ; too
             GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)