py3: port symlink middleware
This patch ports the symlink middleware to py3. The middleware itself seems to be mostly fine and most changes are in the symlink unit tests. Change-Id: I973c2e1bb8969cf6bffece8ce68881c393efbaef
This commit is contained in:

committed by
Tim Burke

parent
2be7dcd850
commit
4f9595f113
@@ -158,16 +158,16 @@ configuration steps are required:
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
from cgi import parse_header
|
from cgi import parse_header
|
||||||
from six.moves.urllib.parse import unquote
|
|
||||||
|
|
||||||
from swift.common.utils import get_logger, register_swift_info, split_path, \
|
from swift.common.utils import get_logger, register_swift_info, split_path, \
|
||||||
MD5_OF_EMPTY_STRING, closing_if_possible, quote
|
MD5_OF_EMPTY_STRING, closing_if_possible
|
||||||
from swift.common.constraints import check_account_format
|
from swift.common.constraints import check_account_format
|
||||||
from swift.common.wsgi import WSGIContext, make_subrequest
|
from swift.common.wsgi import WSGIContext, make_subrequest
|
||||||
from swift.common.request_helpers import get_sys_meta_prefix, \
|
from swift.common.request_helpers import get_sys_meta_prefix, \
|
||||||
check_path_header
|
check_path_header
|
||||||
from swift.common.swob import Request, HTTPBadRequest, HTTPTemporaryRedirect, \
|
from swift.common.swob import Request, HTTPBadRequest, HTTPTemporaryRedirect, \
|
||||||
HTTPException, HTTPConflict, HTTPPreconditionFailed
|
HTTPException, HTTPConflict, HTTPPreconditionFailed, wsgi_quote, \
|
||||||
|
wsgi_unquote
|
||||||
from swift.common.http import is_success
|
from swift.common.http import is_success
|
||||||
from swift.common.exceptions import LinkIterError
|
from swift.common.exceptions import LinkIterError
|
||||||
from swift.common.header_key_dict import HeaderKeyDict
|
from swift.common.header_key_dict import HeaderKeyDict
|
||||||
@@ -197,29 +197,39 @@ def _check_symlink_header(req):
|
|||||||
# copy middleware may accept the format. In the symlink, API
|
# copy middleware may accept the format. In the symlink, API
|
||||||
# says apparently to use "container/object" format so add the
|
# says apparently to use "container/object" format so add the
|
||||||
# validation first, here.
|
# validation first, here.
|
||||||
if unquote(req.headers[TGT_OBJ_SYMLINK_HDR]).startswith('/'):
|
error_body = 'X-Symlink-Target header must be of the form ' \
|
||||||
|
'<container name>/<object name>'
|
||||||
|
try:
|
||||||
|
if wsgi_unquote(req.headers[TGT_OBJ_SYMLINK_HDR]).startswith('/'):
|
||||||
|
raise HTTPPreconditionFailed(
|
||||||
|
body=error_body,
|
||||||
|
request=req, content_type='text/plain')
|
||||||
|
except TypeError:
|
||||||
raise HTTPPreconditionFailed(
|
raise HTTPPreconditionFailed(
|
||||||
body='X-Symlink-Target header must be of the '
|
body=error_body,
|
||||||
'form <container name>/<object name>',
|
|
||||||
request=req, content_type='text/plain')
|
request=req, content_type='text/plain')
|
||||||
|
|
||||||
# check container and object format
|
# check container and object format
|
||||||
container, obj = check_path_header(
|
container, obj = check_path_header(
|
||||||
req, TGT_OBJ_SYMLINK_HDR, 2,
|
req, TGT_OBJ_SYMLINK_HDR, 2,
|
||||||
'X-Symlink-Target header must be of the '
|
error_body)
|
||||||
'form <container name>/<object name>')
|
req.headers[TGT_OBJ_SYMLINK_HDR] = wsgi_quote('%s/%s' % (container, obj))
|
||||||
req.headers[TGT_OBJ_SYMLINK_HDR] = quote('%s/%s' % (container, obj))
|
|
||||||
|
|
||||||
# Check account format if it exists
|
# Check account format if it exists
|
||||||
account = check_account_format(
|
try:
|
||||||
req, unquote(req.headers[TGT_ACCT_SYMLINK_HDR])) \
|
account = check_account_format(
|
||||||
if TGT_ACCT_SYMLINK_HDR in req.headers else None
|
req, wsgi_unquote(req.headers[TGT_ACCT_SYMLINK_HDR])) \
|
||||||
|
if TGT_ACCT_SYMLINK_HDR in req.headers else None
|
||||||
|
except TypeError:
|
||||||
|
raise HTTPPreconditionFailed(
|
||||||
|
body='Account name cannot contain slashes',
|
||||||
|
request=req, content_type='text/plain')
|
||||||
|
|
||||||
# Extract request path
|
# Extract request path
|
||||||
_junk, req_acc, req_cont, req_obj = req.split_path(4, 4, True)
|
_junk, req_acc, req_cont, req_obj = req.split_path(4, 4, True)
|
||||||
|
|
||||||
if account:
|
if account:
|
||||||
req.headers[TGT_ACCT_SYMLINK_HDR] = quote(account)
|
req.headers[TGT_ACCT_SYMLINK_HDR] = wsgi_quote(account)
|
||||||
else:
|
else:
|
||||||
account = req_acc
|
account = req_acc
|
||||||
|
|
||||||
@@ -383,7 +393,7 @@ class SymlinkObjectContext(WSGIContext):
|
|||||||
"""
|
"""
|
||||||
version, account, _junk = req.split_path(2, 3, True)
|
version, account, _junk = req.split_path(2, 3, True)
|
||||||
account = self._response_header_value(
|
account = self._response_header_value(
|
||||||
TGT_ACCT_SYSMETA_SYMLINK_HDR) or quote(account)
|
TGT_ACCT_SYSMETA_SYMLINK_HDR) or wsgi_quote(account)
|
||||||
target_path = os.path.join(
|
target_path = os.path.join(
|
||||||
'/', version, account,
|
'/', version, account,
|
||||||
symlink_target.lstrip('/'))
|
symlink_target.lstrip('/'))
|
||||||
@@ -488,7 +498,7 @@ class SymlinkObjectContext(WSGIContext):
|
|||||||
if tgt_co:
|
if tgt_co:
|
||||||
version, account, _junk = req.split_path(2, 3, True)
|
version, account, _junk = req.split_path(2, 3, True)
|
||||||
target_acc = self._response_header_value(
|
target_acc = self._response_header_value(
|
||||||
TGT_ACCT_SYSMETA_SYMLINK_HDR) or quote(account)
|
TGT_ACCT_SYSMETA_SYMLINK_HDR) or wsgi_quote(account)
|
||||||
location_hdr = os.path.join(
|
location_hdr = os.path.join(
|
||||||
'/', version, target_acc, tgt_co)
|
'/', version, target_acc, tgt_co)
|
||||||
req.environ['swift.leave_relative_location'] = True
|
req.environ['swift.leave_relative_location'] = True
|
||||||
|
@@ -18,7 +18,7 @@ import unittest
|
|||||||
import json
|
import json
|
||||||
import mock
|
import mock
|
||||||
|
|
||||||
from six.moves.urllib.parse import quote, parse_qs
|
from six.moves.urllib.parse import parse_qs
|
||||||
from swift.common import swob
|
from swift.common import swob
|
||||||
from swift.common.middleware import symlink, copy, versioned_writes, \
|
from swift.common.middleware import symlink, copy, versioned_writes, \
|
||||||
listing_formats
|
listing_formats
|
||||||
@@ -56,7 +56,7 @@ class TestSymlinkMiddlewareBase(unittest.TestCase):
|
|||||||
headers[0] = h
|
headers[0] = h
|
||||||
|
|
||||||
body_iter = app(req.environ, start_response)
|
body_iter = app(req.environ, start_response)
|
||||||
body = ''
|
body = b''
|
||||||
caught_exc = None
|
caught_exc = None
|
||||||
try:
|
try:
|
||||||
for chunk in body_iter:
|
for chunk in body_iter:
|
||||||
@@ -112,15 +112,15 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
body='')
|
body='')
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '412 Precondition Failed')
|
self.assertEqual(status, '412 Precondition Failed')
|
||||||
self.assertEqual(body, "X-Symlink-Target header must be of "
|
self.assertEqual(body, b"X-Symlink-Target header must be of "
|
||||||
"the form <container name>/<object name>")
|
b"the form <container name>/<object name>")
|
||||||
|
|
||||||
def test_symlink_put_non_zero_length(self):
|
def test_symlink_put_non_zero_length(self):
|
||||||
req = Request.blank('/v1/a/c/symlink', method='PUT', body='req_body',
|
req = Request.blank('/v1/a/c/symlink', method='PUT', body='req_body',
|
||||||
headers={'X-Symlink-Target': 'c1/o'})
|
headers={'X-Symlink-Target': 'c1/o'})
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '400 Bad Request')
|
self.assertEqual(status, '400 Bad Request')
|
||||||
self.assertEqual(body, 'Symlink requests require a zero byte body')
|
self.assertEqual(body, b'Symlink requests require a zero byte body')
|
||||||
|
|
||||||
def test_symlink_put_bad_object_header(self):
|
def test_symlink_put_bad_object_header(self):
|
||||||
req = Request.blank('/v1/a/c/symlink', method='PUT',
|
req = Request.blank('/v1/a/c/symlink', method='PUT',
|
||||||
@@ -128,8 +128,8 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
body='')
|
body='')
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, "412 Precondition Failed")
|
self.assertEqual(status, "412 Precondition Failed")
|
||||||
self.assertEqual(body, "X-Symlink-Target header must be of "
|
self.assertEqual(body, b"X-Symlink-Target header must be of "
|
||||||
"the form <container name>/<object name>")
|
b"the form <container name>/<object name>")
|
||||||
|
|
||||||
def test_symlink_put_bad_account_header(self):
|
def test_symlink_put_bad_account_header(self):
|
||||||
req = Request.blank('/v1/a/c/symlink', method='PUT',
|
req = Request.blank('/v1/a/c/symlink', method='PUT',
|
||||||
@@ -138,7 +138,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
body='')
|
body='')
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, "412 Precondition Failed")
|
self.assertEqual(status, "412 Precondition Failed")
|
||||||
self.assertEqual(body, "Account name cannot contain slashes")
|
self.assertEqual(body, b"Account name cannot contain slashes")
|
||||||
|
|
||||||
def test_get_symlink(self):
|
def test_get_symlink(self):
|
||||||
self.app.register('GET', '/v1/a/c/symlink', swob.HTTPOk,
|
self.app.register('GET', '/v1/a/c/symlink', swob.HTTPOk,
|
||||||
@@ -176,7 +176,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
headers=req_headers)
|
headers=req_headers)
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '200 OK')
|
self.assertEqual(status, '200 OK')
|
||||||
self.assertEqual(body, 'resp_body')
|
self.assertEqual(body, b'resp_body')
|
||||||
self.assertNotIn('X-Symlink-Target', dict(headers))
|
self.assertNotIn('X-Symlink-Target', dict(headers))
|
||||||
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
||||||
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
||||||
@@ -195,7 +195,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
req = Request.blank('/v1/a/c/symlink', method='GET')
|
req = Request.blank('/v1/a/c/symlink', method='GET')
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '404 Not Found')
|
self.assertEqual(status, '404 Not Found')
|
||||||
self.assertEqual(body, '')
|
self.assertEqual(body, b'')
|
||||||
self.assertNotIn('X-Symlink-Target', dict(headers))
|
self.assertNotIn('X-Symlink-Target', dict(headers))
|
||||||
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
||||||
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
||||||
@@ -211,8 +211,8 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '416 Requested Range Not Satisfiable')
|
self.assertEqual(status, '416 Requested Range Not Satisfiable')
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
body, '<html><h1>Requested Range Not Satisfiable</h1>'
|
body, b'<html><h1>Requested Range Not Satisfiable</h1>'
|
||||||
'<p>The Range requested is not available.</p></html>')
|
b'<p>The Range requested is not available.</p></html>')
|
||||||
self.assertNotIn('X-Symlink-Target', dict(headers))
|
self.assertNotIn('X-Symlink-Target', dict(headers))
|
||||||
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
||||||
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
||||||
@@ -228,7 +228,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
headers={'Range': 'bytes=1-2'})
|
headers={'Range': 'bytes=1-2'})
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '200 OK')
|
self.assertEqual(status, '200 OK')
|
||||||
self.assertEqual(body, 'es')
|
self.assertEqual(body, b'es')
|
||||||
self.assertNotIn('X-Symlink-Target', dict(headers))
|
self.assertNotIn('X-Symlink-Target', dict(headers))
|
||||||
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
self.assertNotIn('X-Symlink-Target-Account', dict(headers))
|
||||||
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
self.assertIn(('Content-Location', '/v1/a2/c1/o'), headers)
|
||||||
@@ -241,7 +241,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
|
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '200 OK')
|
self.assertEqual(status, '200 OK')
|
||||||
self.assertEqual(body, 'resp_body')
|
self.assertEqual(body, b'resp_body')
|
||||||
|
|
||||||
# Assert special headers for symlink are not in response
|
# Assert special headers for symlink are not in response
|
||||||
self.assertNotIn('X-Symlink-Target', dict(headers))
|
self.assertNotIn('X-Symlink-Target', dict(headers))
|
||||||
@@ -359,9 +359,10 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
headers={'X-Object-Meta-Color': 'Red'})
|
headers={'X-Object-Meta-Color': 'Red'})
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '307 Temporary Redirect')
|
self.assertEqual(status, '307 Temporary Redirect')
|
||||||
self.assertEqual(body,
|
self.assertEqual(
|
||||||
'The requested POST was applied to a symlink. POST '
|
body,
|
||||||
'directly to the target to apply requested metadata.')
|
b'The requested POST was applied to a symlink. POST '
|
||||||
|
b'directly to the target to apply requested metadata.')
|
||||||
method, path, hdrs = self.app.calls_with_headers[0]
|
method, path, hdrs = self.app.calls_with_headers[0]
|
||||||
val = hdrs.get('X-Object-Meta-Color')
|
val = hdrs.get('X-Object-Meta-Color')
|
||||||
self.assertEqual(val, 'Red')
|
self.assertEqual(val, 'Red')
|
||||||
@@ -379,8 +380,8 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
headers={'X-Symlink-Target': 'c1/regular_obj'})
|
headers={'X-Symlink-Target': 'c1/regular_obj'})
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(status, '400 Bad Request')
|
self.assertEqual(status, '400 Bad Request')
|
||||||
self.assertEqual(body, "A PUT request is required to set a symlink "
|
self.assertEqual(body, b"A PUT request is required to set a symlink "
|
||||||
"target")
|
b"target")
|
||||||
|
|
||||||
def test_symlink_post_but_fail_at_server(self):
|
def test_symlink_post_but_fail_at_server(self):
|
||||||
self.app.register('POST', '/v1/a/c/o', swob.HTTPNotFound, {})
|
self.app.register('POST', '/v1/a/c/o', swob.HTTPNotFound, {})
|
||||||
@@ -405,16 +406,15 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
# URL encoded is safe
|
# URL encoded is safe
|
||||||
do_test({'X-Symlink-Target': 'c1%2Fo1'})
|
do_test({'X-Symlink-Target': 'c1%2Fo1'})
|
||||||
# URL encoded + multibytes is also safe
|
# URL encoded + multibytes is also safe
|
||||||
do_test(
|
|
||||||
{'X-Symlink-Target':
|
|
||||||
u'\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'})
|
|
||||||
target = u'\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
target = u'\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
||||||
encoded_target = quote(target.encode('utf-8'), '')
|
target = swob.bytes_to_wsgi(target.encode('utf8'))
|
||||||
do_test({'X-Symlink-Target': encoded_target})
|
do_test({'X-Symlink-Target': target})
|
||||||
|
do_test({'X-Symlink-Target': swob.wsgi_quote(target)})
|
||||||
|
|
||||||
|
target = swob.bytes_to_wsgi(u'\u30b0\u30e9\u30d6\u30eb'.encode('utf8'))
|
||||||
do_test(
|
do_test(
|
||||||
{'X-Symlink-Target': 'cont/obj',
|
{'X-Symlink-Target': 'cont/obj',
|
||||||
'X-Symlink-Target-Account': u'\u30b0\u30e9\u30d6\u30eb'})
|
'X-Symlink-Target-Account': target})
|
||||||
|
|
||||||
def test_check_symlink_header_invalid_format(self):
|
def test_check_symlink_header_invalid_format(self):
|
||||||
def do_test(headers, status, err_msg):
|
def do_test(headers, status, err_msg):
|
||||||
@@ -428,54 +428,59 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
|
|
||||||
do_test({'X-Symlink-Target': '/c1/o1'},
|
do_test({'X-Symlink-Target': '/c1/o1'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'X-Symlink-Target header must be of the '
|
b'X-Symlink-Target header must be of the '
|
||||||
'form <container name>/<object name>')
|
b'form <container name>/<object name>')
|
||||||
do_test({'X-Symlink-Target': 'c1o1'},
|
do_test({'X-Symlink-Target': 'c1o1'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'X-Symlink-Target header must be of the '
|
b'X-Symlink-Target header must be of the '
|
||||||
'form <container name>/<object name>')
|
b'form <container name>/<object name>')
|
||||||
do_test({'X-Symlink-Target': 'c1/o1',
|
do_test({'X-Symlink-Target': 'c1/o1',
|
||||||
'X-Symlink-Target-Account': '/another'},
|
'X-Symlink-Target-Account': '/another'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'Account name cannot contain slashes')
|
b'Account name cannot contain slashes')
|
||||||
do_test({'X-Symlink-Target': 'c1/o1',
|
do_test({'X-Symlink-Target': 'c1/o1',
|
||||||
'X-Symlink-Target-Account': 'an/other'},
|
'X-Symlink-Target-Account': 'an/other'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'Account name cannot contain slashes')
|
b'Account name cannot contain slashes')
|
||||||
# url encoded case
|
# url encoded case
|
||||||
do_test({'X-Symlink-Target': '%2Fc1%2Fo1'},
|
do_test({'X-Symlink-Target': '%2Fc1%2Fo1'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'X-Symlink-Target header must be of the '
|
b'X-Symlink-Target header must be of the '
|
||||||
'form <container name>/<object name>')
|
b'form <container name>/<object name>')
|
||||||
do_test({'X-Symlink-Target': 'c1/o1',
|
do_test({'X-Symlink-Target': 'c1/o1',
|
||||||
'X-Symlink-Target-Account': '%2Fanother'},
|
'X-Symlink-Target-Account': '%2Fanother'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'Account name cannot contain slashes')
|
b'Account name cannot contain slashes')
|
||||||
do_test({'X-Symlink-Target': 'c1/o1',
|
do_test({'X-Symlink-Target': 'c1/o1',
|
||||||
'X-Symlink-Target-Account': 'an%2Fother'},
|
'X-Symlink-Target-Account': 'an%2Fother'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'Account name cannot contain slashes')
|
b'Account name cannot contain slashes')
|
||||||
# with multi-bytes
|
# with multi-bytes
|
||||||
do_test(
|
do_test(
|
||||||
{'X-Symlink-Target':
|
{'X-Symlink-Target':
|
||||||
u'/\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'},
|
u'/\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'X-Symlink-Target header must be of the '
|
b'X-Symlink-Target header must be of the '
|
||||||
'form <container name>/<object name>')
|
b'form <container name>/<object name>')
|
||||||
target = u'/\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
target = u'/\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
||||||
encoded_target = quote(target.encode('utf-8'), '')
|
target = swob.bytes_to_wsgi(target.encode('utf8'))
|
||||||
do_test(
|
do_test(
|
||||||
{'X-Symlink-Target': encoded_target},
|
{'X-Symlink-Target': swob.wsgi_quote(target)},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'X-Symlink-Target header must be of the '
|
b'X-Symlink-Target header must be of the '
|
||||||
'form <container name>/<object name>')
|
b'form <container name>/<object name>')
|
||||||
account = u'\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
account = u'\u30b0\u30e9\u30d6\u30eb/\u30a2\u30ba\u30ec\u30f3'
|
||||||
encoded_account = quote(account.encode('utf-8'), '')
|
|
||||||
do_test(
|
do_test(
|
||||||
{'X-Symlink-Target': 'c/o',
|
{'X-Symlink-Target': 'c/o',
|
||||||
'X-Symlink-Target-Account': encoded_account},
|
'X-Symlink-Target-Account': account},
|
||||||
'412 Precondition Failed',
|
'412 Precondition Failed',
|
||||||
'Account name cannot contain slashes')
|
b'Account name cannot contain slashes')
|
||||||
|
account = swob.bytes_to_wsgi(account.encode('utf8'))
|
||||||
|
do_test(
|
||||||
|
{'X-Symlink-Target': 'c/o',
|
||||||
|
'X-Symlink-Target-Account': swob.wsgi_quote(account)},
|
||||||
|
'412 Precondition Failed',
|
||||||
|
b'Account name cannot contain slashes')
|
||||||
|
|
||||||
def test_check_symlink_header_points_to_itself(self):
|
def test_check_symlink_header_points_to_itself(self):
|
||||||
req = Request.blank('/v1/a/c/o', method='PUT',
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
||||||
@@ -483,7 +488,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
with self.assertRaises(swob.HTTPException) as cm:
|
with self.assertRaises(swob.HTTPException) as cm:
|
||||||
symlink._check_symlink_header(req)
|
symlink._check_symlink_header(req)
|
||||||
self.assertEqual(cm.exception.status, '400 Bad Request')
|
self.assertEqual(cm.exception.status, '400 Bad Request')
|
||||||
self.assertEqual(cm.exception.body, 'Symlink cannot target itself')
|
self.assertEqual(cm.exception.body, b'Symlink cannot target itself')
|
||||||
|
|
||||||
# Even if set account to itself, it will fail as well
|
# Even if set account to itself, it will fail as well
|
||||||
req = Request.blank('/v1/a/c/o', method='PUT',
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
||||||
@@ -492,7 +497,7 @@ class TestSymlinkMiddleware(TestSymlinkMiddlewareBase):
|
|||||||
with self.assertRaises(swob.HTTPException) as cm:
|
with self.assertRaises(swob.HTTPException) as cm:
|
||||||
symlink._check_symlink_header(req)
|
symlink._check_symlink_header(req)
|
||||||
self.assertEqual(cm.exception.status, '400 Bad Request')
|
self.assertEqual(cm.exception.status, '400 Bad Request')
|
||||||
self.assertEqual(cm.exception.body, 'Symlink cannot target itself')
|
self.assertEqual(cm.exception.body, b'Symlink cannot target itself')
|
||||||
|
|
||||||
# sanity, the case to another account is safe
|
# sanity, the case to another account is safe
|
||||||
req = Request.blank('/v1/a/c/o', method='PUT',
|
req = Request.blank('/v1/a/c/o', method='PUT',
|
||||||
@@ -866,10 +871,10 @@ class TestSymlinkContainerContext(TestSymlinkMiddlewareBase):
|
|||||||
|
|
||||||
def test_no_affect_for_account_request(self):
|
def test_no_affect_for_account_request(self):
|
||||||
with mock.patch.object(self.sym, 'app') as mock_app:
|
with mock.patch.object(self.sym, 'app') as mock_app:
|
||||||
mock_app.return_value = 'ok'
|
mock_app.return_value = (b'ok',)
|
||||||
req = Request.blank(path='/v1/a')
|
req = Request.blank(path='/v1/a')
|
||||||
status, headers, body = self.call_sym(req)
|
status, headers, body = self.call_sym(req)
|
||||||
self.assertEqual(body, 'ok')
|
self.assertEqual(body, b'ok')
|
||||||
|
|
||||||
def test_get_container_simple_with_listing_format(self):
|
def test_get_container_simple_with_listing_format(self):
|
||||||
self.app.register(
|
self.app.register(
|
||||||
@@ -916,14 +921,14 @@ class TestSymlinkContainerContext(TestSymlinkMiddlewareBase):
|
|||||||
req = Request.blank(path='/v1/a/c?format=xml')
|
req = Request.blank(path='/v1/a/c?format=xml')
|
||||||
status, headers, body = self.call_app(req, app=self.lf)
|
status, headers, body = self.call_app(req, app=self.lf)
|
||||||
self.assertEqual(status, '200 OK')
|
self.assertEqual(status, '200 OK')
|
||||||
self.assertEqual(body.split('\n'), [
|
self.assertEqual(body.split(b'\n'), [
|
||||||
'<?xml version="1.0" encoding="UTF-8"?>',
|
b'<?xml version="1.0" encoding="UTF-8"?>',
|
||||||
'<container name="c"><object><name>sym_obj</name>'
|
b'<container name="c"><object><name>sym_obj</name>'
|
||||||
'<hash>etag</hash><bytes>0</bytes>'
|
b'<hash>etag</hash><bytes>0</bytes>'
|
||||||
'<content_type>text/plain</content_type>'
|
b'<content_type>text/plain</content_type>'
|
||||||
'<last_modified>2014-11-21T14:23:02.206740</last_modified>'
|
b'<last_modified>2014-11-21T14:23:02.206740</last_modified>'
|
||||||
'</object>'
|
b'</object>'
|
||||||
'<object><name>normal_obj</name><hash>etag2</hash>'
|
b'<object><name>normal_obj</name><hash>etag2</hash>'
|
||||||
'<bytes>32</bytes><content_type>text/plain</content_type>'
|
b'<bytes>32</bytes><content_type>text/plain</content_type>'
|
||||||
'<last_modified>2014-11-21T14:14:27.409100</last_modified>'
|
b'<last_modified>2014-11-21T14:14:27.409100</last_modified>'
|
||||||
'</object></container>'])
|
b'</object></container>'])
|
||||||
|
1
tox.ini
1
tox.ini
@@ -68,6 +68,7 @@ commands =
|
|||||||
test/unit/common/middleware/test_slo.py \
|
test/unit/common/middleware/test_slo.py \
|
||||||
test/unit/common/middleware/test_subrequest_logging.py \
|
test/unit/common/middleware/test_subrequest_logging.py \
|
||||||
test/unit/common/middleware/test_staticweb.py \
|
test/unit/common/middleware/test_staticweb.py \
|
||||||
|
test/unit/common/middleware/test_symlink.py \
|
||||||
test/unit/common/middleware/test_tempauth.py \
|
test/unit/common/middleware/test_tempauth.py \
|
||||||
test/unit/common/middleware/test_versioned_writes.py \
|
test/unit/common/middleware/test_versioned_writes.py \
|
||||||
test/unit/common/middleware/test_xprofile.py \
|
test/unit/common/middleware/test_xprofile.py \
|
||||||
|
Reference in New Issue
Block a user