[Checkins] SVN: zope.publisher/branches/py3-attempt2/ Py3 effort in progress
Andrey Lebedev
cvs-admin at zope.org
Mon Feb 18 14:38:24 UTC 2013
Log message for revision 129449:
Py3 effort in progress
Changed:
U zope.publisher/branches/py3-attempt2/src/zope/publisher/ftp.py
U zope.publisher/branches/py3-attempt2/src/zope/publisher/http.py
U zope.publisher/branches/py3-attempt2/src/zope/publisher/httpresults.txt
U zope.publisher/branches/py3-attempt2/src/zope/publisher/testing.py
U zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_baseresponse.py
U zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_ftp.py
U zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_http.py
U zope.publisher/branches/py3-attempt2/tox.ini
-=-
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/ftp.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/ftp.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/ftp.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -13,6 +13,7 @@
##############################################################################
"""FTP Publisher
"""
+import six
from zope.interface import implementer
from zope.publisher.interfaces.ftp import IFTPCredentials, IFTPRequest
from zope.publisher.base import BaseResponse, BaseRequest
@@ -27,7 +28,7 @@
def getResult(self):
if getattr(self, '_exc', None) is not None:
- raise self._exc[0], self._exc[1], self._exc[2]
+ six.reraise(self._exc[0], self._exc[1], self._exc[2])
return self._result
def handleException(self, exc_info):
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/http.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/http.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/http.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -14,6 +14,7 @@
"""HTTP Publisher
"""
import sys
+import base64
from io import BytesIO
from zope.i18n.interfaces import IUserPreferredCharsets
from zope.i18n.interfaces import IUserPreferredLanguages
@@ -49,6 +50,7 @@
else:
import http.cookies as cookies
from urllib.parse import splitport, quote, urlsplit
+ basestring = unicode = str
# Default Encoding
ENCODING = 'UTF-8'
@@ -77,7 +79,7 @@
dict[key] = val
if 'HTTP_CGI_AUTHORIZATION' in dict:
dict['HTTP_AUTHORIZATION'] = dict.pop('HTTP_CGI_AUTHORIZATION')
- if 'PATH_INFO' in dict:
+ if 'PATH_INFO' in dict and isinstance('PATH_INFO', bytes):
dict['PATH_INFO'] = dict['PATH_INFO'].decode('utf-8')
return dict
@@ -385,7 +387,7 @@
else:
protocol = 'http'
- if environ.has_key('HTTP_HOST'):
+ if 'HTTP_HOST' in environ:
host = environ['HTTP_HOST'].strip()
hostname, port = splitport(host)
else:
@@ -497,7 +499,9 @@
'See IHTTPCredentials'
if self._auth and self._auth.lower().startswith('basic '):
encoded = self._auth.split(None, 1)[-1]
- name, password = encoded.decode("base64").split(':', 1)
+ decoded = base64.b64decode(encoded.encode('iso-8859-1'))
+ name, password = bytes.split(decoded, b':', 1)
+ #name, password = base64.b64decode(encoded.encode('ascii')).split(':', 1)
return name, password
def unauthorized(self, challenge):
@@ -933,14 +937,13 @@
"for more information."
)
-def sort_charsets(x, y):
- if y[1] == 'utf-8':
- return 1
- if x[1] == 'utf-8':
- return -1
- return cmp(y, x)
+def sort_charsets(charset):
+ # Make utf-8 to be the last element of the sorted list
+ if charset[1] == 'utf-8':
+ return (1, charset)
+ # Otherwise, sort by charset
+ return (0, charset)
-
def extract_host(url):
scheme, host, path, query, fragment = urlsplit(url)
if ':' not in host:
@@ -1000,7 +1003,7 @@
# range , unlike many other encodings. Since Zope can easily use very
# different ranges, like providing a French-Chinese dictionary, it is
# always good to use UTF-8.
- charsets.sort(sort_charsets)
+ charsets.sort(key=sort_charsets, reverse=True)
charsets = [charset for quality, charset in charsets]
if sawstar and 'utf-8' not in charsets:
charsets.insert(0, 'utf-8')
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/httpresults.txt
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/httpresults.txt 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/httpresults.txt 2013-02-18 14:38:24 UTC (rev 129449)
@@ -67,6 +67,13 @@
To close, we'll build a quick example so you can see it working.
+(To make the code work in both python-2.x and python-3.x, define ``unicode`` name for
+python-3.x:
+
+ >>> import sys
+ >>> if sys.version_info[0] > 2:
+ ... unicode = str
+
>>> import zope.interface
>>> import zope.component
>>> from zope.publisher.browser import TestRequest
@@ -93,7 +100,7 @@
'text/html;charset=utf-8'
>>> res = tuple(request.response.consumeBodyIter())
>>> res
- ('<html>\n<head>\n<title>raw</title>\n</head>\n<body>\n<h1>Foo!</h1>\n</body>\n</html>',)
+ (b'<html>\n<head>\n<title>raw</title>\n</head>\n<body>\n<h1>Foo!</h1>\n</body>\n</html>',)
>>> len(res[0]) == int(request.response.getHeader('content-length'))
True
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/testing.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/testing.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/testing.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -13,17 +13,30 @@
##############################################################################
import sys
+import re
import contextlib
import zope.publisher.browser
import zope.security.management
import zope.security.testing
+from zope.testing import renormalizing
PY2 = sys.version_info[0] == 2
if PY2:
_u = unicode
+ import doctest
+ rules = [(re.compile("b('.*?')"), r"\1"),
+ (re.compile('b(".*?")'), r"\1"),
+ ]
+ output_checker = renormalizing.RENormalizing(rules)
else:
_u = str
+ rules = [(re.compile("u('.*?')"), r"\1"),
+ (re.compile('u(".*?")'), r"\1"),
+ (re.compile("b('.*?')"), r"\1"),
+ (re.compile('b(".*?")'), r"\1"),
+ ]
+ output_checker = renormalizing.RENormalizing(rules)
# These are enhanced versions of the ones in zope.security.testing,
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_baseresponse.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_baseresponse.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_baseresponse.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -18,9 +18,7 @@
from zope.publisher.base import BaseResponse
from zope.publisher.interfaces import IResponse
from zope.interface.verify import verifyObject
-from StringIO import StringIO
-
class TestBaseResponse(TestCase):
def test_interface(self):
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_ftp.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_ftp.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_ftp.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -14,14 +14,14 @@
"""FTP Publisher Tests
"""
import sys
-from cStringIO import StringIO
+from io import BytesIO
from unittest import TestCase, TestSuite, main, makeSuite
import zope.publisher.ftp
class Test(TestCase):
def setUp(self):
- self.__input = StringIO('')
+ self.__input = BytesIO(b'')
env = {'credentials': ('bob', '123'),
'path': '/a/b/c',
'command': 'foo',
Modified: zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_http.py
===================================================================
--- zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_http.py 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/src/zope/publisher/tests/test_http.py 2013-02-18 14:38:24 UTC (rev 129449)
@@ -17,8 +17,7 @@
import sys
import tempfile
import unittest
-from cStringIO import StringIO
-from Cookie import CookieError
+from io import BytesIO
from doctest import DocFileSuite
import zope.event
@@ -47,7 +46,12 @@
import BaseTestIPublisherRequest
from zope.publisher.tests.basetestiapplicationrequest \
import BaseTestIApplicationRequest
+from zope.publisher.testing import output_checker
+if sys.version_info[0] > 2:
+ from http.cookies import CookieError
+else:
+ from Cookie import CookieError
@implementer(ILoggingInfo)
@@ -63,7 +67,7 @@
return self._id
-data = '''\
+data = b'''\
line 1
line 2
line 3'''
@@ -86,7 +90,7 @@
return result
def testRead(self):
- stream = HTTPInputStream(StringIO(data), {})
+ stream = HTTPInputStream(BytesIO(data), {})
output = ''
self.assertEqual(output, self.getCacheStreamValue(stream))
output += stream.read(5)
@@ -96,7 +100,7 @@
self.assertEqual(data, self.getCacheStreamValue(stream))
def testReadLine(self):
- stream = HTTPInputStream(StringIO(data), {})
+ stream = HTTPInputStream(BytesIO(data), {})
output = stream.readline()
self.assertEqual(output, self.getCacheStreamValue(stream))
output += stream.readline()
@@ -108,7 +112,7 @@
self.assertEqual(data, self.getCacheStreamValue(stream))
def testReadLines(self):
- stream = HTTPInputStream(StringIO(data), {})
+ stream = HTTPInputStream(BytesIO(data), {})
output = ''.join(stream.readlines(4))
self.assertEqual(output, self.getCacheStreamValue(stream))
output += ''.join(stream.readlines())
@@ -116,7 +120,7 @@
self.assertEqual(data, self.getCacheStreamValue(stream))
def testGetCacheStream(self):
- stream = HTTPInputStream(StringIO(data), {})
+ stream = HTTPInputStream(BytesIO(data), {})
stream.read(5)
self.assertEqual(data, stream.getCacheStream().read())
@@ -127,24 +131,24 @@
# definitely over that).
# HTTPInputStream understands both CONTENT_LENGTH...
- stream = HTTPInputStream(StringIO(data), {'CONTENT_LENGTH': '100000'})
+ stream = HTTPInputStream(BytesIO(data), {'CONTENT_LENGTH': '100000'})
self.assert_(isinstance(stream.getCacheStream(), TempFileType))
# ... and HTTP_CONTENT_LENGTH.
- stream = HTTPInputStream(StringIO(data), {'HTTP_CONTENT_LENGTH':
+ stream = HTTPInputStream(BytesIO(data), {'HTTP_CONTENT_LENGTH':
'100000'})
self.assert_(isinstance(stream.getCacheStream(), TempFileType))
# If CONTENT_LENGTH is absent or empty, it takes the value
# given in HTTP_CONTENT_LENGTH:
- stream = HTTPInputStream(StringIO(data),
+ stream = HTTPInputStream(BytesIO(data),
{'CONTENT_LENGTH': '',
'HTTP_CONTENT_LENGTH': '100000'})
self.assert_(isinstance(stream.getCacheStream(), TempFileType))
# In fact, HTTPInputStream can be instantiated with both an
# empty CONTENT_LENGTH and an empty HTTP_CONTENT_LENGTH:
- stream = HTTPInputStream(StringIO(data),
+ stream = HTTPInputStream(BytesIO(data),
{'CONTENT_LENGTH': '',
'HTTP_CONTENT_LENGTH': ''})
@@ -194,26 +198,26 @@
class Item(object):
"""Required docstring for the publisher."""
def __call__(self, a, b):
- return "%s, %s" % (`a`, `b`)
+ return "%s, %s" % (repr(a), repr(b))
self.app = AppRoot()
self.app.folder = Folder()
self.app.folder.item = Item()
self.app.xxx = Item()
- def _createRequest(self, extra_env={}, body=""):
+ def _createRequest(self, extra_env={}, body=b""):
env = self._testEnv.copy()
env.update(extra_env)
if len(body):
env['CONTENT_LENGTH'] = str(len(body))
publication = DefaultPublication(self.app)
- instream = StringIO(body)
+ instream = BytesIO(body)
request = HTTPRequest(instream, env)
request.setPublication(publication)
return request
- def _publisherResults(self, extra_env={}, body=""):
+ def _publisherResults(self, extra_env={}, body=b""):
request = self._createRequest(extra_env, body)
response = request.response
publish(request, handle_errors=False)
@@ -264,7 +268,7 @@
# test HTTP/1.0
env = {'SERVER_PROTOCOL':'HTTP/1.0'}
- request = self._createRequest(env, '')
+ request = self._createRequest(env, b'')
location = request.response.redirect('http://foobar.com/redirected')
self.assertEquals(location, 'http://foobar.com/redirected')
self.assertEquals(request.response.getStatus(), 302)
@@ -273,17 +277,17 @@
# test HTTP/1.1
env = {'SERVER_PROTOCOL':'HTTP/1.1'}
- request = self._createRequest(env, '')
+ request = self._createRequest(env, b'')
location = request.response.redirect('http://foobar.com/redirected')
self.assertEquals(request.response.getStatus(), 303)
# test explicit status
- request = self._createRequest(env, '')
+ request = self._createRequest(env, b'')
request.response.redirect('http://foobar.com/explicit', 304)
self.assertEquals(request.response.getStatus(), 304)
# test non-string location, like URLGetter
- request = self._createRequest(env, '')
+ request = self._createRequest(env, b'')
request.response.redirect(request.URL)
self.assertEquals(request.response.getStatus(), 303)
self.assertEquals(request.response.getHeader('location'),
@@ -292,7 +296,7 @@
def testUntrustedRedirect(self):
# Redirects are by default only allowed to target the same host as the
# request was directed to. This is to counter fishing.
- request = self._createRequest({}, '')
+ request = self._createRequest({}, b'')
self.assertRaises(
ValueError,
request.response.redirect, 'http://phishing-inc.com')
@@ -339,7 +343,7 @@
def testUnregisteredStatus(self):
# verify we can set the status to an unregistered int value
- request = self._createRequest({}, '')
+ request = self._createRequest({}, b'')
request.response.setStatus(289)
self.assertEquals(request.response.getStatus(), 289)
@@ -502,14 +506,15 @@
def testBasicAuth(self):
from zope.publisher.interfaces.http import IHTTPCredentials
+ import base64
req = self._createRequest()
verifyObject(IHTTPCredentials, req)
lpq = req._authUserPW()
self.assertEquals(lpq, None)
env = {}
- login, password = ("tim", "123:456")
- s = ("%s:%s" % (login, password)).encode("base64").rstrip()
- env['HTTP_AUTHORIZATION'] = "Basic %s" % s
+ login, password = (b"tim", b"123:456")
+ s = base64.b64encode(b':'.join((login, password)))
+ env['HTTP_AUTHORIZATION'] = "Basic %s" % s.decode('ascii')
req = self._createRequest(env)
lpw = req._authUserPW()
self.assertEquals(lpw, (login, password))
@@ -922,11 +927,11 @@
def _Test__new(self, environ=None, **kw):
if environ is None:
environ = kw
- return HTTPRequest(StringIO(''), environ)
+ return HTTPRequest(BytesIO(b''), environ)
def test_IApplicationRequest_bodyStream(self):
- request = HTTPRequest(StringIO('spam'), {})
- self.assertEqual(request.bodyStream.read(), 'spam')
+ request = HTTPRequest(BytesIO(b'spam'), {})
+ self.assertEqual(request.bodyStream.read(), b'spam')
# Needed by BaseTestIEnumerableMapping tests:
def _IEnumerableMapping__stateDict(self):
@@ -973,7 +978,8 @@
suite.addTest(unittest.makeSuite(TestHTTPResponse))
suite.addTest(unittest.makeSuite(HTTPInputStreamTests))
suite.addTest(DocFileSuite(
- '../httpresults.txt', setUp=cleanUp, tearDown=cleanUp))
+ '../httpresults.txt', setUp=cleanUp, tearDown=cleanUp,
+ checker=output_checker))
suite.addTest(unittest.makeSuite(APITests))
return suite
Modified: zope.publisher/branches/py3-attempt2/tox.ini
===================================================================
--- zope.publisher/branches/py3-attempt2/tox.ini 2013-02-18 14:00:53 UTC (rev 129448)
+++ zope.publisher/branches/py3-attempt2/tox.ini 2013-02-18 14:38:24 UTC (rev 129449)
@@ -18,7 +18,6 @@
zope.proxy
zope.security
-
[testenv:coverage]
basepython =
python2.7
More information about the checkins
mailing list