openstack/swift
Code Issues Proposed changes

Merge "Metadata retrieving from sqlite must be str for key"

Browse Source
This commit is contained in:
Jenkins 2013-05-16 12:37:33 +00:00 committed by Gerrit Code Review
commit d85e090a14
3 changed files with 89 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
swift/common/db.py
View File

@ -49,6 +49,14 @@ def utf8encode(*args):
return [(s.encode('utf8') if isinstance(s, unicode) else s) for s in args]
def utf8encodekeys(metadata):
uni_keys = [k for k in metadata.keys() if isinstance(k, unicode)]
for k in uni_keys:
sv = metadata[k]
del metadata[k]
metadata[k.encode('utf-8')] = sv
class DatabaseConnectionError(sqlite3.DatabaseError):
"""More friendly error messages for DB Errors."""
@ -550,6 +558,7 @@ class DatabaseBroker(object):
metadata = ''
if metadata:
metadata = json.loads(metadata)
utf8encodekeys(metadata)
else:
metadata = {}
return metadata
@ -575,6 +584,7 @@ class DatabaseBroker(object):
md = conn.execute('SELECT metadata FROM %s_stat' %
self.db_type).fetchone()[0]
md = md and json.loads(md) or {}
utf8encodekeys(md)
except sqlite3.OperationalError, err:
if 'no such column: metadata' not in str(err):
raise

39
test/functionalnosetests/test_account.py
View File

@ -62,6 +62,45 @@ class TestAccount(unittest.TestCase):
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader('x-account-meta-test'), 'Value')
def test_unicode_metadata(self):
if skip:
raise SkipTest
def post(url, token, parsed, conn, name, value):
conn.request('POST', parsed.path, '',
{'X-Auth-Token': token, name: value})
return check_response(conn)
def head(url, token, parsed, conn):
conn.request('HEAD', parsed.path, '', {'X-Auth-Token': token})
return check_response(conn)
uni_key = u'X-Account-Meta-uni\u0E12'
uni_value = u'uni\u0E12'
resp = retry(post, uni_key, '1')
resp.read()
self.assertTrue(resp.status in (201, 204))
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader(uni_key.encode('utf-8')), '1')
resp = retry(post, 'X-Account-Meta-uni', uni_value)
resp.read()
self.assertEquals(resp.status, 204)
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader('X-Account-Meta-uni'),
uni_value.encode('utf-8'))
resp = retry(post, uni_key, uni_value)
resp.read()
self.assertEquals(resp.status, 204)
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader(uni_key.encode('utf-8')),
uni_value.encode('utf-8'))
def test_multi_metadata(self):
if skip:
raise SkipTest

40
test/functionalnosetests/test_container.py
View File

@ -99,6 +99,46 @@ class TestContainer(unittest.TestCase):
self.assertEquals(resp.getheader('x-container-meta-one'), '1')
self.assertEquals(resp.getheader('x-container-meta-two'), '2')
def test_unicode_metadata(self):
if skip:
raise SkipTest
def post(url, token, parsed, conn, name, value):
conn.request('POST', parsed.path + '/' + self.name, '',
{'X-Auth-Token': token, name: value})
return check_response(conn)
def head(url, token, parsed, conn):
conn.request('HEAD', parsed.path + '/' + self.name, '',
{'X-Auth-Token': token})
return check_response(conn)
uni_key = u'X-Container-Meta-uni\u0E12'
uni_value = u'uni\u0E12'
resp = retry(post, uni_key, '1')
resp.read()
self.assertEquals(resp.status, 204)
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader(uni_key.encode('utf-8')), '1')
resp = retry(post, 'X-Container-Meta-uni', uni_value)
resp.read()
self.assertEquals(resp.status, 204)
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader('X-Container-Meta-uni'),
uni_value.encode('utf-8'))
resp = retry(post, uni_key, uni_value)
resp.read()
self.assertEquals(resp.status, 204)
resp = retry(head)
resp.read()
self.assert_(resp.status in (200, 204), resp.status)
self.assertEquals(resp.getheader(uni_key.encode('utf-8')),
uni_value.encode('utf-8'))
def test_PUT_metadata(self):
if skip:
raise SkipTest