swift/test/unit/proxy/controllers/test_info.py
Samuel Merritt 736cf54adf Remove simplejson from tests
Since we're dropping Python 2.6 support, we can rely on stdlib's json
and get rid of our dependency on simplejson.

This commit just takes simplejson out of the unit and functional
tests. They still pass.

Change-Id: I96f17df81fa5d265395a938b19213d2638682106
2015-05-28 16:05:57 -07:00

294 lines
12 KiB
Python

# Copyright (c) 2010-2012 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import unittest
import time
from mock import Mock
from swift.proxy.controllers import InfoController
from swift.proxy.server import Application as ProxyApp
from swift.common import utils
from swift.common.swob import Request, HTTPException
class TestInfoController(unittest.TestCase):
def setUp(self):
utils._swift_info = {}
utils._swift_admin_info = {}
def get_controller(self, expose_info=None, disallowed_sections=None,
admin_key=None):
disallowed_sections = disallowed_sections or []
app = Mock(spec=ProxyApp)
return InfoController(app, None, expose_info,
disallowed_sections, admin_key)
def start_response(self, status, headers):
self.got_statuses.append(status)
for h in headers:
self.got_headers.append({h[0]: h[1]})
def test_disabled_info(self):
controller = self.get_controller(expose_info=False)
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('403 Forbidden', str(resp))
def test_get_info(self):
controller = self.get_controller(expose_info=True)
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
info = json.loads(resp.body)
self.assertTrue('admin' not in info)
self.assertTrue('foo' in info)
self.assertTrue('bar' in info['foo'])
self.assertEqual(info['foo']['bar'], 'baz')
def test_options_info(self):
controller = self.get_controller(expose_info=True)
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'GET'})
resp = controller.OPTIONS(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
self.assertTrue('Allow' in resp.headers)
def test_get_info_cors(self):
controller = self.get_controller(expose_info=True)
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'GET'},
headers={'Origin': 'http://example.com'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
info = json.loads(resp.body)
self.assertTrue('admin' not in info)
self.assertTrue('foo' in info)
self.assertTrue('bar' in info['foo'])
self.assertEqual(info['foo']['bar'], 'baz')
self.assertTrue('Access-Control-Allow-Origin' in resp.headers)
self.assertTrue('Access-Control-Expose-Headers' in resp.headers)
def test_head_info(self):
controller = self.get_controller(expose_info=True)
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'HEAD'})
resp = controller.HEAD(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
def test_disallow_info(self):
controller = self.get_controller(expose_info=True,
disallowed_sections=['foo2'])
utils._swift_info = {'foo': {'bar': 'baz'},
'foo2': {'bar2': 'baz2'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
req = Request.blank(
'/info', environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
info = json.loads(resp.body)
self.assertTrue('foo' in info)
self.assertTrue('bar' in info['foo'])
self.assertEqual(info['foo']['bar'], 'baz')
self.assertTrue('foo2' not in info)
def test_disabled_admin_info(self):
controller = self.get_controller(expose_info=True, admin_key='')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, '')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('403 Forbidden', str(resp))
def test_get_admin_info(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
info = json.loads(resp.body)
self.assertTrue('admin' in info)
self.assertTrue('qux' in info['admin'])
self.assertTrue('quux' in info['admin']['qux'])
self.assertEqual(info['admin']['qux']['quux'], 'corge')
def test_head_admin_info(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'HEAD'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
expires = int(time.time() + 86400)
sig = utils.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'HEAD'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
def test_get_admin_info_invalid_method(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('HEAD', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('401 Unauthorized', str(resp))
def test_get_admin_info_invalid_expires(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = 1
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('401 Unauthorized', str(resp))
expires = 'abc'
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('401 Unauthorized', str(resp))
def test_get_admin_info_invalid_path(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/foo', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('401 Unauthorized', str(resp))
def test_get_admin_info_invalid_key(self):
controller = self.get_controller(expose_info=True,
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/foo', expires, 'invalid-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('401 Unauthorized', str(resp))
def test_admin_disallow_info(self):
controller = self.get_controller(expose_info=True,
disallowed_sections=['foo2'],
admin_key='secret-admin-key')
utils._swift_info = {'foo': {'bar': 'baz'},
'foo2': {'bar2': 'baz2'}}
utils._swift_admin_info = {'qux': {'quux': 'corge'}}
expires = int(time.time() + 86400)
sig = utils.get_hmac('GET', '/info', expires, 'secret-admin-key')
path = '/info?swiftinfo_sig={sig}&swiftinfo_expires={expires}'.format(
sig=sig, expires=expires)
req = Request.blank(
path, environ={'REQUEST_METHOD': 'GET'})
resp = controller.GET(req)
self.assertTrue(isinstance(resp, HTTPException))
self.assertEqual('200 OK', str(resp))
info = json.loads(resp.body)
self.assertTrue('foo2' not in info)
self.assertTrue('admin' in info)
self.assertTrue('disallowed_sections' in info['admin'])
self.assertTrue('foo2' in info['admin']['disallowed_sections'])
self.assertTrue('qux' in info['admin'])
self.assertTrue('quux' in info['admin']['qux'])
self.assertEqual(info['admin']['qux']['quux'], 'corge')
if __name__ == '__main__':
unittest.main()