Commit 52d217c5 authored by Kazuhiko Shiozaki's avatar Kazuhiko Shiozaki

py2/py3: use six.moves.urllib for compatibility.

parent ff71de01
......@@ -31,7 +31,7 @@ from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.HTTPResponse import HTTPResponse
from Products.PluggableAuthService.interfaces.plugins import IAuthenticationPlugin
from DateTime import DateTime
import urllib
from six.moves.urllib.parse import urlencode
import six.moves.http_client
import base64
from six.moves import cStringIO as StringIO
......@@ -140,7 +140,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
response = self.publish('/%s/Base_getUserCaption?%s' % (
self.portal.getId(),
urllib.urlencode({
urlencode({
'access_token': access_token.getId(),
'access_token_secret': access_token.getReference()})))
self.assertEqual(response.getStatus(), six.moves.http_client.OK)
......
......@@ -755,7 +755,7 @@ class TestAuthenticationPolicy(ERP5TypeTestCase):
response = self.publish(
portal.absolute_url_path() + '/logged_in',
basic='test-05:used_ALREADY_1234',
stdin=StringIO(urllib.urlencode({'came_from': 'https://www.erp5.com'})),
stdin=StringIO(six.moves.urllib.parse.urlencode({'came_from': 'https://www.erp5.com'})),
request_method='POST',
)
redirect_url = six.moves.urllib.parse.urlparse(response.getHeader("Location"))
......@@ -818,7 +818,7 @@ class TestAuthenticationPolicy(ERP5TypeTestCase):
def submit_reset_password_dialog(new_password):
return self.publish(
'%s/portal_password' % self.portal.getPath(),
stdin=StringIO(urllib.urlencode({
stdin=StringIO(six.moves.urllib.parse.urlencode({
'Base_callDialogMethod:method': '',
'dialog_id': 'PasswordTool_viewResetPassword',
'dialog_method': 'PasswordTool_changeUserPassword',
......@@ -873,7 +873,7 @@ class TestAuthenticationPolicy(ERP5TypeTestCase):
return self.publish(
'%s/portal_preferences' % self.portal.getPath(),
basic='%s:current' % self.id(),
stdin=StringIO(urllib.urlencode({
stdin=StringIO(six.moves.urllib.parse.urlencode({
'Base_callDialogMethod:method': '',
'dialog_id': 'PreferenceTool_viewChangePasswordDialog',
'dialog_method': 'PreferenceTool_setNewPassword',
......
......@@ -26,7 +26,7 @@
##############################################################################
import os
from urllib import urlencode
from six.moves.urllib.parse import urlencode
import tempfile
from DateTime import DateTime
from zLOG import LOG
......
from Products.ERP5Type.UnrestrictedMethod import UnrestrictedMethod
import urllib
from six.moves.urllib.parse import urlencode
import mechanize
def getProductPrice(product):
......@@ -9,7 +9,7 @@ def getProductPrice(product):
def submitPaypalNVPRequest(parameter_dict, nvp_url):
request = mechanize.Request(nvp_url)
params = urllib.urlencode(parameter_dict)
params = urlencode(parameter_dict)
try:
response = mechanize.urlopen(request, data=params)
except:
......
......@@ -30,7 +30,7 @@
import os
import string
import unittest
import urllib
from six.moves.urllib.parse import quote
from unittest import skip
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
......@@ -567,14 +567,14 @@ class TestCommerce(ERP5TypeTestCase):
self.logout()
self.assertEqual(1, len(self.website.SaleOrder_getShoppingCartItemList()))
self.website.SaleOrder_paymentRedirect()
self.assertTrue(urllib.quote("You need to create an account to " \
self.assertTrue(quote("You need to create an account to " \
"continue. If you already have please login.") in
self.app.REQUEST.RESPONSE.getHeader('location'))
# but it should work if the user is authenticated
self.loginByUserName('customer')
self.portal.SaleOrder_paymentRedirect()
self.assertTrue(urllib.quote("SaleOrder_viewAsWeb") in
self.assertTrue(quote("SaleOrder_viewAsWeb") in
self.app.REQUEST.RESPONSE.getHeader('location'))
def test_10_deleteShoppingCartItem(self):
......@@ -587,7 +587,7 @@ class TestCommerce(ERP5TypeTestCase):
# Trying to remove
self.portal.SaleOrder_deleteShoppingCartItem()
self.assertTrue(urllib.quote("Please select an item.") in
self.assertTrue(quote("Please select an item.") in
self.app.REQUEST.RESPONSE.getHeader('location'))
# Check if the item still into the Shopping Cart
......@@ -600,7 +600,7 @@ class TestCommerce(ERP5TypeTestCase):
# Check if the Product have been removed sucessfully
self.assertTrue(
urllib.quote("Successfully removed from shopping cart.") in
quote("Successfully removed from shopping cart.") in
self.app.REQUEST.RESPONSE.getHeader('location'))
# Check if the Shopping Cart is empty
......@@ -861,7 +861,7 @@ class TestCommerce(ERP5TypeTestCase):
self.loginByUserName('toto')
self.portal.SaleOrder_paymentRedirect()
self.assertTrue(urllib.quote("SaleOrder_viewAsWeb") in
self.assertTrue(quote("SaleOrder_viewAsWeb") in
self.app.REQUEST.RESPONSE.getHeader('location'))
def test_23_getShoppingCartCustomer(self):
......
......@@ -31,7 +31,7 @@
from functools import partial
from six.moves import cStringIO as StringIO
import unittest
import urllib
from six.moves.urllib.parse import urlencode
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from DateTime import DateTime
from ZPublisher.cookie import normalizeCookieParameterName
......@@ -73,7 +73,7 @@ class TestAuoLogout(ERP5TypeTestCase):
"""
portal = self.getPortal()
stdin = urllib.urlencode({
stdin = urlencode({
'__ac_name': self.manager_username,
'__ac_password': self.manager_password,
})
......
......@@ -35,7 +35,7 @@ from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Acquisition import aq_base
from App.config import getConfiguration
from Products.ERP5Type.tests.Sequence import SequenceList, Sequence
from urllib import pathname2url
from six.moves.urllib.request import pathname2url
from Products.ERP5Type.Globals import PersistentMapping
from Products.ERP5Type.dynamic.lazy_class import ERP5BaseBroken
from Products.ERP5Type.tests.utils import LogInterceptor
......
......@@ -14,7 +14,7 @@
import base64
from six.moves import cStringIO as StringIO
import unittest
import urllib
from six.moves.urllib.parse import quote
from OFS.DTMLMethod import DTMLMethod
from OFS.Folder import Folder
......@@ -69,7 +69,7 @@ class ERP5CookieCrumblerTests (CookieCrumblerTests):
self.responseOut = StringIO()
self.req = makerequest(root, self.responseOut)
self.credentials = urllib.quote(
self.credentials = quote(
base64.encodestring('abraham:pass-w').replace('\012', ''))
def testCookieLongLogin(self):
......
......@@ -31,7 +31,6 @@ import pprint
import six.moves.http_client
import six.moves.urllib.parse
import base64
import urllib
import lxml.html
from AccessControl.SecurityManagement import newSecurityManager
......@@ -538,7 +537,7 @@ class TestERP5Core(ERP5TypeTestCase, ZopeTestCase.Functional):
self.assertEqual(0, person.getRelationCountForDeletion())
def delete(assert_deleted, obj):
redirect = self._Folder_delete(obj)
self.assertTrue((urllib.quote('Sorry, 1 item is in use.'), 'Deleted.')[assert_deleted]
self.assertTrue((six.moves.urllib.parse.quote('Sorry, 1 item is in use.'), 'Deleted.')[assert_deleted]
in redirect, redirect)
self.tic()
delete(0, organisation)
......@@ -567,7 +566,7 @@ class TestERP5Core(ERP5TypeTestCase, ZopeTestCase.Functional):
document_1.manage_permission('View', [], acquire=0)
document_1.manage_permission('Access contents information', [], acquire=0)
redirect = self._Folder_delete(document_2)
self.assert_(urllib.quote('Sorry, 1 item is in use.') in redirect, redirect)
self.assert_(six.moves.urllib.parse.quote('Sorry, 1 item is in use.') in redirect, redirect)
self.assertEqual(module.objectCount(), 2)
def test_getPropertyForUid(self):
......
......@@ -64,7 +64,8 @@ from Products.ERP5Type.Globals import get_request
import os
from threading import Thread
import six.moves.http_client
import urllib
from six.moves.urllib.request import urlopen
from six.moves.urllib.parse import urlencode
import difflib
import re
from AccessControl import Unauthorized
......@@ -228,7 +229,7 @@ class TestDocument(TestDocumentMixin):
# __ac=RVJQNVR5cGVUZXN0Q2FzZTo%3D is encoded ERP5TypeTestCase with empty password
url = '%s?%s&__ac=%s' %(uri, make_query(kw), 'RVJQNVR5cGVUZXN0Q2FzZTo%3D')
format_=kw.get('format', 'jpeg')
infile = urllib.urlopen(url)
infile = urlopen(url)
# save as file with proper incl. format filename (for some reasons PIL uses this info)
filename = "%s%stest-image-format-resize.%s" %(os.getcwd(), os.sep, format_)
f = open(filename, "w")
......@@ -2286,8 +2287,8 @@ return 1
"""
def getURL(uri, **kw):
# __ac=RVJQNVR5cGVUZXN0Q2FzZTo%3D is encoded ERP5TypeTestCase with empty password
url = '%s?%s&__ac=%s' %(uri, urllib.urlencode(kw), 'RVJQNVR5cGVUZXN0Q2FzZTo%3D')
return urllib.urlopen(url)
url = '%s?%s&__ac=%s' %(uri, urlencode(kw), 'RVJQNVR5cGVUZXN0Q2FzZTo%3D')
return urlopen(url)
ooo_document = self.portal.document_module.newContent(portal_type='Presentation')
upload_file = makeFileUpload('TEST-en-003.odp')
......
......@@ -307,7 +307,7 @@ class TestFreeSubscription(ERP5TypeTestCase):
def stepClickUnsubscriptionLinkInEvent(self, sequence=None, sequence_list=None,
**kw):
from urllib import urlopen
from six.moves.urllib.request import urlopen
link = sequence['unsubscription_link']
self.logout()
data = urlopen(link)
......
import feedparser, urllib2, socket
import feedparser, six.moves.urllib.request, six.moves.urllib.error, socket
from hashlib import md5
def getRssDataAsDict(context, url, username=None, password=None):
......@@ -11,9 +11,9 @@ def getRssDataAsDict(context, url, username=None, password=None):
# use authentication or not?
handlers = []
if username is not None and password is not None:
passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
passman = six.moves.urllib.request.HTTPPasswordMgrWithDefaultRealm()
passman.add_password(None, url, username, password)
auth_handler = urllib2.HTTPBasicAuthHandler(passman)
auth_handler = six.moves.urllib.request.HTTPBasicAuthHandler(passman)
handlers.append(auth_handler)
# set shorter timeouts and revert default at enf of read
......@@ -24,7 +24,7 @@ def getRssDataAsDict(context, url, username=None, password=None):
finally:
socket.setdefaulttimeout(default_timeout)
if d.bozo and isinstance(d.bozo_exception, urllib2.URLError):
if d.bozo and isinstance(d.bozo_exception, six.moves.urllib.error.URLError):
# we have an URL error
return {'status':-2}
elif d.bozo:
......
......@@ -26,7 +26,7 @@
##############################################################################
import suds
import urllib2
import six.moves.urllib.request
import ssl
import lxml.etree
......@@ -39,7 +39,7 @@ from suds.transport.https import HttpAuthenticated
class HTTPAuthenticatedUnverifiedSSL(HttpAuthenticated):
def u2handlers(self):
handlers = [ urllib2.HTTPSHandler(context=ssl._create_unverified_context()) ]
handlers = [ six.moves.urllib.request.HTTPSHandler(context=ssl._create_unverified_context()) ]
handlers.extend(HttpAuthenticated.u2handlers(self))
return handlers
......
......@@ -27,8 +27,8 @@
#
##############################################################################
import zope
from urllib import urlencode
from urllib2 import urlopen, Request
from six.moves.urllib.parse import urlencode
from six.moves.urllib.request import urlopen, Request
from zLOG import LOG, DEBUG
from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
......
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2006-2007 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# 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, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from DateTime import DateTime
from six.moves.urllib.request import urlopen
from lxml import etree
import re
from zLOG import LOG
def getObjectStringList(xml_string, element_to_find='object'):
"""
this function splits an ERP5 XML string into object
string list, each object string is converted
into utf-8 encoding and html entities are
translated into corresponding unicode code
"""
rss_doc = etree.fromstring(xml_string)
return [etree.tostring(node) for node in rss_doc.xpath('//%s' % element_to_find)]
def setTextContent(self):
"""
Edit text_content
"""
try:
text_content = urlopen(self.asURL()).read()
try:
text_content = unicode(text_content, "utf-8").encode("utf-8")
except UnicodeDecodeError:
text_content = unicode(text_content, "iso-8859-1").encode("utf-8")
except IOError:
text_content = None
if text_content is not None:
self.edit(text_content=text_content)
def setRSSItemProperties(self, rss_item):
pass
\ No newline at end of file
......@@ -57,8 +57,8 @@ def parseTestReport(text):
Return the content of a web page
"""
def urlread(url):
import urllib
return urllib.urlopen(url).read()
from six.moves.urllib.request import urlopen
return urlopen(url).read()
"""
Remove everything but the test in a webpage
......
......@@ -29,7 +29,8 @@
"""Receive or send SMS"""
#Import python module
import urllib
from six.moves.urllib.parse import urlencode, unquote
from six.moves.urllib.request import urlopen
from lxml import etree
from DateTime import DateTime
......@@ -91,14 +92,14 @@ class EssendexGateway(XMLObject):
if len(parts) == 1:
data = parts[0].split('=')
#Remove \n et \r from value
result[data[0]] = urllib.unquote(data[1].replace('\r','').replace('\n',''))
result[data[0]] = unquote(data[1].replace('\r','').replace('\n',''))
else:
#Mutil values
subresult = {}
for part in parts:
data = part.split('=')
subresult[data[0]] = urllib.unquote(data[1].replace('\r','').replace('\n',''))
subresult[data[0]] = unquote(data[1].replace('\r','').replace('\n',''))
result[index] = subresult
#Increment index for next
index += 1
......@@ -161,8 +162,8 @@ class EssendexGateway(XMLObject):
params['Test'] = 1
LOG("EssendexGateway", INFO, params)
params = urllib.urlencode(params)
page = urllib.urlopen(base_url, params)
params = urlencode(params)
page = urlopen(base_url, params)
result = self._fetchPageAsDict(page)
if result['Result'] == "OK":
message_ids = result.get('MessageIDs', "")
......@@ -171,7 +172,7 @@ class EssendexGateway(XMLObject):
return message_ids.split(",")
elif result['Result'] == "Error":
#we get an error when call the gateway
raise SMSGatewayError(urllib.unquote(result.get('Message', "Impossible to send the SMS")))
raise SMSGatewayError(unquote(result.get('Message', "Impossible to send the SMS")))
elif result['Result'] == "Test":
#just a test, no message id
return None
......@@ -190,15 +191,15 @@ class EssendexGateway(XMLObject):
'MessageID': message_id,
}
params = urllib.urlencode(params)
page = urllib.urlopen(base_url, params)
params = urlencode(params)
page = urlopen(base_url, params)
result = self._fetchPageAsDict(page)
if result['Result'] == "OK":
return result.get('MessageStatus').lower()
elif result['Result'] == "Error":
#we get an error when call the gateway
raise SMSGatewayError(urllib.unquote(result.get('Message', "Impossible to get the message status")))
raise SMSGatewayError(unquote(result.get('Message', "Impossible to get the message status")))
security.declarePublic('receive')
def receive(self, REQUEST, **kw):
......@@ -319,8 +320,8 @@ class EssendexGateway(XMLObject):
params['Test'] = 1
LOG("EssendexGateway", INFO, params)
params = urllib.urlencode(params)
page = urllib.urlopen(base_url, params)
params = urlencode(params)
page = urlopen(base_url, params)
result = self._fetchPageAsDict(page)
if result['Result'] == "OK":
......@@ -344,6 +345,6 @@ class EssendexGateway(XMLObject):
LOG("EssendexGateway", INFO, result)
elif result['Result'] == "Error":
#we get an error when call the gateway
raise SMSGatewayError(urllib.unquote(result.get('Message', "Impossible to get last message list")))
raise SMSGatewayError(unquote(result.get('Message', "Impossible to get last message list")))
......@@ -29,7 +29,8 @@
"""Receive or send SMS"""
#Import python module
import urllib
from six.moves.urllib.parse import urlencode, unquote
from six.moves.urllib.request import urlopen
from DateTime import DateTime
#Import Zope module
......@@ -179,8 +180,8 @@ class MobytGateway(XMLObject):
LOG("MobytGateway", INFO, params)
result = {'status': "Test"}
else:
params = urllib.urlencode(params)
page = urllib.urlopen(base_url, params)
params = urlencode(params)
page = urlopen(base_url, params)
result = self._fetchSendResponseAsDict(page)
#Check result and return
......@@ -188,7 +189,7 @@ class MobytGateway(XMLObject):
return [result.get('status_info', "")] #return message id (gateway side)
elif result['status'] == "KO":
#we get an error when call the gateway
raise SMSGatewayError(urllib.unquote(result.get('status_info', "Impossible to send the SMS")))
raise SMSGatewayError(unquote(result.get('status_info', "Impossible to send the SMS")))
elif result['status'] == "Test":
#just a test, no message id
return None
......@@ -206,8 +207,8 @@ class MobytGateway(XMLObject):
"type" : 'notify',
"schema" : 1 }
params = urllib.urlencode(params)
page = urllib.urlopen(base_url, params)
params = urlencode(params)
page = urlopen(base_url, params)
result = self._fetchStatusResponseAsDict(page)
if result['status'] == "OK":
......@@ -223,7 +224,7 @@ class MobytGateway(XMLObject):
elif result['status'] == "KO":
#we get an error when call the gateway
raise SMSGatewayError(urllib.unquote(result.get('status_info', "Impossible to get the message status")))
raise SMSGatewayError(unquote(result.get('status_info', "Impossible to get the message status")))
security.declarePublic('receive')
def receive(self,REQUEST):
......
......@@ -25,8 +25,8 @@
#
##############################################################################
from urllib import urlencode
from urllib2 import urlopen
from six.moves.urllib.parse import urlencode
from six.moves.urllib.request import urlopen
def OpenAMLogout(self, url, token):
code = urlopen(url, urlencode({'subjectid':token})).getcode()
......
......@@ -27,7 +27,7 @@
##############################################################################
from urllib import quote_plus
from six.moves.urllib.parse import quote_plus
from six.moves.urllib.parse import urlparse
from six.moves.urllib.parse import urljoin
import logging
......
......@@ -33,9 +33,9 @@ def UpdateImage(image):
image._update_image_info()
def urlread(url, safe_return=0):
import urllib
from six.moves.urllib.request import urlopen
try:
return urllib.urlopen(url).read()
return urlopen(url).read()
except IOError as e:
if safe_return:
# Return an Selenium test code that will obviously fail. This
......
......@@ -37,7 +37,7 @@ from Products.ERP5Type import Permissions
from webdav.NullResource import NullResource
import urllib
from six.moves.urllib.parse import unquote
MARKER = []
......@@ -69,7 +69,7 @@ class StaticWebSection(WebSection):
# Drop the automatically added VirtualHostMonster object ID
virtual_url_part_tuple = request.get('VIRTUAL_URL_PARTS', None)
if (virtual_url_part_tuple is not None) and \
(not urllib.unquote(virtual_url_part_tuple[-1]).endswith("/".join(url_list))):
(not unquote(virtual_url_part_tuple[-1]).endswith("/".join(url_list))):
url_list.pop(0)
if request.get('ACTUAL_URL', '').endswith("/"): # or len(url_list) == 0:
......
......@@ -3,7 +3,8 @@ from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
from zLOG import LOG, WARNING
import random, string, hashlib, urllib2, socket
import random, string, hashlib, socket
from six.moves.urllib.request import Request, urlopen
from six.moves.urllib.parse import urlparse
from six import string_types as basestring
try:
......@@ -103,8 +104,8 @@ class WechatService(XMLObject):
params['sign'] = self.calculateSign(params, self.getServiceApiKey())
LOG('WechatService', WARNING,
"getSandboxKey : data = %s SANDBOX_KEY_URL = %s" % (self.convert_dict_to_xml(params), SANDBOX_KEY_URL), error=False)
result = urllib2.Request(SANDBOX_KEY_URL, data=self.convert_dict_to_xml(params))
result_data = urllib2.urlopen(result)
result = Request(SANDBOX_KEY_URL, data=self.convert_dict_to_xml(params))
result_data = urlopen(result)
result_read = result_data.read()
result_dict_content = self.convert_xml_to_dict(result_read)
return_code = result_dict_content.get('return_code', '')
......@@ -151,8 +152,8 @@ class WechatService(XMLObject):
LOG('callWechatApi', WARNING,
"data = %s URL = %s" % (self.convert_dict_to_xml(wechat_dict), wechat_url + URL), error=False)
# send data
result = urllib2.Request(wechat_url + URL, data=self.convert_dict_to_xml(wechat_dict))
result_data = urllib2.urlopen(result)
result = Request(wechat_url + URL, data=self.convert_dict_to_xml(wechat_dict))
result_data = urlopen(result)
result_read = result_data.read()
result_dict_content = self.convert_xml_to_dict(result_read)
return_code = result_dict_content['return_code']
......
......@@ -15,13 +15,13 @@
from threading import Thread
from time import sleep
from urllib import addinfourl
from urllib import splithost
from urllib import splituser
from urllib import unquote
from urllib import splittype
from six.moves.urllib.parse import splithost
from six.moves.urllib.parse import splituser
from six.moves.urllib.parse import unquote
from six.moves.urllib.parse import splittype
import string
from urllib import FancyURLopener
from six.moves.urllib.request import FancyURLopener
from six.moves.http_cookies import SimpleCookie
def main():
......
......@@ -30,7 +30,7 @@
import threading
import unittest
import urllib
from six.moves.urllib.request import urlopen
import transaction
import pkg_resources
from DateTime import DateTime
......@@ -161,7 +161,7 @@ class TestInvalidationBug(ERP5TypeTestCase):
storage._server = None
# ... monkey-patch done
## create object
urllib.urlopen(new_content_url).read()
urlopen(new_content_url).read()
## validate reindex activity
activity_tool.distribute()
self.assertEqual(1, len(activity_tool.getMessageList()))
......
......@@ -29,7 +29,7 @@ import os
import tarfile
import xml.parsers.expat
import xml.dom.minidom
from urllib import url2pathname
from six.moves.urllib.request import url2pathname
from ZODB.DemoStorage import DemoStorage
from ZODB import DB
from Products.ERP5Type.XMLExportImport import importXML
......
......@@ -36,7 +36,7 @@ import os
import dircache
import mimetypes, mimetools
from email.utils import formatdate
class DirectoryFileHandler(urllib2.FileHandler):
class DirectoryFileHandler(six.moves.urllib.request.FileHandler):
"""
Extends the file handler to provide an HTML
representation of local directories.
......@@ -55,7 +55,7 @@ class DirectoryFileHandler(urllib2.FileHandler):
def open_local_file(self, req):
host = req.get_host()
file = req.get_selector()
localfile = urllib2.url2pathname(file)
localfile = six.moves.urllib.request.url2pathname(file)
stats = os.stat(localfile)
size = stats.st_size
modified = formatdate(stats.st_mtime, usegmt=True)
......@@ -64,7 +64,7 @@ class DirectoryFileHandler(urllib2.FileHandler):
'Content-type: %s\nContent-length: %d\nLast-modified: %s\n' %
(mtype or 'text/plain', size, modified)))
if host:
host, port = urllib.splitport(host)
host, port = six.moves.urllib.parse.splitport(host)
if not host or \
(not port and socket.gethostbyname(host) in self.get_names()):
try:
......@@ -73,16 +73,16 @@ class DirectoryFileHandler(urllib2.FileHandler):
s.write('<html><head><base href="%s"/></head><body>' % ('file:' + file))
s.write('<p>Directory Content:</p>')
for f in file_list:
s.write('<p><a href="%s">%s</a></p>\n' % (urllib.quote(f), f))
s.write('<p><a href="%s">%s</a></p>\n' % (six.moves.urllib.parse.quote(f), f))
s.write('</body></html>')
s.seek(0)
headers = mimetools.Message(StringIO(
'Content-type: %s\nContent-length: %d\nLast-modified: %s\n' %
('text/html', size, modified)))
return urllib2.addinfourl(s, headers, 'file:' + file)
return six.moves.urllib.response.addinfourl(s, headers, 'file:' + file)
except OSError:
return urllib2.addinfourl(open(localfile, 'rb'),
return six.moves.urllib.response.addinfourl(open(localfile, 'rb'),
headers, 'file:'+file)
raise urllib2.URLError('file not on local host')
opener = urllib2.build_opener(DirectoryFileHandler)
urllib2.install_opener(opener)
raise six.moves.urllib.error.URLError('file not on local host')
opener = six.moves.urllib.request.build_opener(DirectoryFileHandler)
six.moves.urllib.request.install_opener(opener)
......@@ -18,7 +18,6 @@ import string
import sys
import time
import traceback
import urllib
from six.moves import configparser
from contextlib import contextmanager
from io import BytesIO
......
......@@ -27,7 +27,7 @@
##############################################################################
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from six.moves import cStringIO as StringIO
import urllib
from six.moves.urllib.parse import urlencode
import six.moves.http_client
......@@ -117,7 +117,7 @@ class TestUpgradeInstanceWithOldDataFs(ERP5TypeTestCase):
ret = self.publish(
'%s/portal_alarms/promise_check_upgrade' % self.portal.getPath(),
basic='%s:current' % self.id(),
stdin=StringIO(urllib.urlencode({
stdin=StringIO(urlencode({
'Base_callDialogMethod:method': '',
'dialog_id': 'Alarm_viewSolveDialog',
'dialog_method': 'Alarm_solve',
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment