[Repoze-checkins] r1073 - in repoze.zope2/trunk: . repoze/zope2 repoze/zope2/tests
Chris McDonough
chrism at agendaless.com
Wed Jun 11 18:53:21 EDT 2008
Author: Chris McDonough <chrism at agendaless.com>
Date: Wed Jun 11 18:53:21 2008
New Revision: 1073
Log:
- Deal with Unauthorized exceptions properly: allow
RESPONSE._unauthorized to be run and return a value rather than
unconditionally triggering a basic authentication dialog.
This version depends on repoze.obob 0.4+.
(also remove experimental json marshalling)
Modified:
repoze.zope2/trunk/CHANGES.txt
repoze.zope2/trunk/repoze/zope2/request.py
repoze.zope2/trunk/repoze/zope2/tests/base.py
repoze.zope2/trunk/repoze/zope2/tests/test_request.py
repoze.zope2/trunk/repoze/zope2/tests/test_z2bob.py
repoze.zope2/trunk/repoze/zope2/z2bob.py
repoze.zope2/trunk/setup.py
Modified: repoze.zope2/trunk/CHANGES.txt
==============================================================================
--- repoze.zope2/trunk/CHANGES.txt (original)
+++ repoze.zope2/trunk/CHANGES.txt Wed Jun 11 18:53:21 2008
@@ -1,8 +1,10 @@
-After 0.4.1
+0.4.2 (2008-06-11)
- - Add json serialization. Send a request with an Accept: header
- with 'application/json' in it to receive a result serialized as
- json.
+ - Deal with Unauthorized exceptions properly: allow
+ RESPONSE._unauthorized to be run and return a value rather than
+ unconditionally triggering a basic authentication dialog.
+
+ This version depends on repoze.obob 0.4+.
0.4.1 (2008-05-27)
Modified: repoze.zope2/trunk/repoze/zope2/request.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/request.py (original)
+++ repoze.zope2/trunk/repoze/zope2/request.py Wed Jun 11 18:53:21 2008
@@ -14,8 +14,12 @@
""" Utilities for constructing Zope request and response objects """
-from ZPublisher.HTTPResponse import HTTPResponse
from ZPublisher.HTTPRequest import HTTPRequest
+
+from ZPublisher.HTTPResponse import HTTPResponse
+from ZPublisher.HTTPResponse import status_codes
+from ZPublisher.HTTPResponse import status_reasons
+
from tempfile import NamedTemporaryFile
class RepozeHTTPResponse(HTTPResponse):
@@ -38,3 +42,11 @@
request.no_acquire_flag = False # used by DefaultPublishTraverse
return request
+def convertResponseCode(code_or_reason):
+ if isinstance(code_or_reason, str):
+ code_or_reason = code_or_reason.lower()
+ code = status_codes.get(code_or_reason, 500)
+ reason = status_reasons.get(code, 'Unknown')
+ return code, reason
+
+
Modified: repoze.zope2/trunk/repoze/zope2/tests/base.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/tests/base.py (original)
+++ repoze.zope2/trunk/repoze/zope2/tests/base.py Wed Jun 11 18:53:21 2008
@@ -29,9 +29,10 @@
class DummyResponse:
unauth_called = False
+ under_unauth_called = False
urls_reset = False
unauthorized_raises = None
- status = None
+ status = 200
def __init__(self):
self.headers = {}
self.cookies = {}
@@ -60,6 +61,8 @@
return self.headers.get(header)
def setHeader(self, header, value):
self.headers[header] = value
+ def _unauthorized(self):
+ self.under_unauth_called = True
def unauthorized(self):
self.unauth_called = True
if self.unauthorized_raises:
@@ -81,7 +84,7 @@
self.environ = env
if not 'TraversalRequestNameStack' in self.environ:
self.environ['TraversalRequestNameStack'] = []
- self.response = DummyResponse()
+ self.response = self.RESPONSE = DummyResponse()
self.steps = self._steps = []
self.args = []
from ZPublisher.BaseRequest import UNSPECIFIED_ROLES
Modified: repoze.zope2/trunk/repoze/zope2/tests/test_request.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/tests/test_request.py (original)
+++ repoze.zope2/trunk/repoze/zope2/tests/test_request.py Wed Jun 11 18:53:21 2008
@@ -2,7 +2,7 @@
class TestMakeRequest(unittest.TestCase):
def _getFUT(self):
- from repoze.zope2.z2bob import makeRequest
+ from repoze.zope2.request import makeRequest
return makeRequest
def test_makeRequest(self):
@@ -16,6 +16,16 @@
from repoze.zope2.request import RepozeHTTPResponse
self.failUnless(isinstance(request.response, RepozeHTTPResponse))
+class TestConvertResponseCode(unittest.TestCase):
+ def _getFUT(self):
+ from repoze.zope2.request import convertResponseCode
+ return convertResponseCode
+
+ def test_convertResponseCode(self):
+ f = self._getFUT()
+ self.assertEqual(f(500), (500, 'Internal Server Error'))
+ self.assertEqual(f('Unauthorized'), (401, 'Unauthorized'))
+
class TestRepozeHTTPResponse(unittest.TestCase):
def _getTargetClass(self):
from repoze.zope2.request import RepozeHTTPResponse
Modified: repoze.zope2/trunk/repoze/zope2/tests/test_z2bob.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/tests/test_z2bob.py (original)
+++ repoze.zope2/trunk/repoze/zope2/tests/test_z2bob.py Wed Jun 11 18:53:21 2008
@@ -200,32 +200,6 @@
helper.setup()
self.assertEqual(helper.default_page, 'POST')
- def test_setup_jsonresponse(self):
- from zope.security.management import queryInteraction
- from zope.security.management import endInteraction
- env = _makeEnviron()
- env['SERVER_NAME'] = 'www.example.com'
- env['SERVER_PORT'] = '80'
- env['HTTP_ACCEPT'] = 'application/json,text/html'
- helper = self._makeOne(env)
- if queryInteraction() is not None:
- endInteraction()
- helper.setup()
- self.assertEqual(helper.json, True)
-
- def test_setup_nonjsonresponse(self):
- from zope.security.management import queryInteraction
- from zope.security.management import endInteraction
- env = _makeEnviron()
- env['SERVER_NAME'] = 'www.example.com'
- env['SERVER_PORT'] = '80'
- env['HTTP_ACCEPT'] = 'text/html'
- helper = self._makeOne(env)
- if queryInteraction() is not None:
- endInteraction()
- helper.setup()
- self.assertEqual(helper.json, False)
-
def test_setup_probablydav(self):
from zope.security.management import queryInteraction
from zope.security.management import endInteraction
@@ -1030,37 +1004,6 @@
body = result[0]
self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, body)
- def test_map_result_None_to_json(self):
- env = _makeEnviron()
- helper = self._makeOne(env)
- helper.json = True
- status, headers, result = helper.map_result(None)
- self.assertEqual(status, '200 OK')
- self.assertEqual(len(headers), 2)
- self.assertEqual(headers[0], ('Content-Length', '4'))
- self.assertEqual(headers[1], ('Content-Type','application/json'))
- self.assertEqual(result, ['null'])
-
- def test_map_result_dict_to_json(self):
- env = _makeEnviron()
- helper = self._makeOne(env)
- helper.json = True
- status, headers, result = helper.map_result({'a':1, 'b':2})
- self.assertEqual(status, '200 OK')
- self.assertEqual(len(headers), 2)
- self.assertEqual(headers[0], ('Content-Length', '16'))
- self.assertEqual(headers[1], ('Content-Type','application/json'))
- self.assertEqual(result, ['{"a": 1, "b": 2}'])
-
- def test_map_result_ob_to_json(self):
- env = _makeEnviron()
- helper = self._makeOne(env)
- helper.json = True
- class Foo:
- a = 1
- foo = Foo()
- self.assertRaises(TypeError, helper.map_result, foo)
-
def test_map_result_tacks_on_charset_to_text_responses(self):
helper = self._makeOne()
response = helper.request.response
@@ -1109,93 +1052,88 @@
response.headers['content-type'] = 'text/html'
self.assertRaises(ValueError, helper.map_result, NotIterable)
- def test_handle_exception_doesnt_catch_unknown(self):
+ def test_handle_exception_reraises_unknown(self):
helper = self._makeOne()
class MyException(Exception):
pass
exc_info = (MyException, MyException('foo'), None)
- self.assertEqual(None, helper.handle_exception(exc_info))
+ self.assertRaises(MyException, helper.handle_exception, exc_info)
def test_handle_exception_unauthorized_string(self):
helper = self._makeOne()
+ helper.request.response.status = '200 OK'
exc_info = ('Unauthorized', 'foo', None)
- self.assertRaises(httpexceptions.HTTPUnauthorized,
- helper.handle_exception, exc_info)
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '3'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(body, ['foo'])
def test_handle_exception_unauthorized_zexceptions(self):
helper = self._makeOne()
from zExceptions import Unauthorized
+ helper.request.response.status = '200 OK'
exc_info = (Unauthorized, Unauthorized('foo'), None)
- self.assertRaises(httpexceptions.HTTPUnauthorized,
- helper.handle_exception, exc_info)
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '51'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(
+ body,
+ ["You are not allowed to access 'foo' in this context"])
def test_handle_exception_unauthorized_accesscontrol(self):
helper = self._makeOne()
from AccessControl import Unauthorized
exc_info = (Unauthorized, Unauthorized('foo'), None)
- self.assertRaises(httpexceptions.HTTPUnauthorized,
- helper.handle_exception, exc_info)
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '51'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(
+ body,
+ ["You are not allowed to access 'foo' in this context"])
def test_handle_exception_unauthorized_zexceptions_value_correct(self):
helper = self._makeOne()
from zExceptions import Unauthorized
exc_info = (Unauthorized, Unauthorized('You arent authorized'), None)
- try:
- helper.handle_exception(exc_info)
- except:
- t, v = sys.exc_info()[:2]
- self.assertEqual(t, httpexceptions.HTTPUnauthorized)
- self.assertEqual(v.detail, 'You arent authorized')
- headers = v.headers
- self.assertEqual(len(headers), 1)
- headername, headervalue = headers[0]
- self.assertEqual(headername, 'WWW-Authenticate')
- self.assertEqual(headervalue, 'Basic realm="Zope"')
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '20'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(body, ['You arent authorized'])
def test_handle_exception_unauthorized_acesscontrol_value_correct(self):
helper = self._makeOne()
from AccessControl import Unauthorized
exc_info = (Unauthorized, Unauthorized('You arent authorized'), None)
- try:
- helper.handle_exception(exc_info)
- except:
- t, v = sys.exc_info()[:2]
- self.assertEqual(t, httpexceptions.HTTPUnauthorized)
- self.assertEqual(v.detail, 'You arent authorized')
- headers = v.headers
- self.assertEqual(len(headers), 1)
- headername, headervalue = headers[0]
- self.assertEqual(headername, 'WWW-Authenticate')
- self.assertEqual(headervalue, 'Basic realm="Zope"')
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '20'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(body, ['You arent authorized'])
def test_handle_exception_unauthorized_string_value_correct(self):
helper = self._makeOne()
exc_info = ('Unauthorized', 'You arent authorized', None)
- try:
- helper.handle_exception(exc_info)
- except:
- t, v = sys.exc_info()[:2]
- self.assertEqual(t, httpexceptions.HTTPUnauthorized)
- self.assertEqual(v.detail, 'You arent authorized')
- headers = v.headers
- self.assertEqual(len(headers), 1)
- headername, headervalue = headers[0]
- self.assertEqual(headername, 'WWW-Authenticate')
- self.assertEqual(headervalue, 'Basic realm="Zope"')
-
- def test_handle_exception_redirect_string_value_correct(self):
- helper = self._makeOne()
- exc_info = ('Redirect', 'http://example.com', None)
- try:
- helper.handle_exception(exc_info)
- except:
- t, v = sys.exc_info()[:2]
- self.assertEqual(t, httpexceptions.HTTPFound)
- headers = v.headers
- self.assertEqual(len(headers), 1)
- headername, header = headers[0]
- self.assertEqual(headername, 'Location')
- self.assertEqual(header, 'http://example.com')
+ status, headers, body = helper.handle_exception(exc_info)
+ self.assertEqual(status, 401)
+ self.assertEqual(len(headers), 2)
+ self.assertEqual(headers[0], ('Content-Length', '20'))
+ self.assertEqual(headers[1],
+ ('Content-Type', 'text/plain; charset=utf-8'))
+ self.assertEqual(body, ['You arent authorized'])
def test_handle_exception_redirect_string(self):
helper = self._makeOne()
@@ -1224,6 +1162,8 @@
headername, header = headers[0]
self.assertEqual(headername, 'Location')
self.assertEqual(header, 'http://example.com')
+ else:
+ raise AssertionError('no redirect')
def test_handle_exception_redirect_string_value_correct(self):
helper = self._makeOne()
@@ -1238,6 +1178,8 @@
headername, header = headers[0]
self.assertEqual(headername, 'Location')
self.assertEqual(header, 'http://example.com')
+ else:
+ raise AssertionError('no redirect')
def test__del__tm_not_active(self):
helper = self._makeOne()
Modified: repoze.zope2/trunk/repoze/zope2/z2bob.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/z2bob.py (original)
+++ repoze.zope2/trunk/repoze/zope2/z2bob.py Wed Jun 11 18:53:21 2008
@@ -22,11 +22,9 @@
import urlparse
import urllib
import xmlrpclib
-import simplejson
from paste import httpexceptions
from paste import httpheaders
-from paste.util import mimeparse
from zope.component import queryMultiAdapter
from zope.interface import Interface
@@ -89,6 +87,7 @@
from repoze.zope2.publishtraverse import DefaultPublishTraverse
from repoze.zope2.publishtraverse import _BETTER_THAN_210
from repoze.zope2.request import makeRequest
+from repoze.zope2.request import convertResponseCode
from repoze.zope2.db import getDB
_FALSETYPES = (None, 0, False, '', u'')
@@ -156,7 +155,6 @@
self.traversed = []
self.default_page = 'index_html'
self.vroot_stack = None
- self.json = False
def _configure(self, config):
self._config = config
@@ -190,14 +188,6 @@
newInteraction() # analogue of Publish.publish
request.processInputs() # analogue of Publish.publish
request_method = request.get('REQUEST_METHOD', 'GET').upper()
- accept = request.get('HTTP_ACCEPT')
-
- if accept:
- parse = mimeparse.parse_media_range
- accept = map(None, [ x.strip() for x in accept.split(',') ])
- if accept:
- accepted_types = [ parse(x)[:2] for x in accept ]
- self.json = ('application', 'json') in accepted_types
response = request.response
isxmlrpc = isinstance(response, XMLRPCResponse)
@@ -449,9 +439,10 @@
response.stdout.seek(0)
result = response.stdout
- status = response.headers.get('status', None)
+ status = response.headers.get('status')
if status is None:
- status = '200 OK'
+ code, reason = convertResponseCode(response.status)
+ status = '%d %s' % (code, reason)
else:
del response.headers['status']
@@ -465,10 +456,6 @@
allow_none=True)
response.setHeader('content-type', 'text/xml')
- elif self.json:
- result = simplejson.dumps(result)
- response.setHeader('content-type', 'application/json')
-
if result is None:
result = ''
@@ -529,13 +516,16 @@
try:
if ((t == 'Unauthorized') or
(inspect.isclass(t) and issubclass(t, _UNAUTH_CLASSES))):
- head = httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' %
- self.realm)
- raise httpexceptions.HTTPUnauthorized(detail=str(v),
- headers=head)
+ response = self.request.RESPONSE
+ response._unauthorized()
+ response.setStatus(401)
+ val = str(v)
+ return self.map_result(val)
elif ((t == 'Redirect') or
(inspect.isclass(t) and issubclass(t, Redirect))):
raise httpexceptions.HTTPFound(headers=[('Location', str(v))])
+ else:
+ raise t, v, tb
finally:
del tb # no memory leak
Modified: repoze.zope2/trunk/setup.py
==============================================================================
--- repoze.zope2/trunk/setup.py (original)
+++ repoze.zope2/trunk/setup.py Wed Jun 11 18:53:21 2008
@@ -53,7 +53,6 @@
'repoze.vhm',
'repoze.retry',
'repoze.obob',
- 'simplejson',
],
install_requires=[
'PasteScript',
@@ -64,7 +63,6 @@
'repoze.retry',
'repoze.vhm',
'repoze.errorlog',
- 'simplejson',
],
test_suite="repoze.zope2.tests",
entry_points = """\
More information about the Repoze-checkins
mailing list