diff options
Diffstat (limited to 'test/unit/proxy/controllers')
| -rw-r--r-- | test/unit/proxy/controllers/test_account.py | 102 | ||||
| -rw-r--r-- | test/unit/proxy/controllers/test_base.py | 113 | ||||
| -rw-r--r-- | test/unit/proxy/controllers/test_container.py | 62 | ||||
| -rw-r--r-- | test/unit/proxy/controllers/test_info.py | 293 |
4 files changed, 554 insertions, 16 deletions
diff --git a/test/unit/proxy/controllers/test_account.py b/test/unit/proxy/controllers/test_account.py index 394ada7..eefd57d 100644 --- a/test/unit/proxy/controllers/test_account.py +++ b/test/unit/proxy/controllers/test_account.py @@ -19,7 +19,9 @@ import unittest from swift.common.swob import Request from swift.proxy import server as proxy_server from swift.proxy.controllers.base import headers_to_account_info +from swift.common.constraints import MAX_ACCOUNT_NAME_LENGTH as MAX_ANAME_LEN from test.unit import fake_http_connect, FakeRing, FakeMemcache +from swift.common.request_helpers import get_sys_meta_prefix class TestAccountController(unittest.TestCase): @@ -32,8 +34,8 @@ class TestAccountController(unittest.TestCase): def test_account_info_in_response_env(self): controller = proxy_server.AccountController(self.app, 'AUTH_bob') with mock.patch('swift.proxy.controllers.base.http_connect', - fake_http_connect(200, 200, body='')): - req = Request.blank('/AUTH_bob', {'PATH_INFO': '/AUTH_bob'}) + fake_http_connect(200, body='')): + req = Request.blank('/v1/AUTH_bob', {'PATH_INFO': '/v1/AUTH_bob'}) resp = controller.HEAD(req) self.assertEqual(2, resp.status_int // 100) self.assertTrue('swift.account/AUTH_bob' in resp.environ) @@ -46,22 +48,110 @@ class TestAccountController(unittest.TestCase): 'x-account-meta-temp-url-key-2': 'value'} controller = proxy_server.AccountController(self.app, 'a') - req = Request.blank('/a') + req = Request.blank('/v1/a') with mock.patch('swift.proxy.controllers.base.http_connect', - fake_http_connect(200, 200, headers=owner_headers)): + fake_http_connect(200, headers=owner_headers)): resp = controller.HEAD(req) self.assertEquals(2, resp.status_int // 100) for key in owner_headers: self.assertTrue(key not in resp.headers) - req = Request.blank('/a', environ={'swift_owner': True}) + req = Request.blank('/v1/a', environ={'swift_owner': True}) with mock.patch('swift.proxy.controllers.base.http_connect', - fake_http_connect(200, 200, headers=owner_headers)): + fake_http_connect(200, headers=owner_headers)): resp = controller.HEAD(req) self.assertEquals(2, resp.status_int // 100) for key in owner_headers: self.assertTrue(key in resp.headers) + def test_get_deleted_account(self): + resp_headers = { + 'x-account-status': 'deleted', + } + controller = proxy_server.AccountController(self.app, 'a') + + req = Request.blank('/v1/a') + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(404, headers=resp_headers)): + resp = controller.HEAD(req) + self.assertEquals(410, resp.status_int) + + def test_long_acct_names(self): + long_acct_name = '%sLongAccountName' % ('Very' * (MAX_ANAME_LEN // 4)) + controller = proxy_server.AccountController(self.app, long_acct_name) + + req = Request.blank('/v1/%s' % long_acct_name) + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200)): + resp = controller.HEAD(req) + self.assertEquals(400, resp.status_int) + + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200)): + resp = controller.GET(req) + self.assertEquals(400, resp.status_int) + + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200)): + resp = controller.POST(req) + self.assertEquals(400, resp.status_int) + + def _make_callback_func(self, context): + def callback(ipaddr, port, device, partition, method, path, + headers=None, query_string=None, ssl=False): + context['method'] = method + context['path'] = path + context['headers'] = headers or {} + return callback + + def test_sys_meta_headers_PUT(self): + # check that headers in sys meta namespace make it through + # the proxy controller + sys_meta_key = '%stest' % get_sys_meta_prefix('account') + sys_meta_key = sys_meta_key.title() + user_meta_key = 'X-Account-Meta-Test' + # allow PUTs to account... + self.app.allow_account_management = True + controller = proxy_server.AccountController(self.app, 'a') + context = {} + callback = self._make_callback_func(context) + hdrs_in = {sys_meta_key: 'foo', + user_meta_key: 'bar', + 'x-timestamp': '1.0'} + req = Request.blank('/v1/a', headers=hdrs_in) + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200, 200, give_connect=callback)): + controller.PUT(req) + self.assertEqual(context['method'], 'PUT') + self.assertTrue(sys_meta_key in context['headers']) + self.assertEqual(context['headers'][sys_meta_key], 'foo') + self.assertTrue(user_meta_key in context['headers']) + self.assertEqual(context['headers'][user_meta_key], 'bar') + self.assertNotEqual(context['headers']['x-timestamp'], '1.0') + + def test_sys_meta_headers_POST(self): + # check that headers in sys meta namespace make it through + # the proxy controller + sys_meta_key = '%stest' % get_sys_meta_prefix('account') + sys_meta_key = sys_meta_key.title() + user_meta_key = 'X-Account-Meta-Test' + controller = proxy_server.AccountController(self.app, 'a') + context = {} + callback = self._make_callback_func(context) + hdrs_in = {sys_meta_key: 'foo', + user_meta_key: 'bar', + 'x-timestamp': '1.0'} + req = Request.blank('/v1/a', headers=hdrs_in) + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200, 200, give_connect=callback)): + controller.POST(req) + self.assertEqual(context['method'], 'POST') + self.assertTrue(sys_meta_key in context['headers']) + self.assertEqual(context['headers'][sys_meta_key], 'foo') + self.assertTrue(user_meta_key in context['headers']) + self.assertEqual(context['headers'][user_meta_key], 'bar') + self.assertNotEqual(context['headers']['x-timestamp'], '1.0') + if __name__ == '__main__': unittest.main() diff --git a/test/unit/proxy/controllers/test_base.py b/test/unit/proxy/controllers/test_base.py index 8214b98..0c94f90 100644 --- a/test/unit/proxy/controllers/test_base.py +++ b/test/unit/proxy/controllers/test_base.py @@ -18,11 +18,13 @@ from mock import patch from swift.proxy.controllers.base import headers_to_container_info, \ headers_to_account_info, headers_to_object_info, get_container_info, \ get_container_memcache_key, get_account_info, get_account_memcache_key, \ - get_object_env_key, _get_cache_key, get_info, get_object_info, Controller -from swift.common.swob import Request + get_object_env_key, _get_cache_key, get_info, get_object_info, \ + Controller, GetOrHeadHandler +from swift.common.swob import Request, HTTPException, HeaderKeyDict from swift.common.utils import split_path from test.unit import fake_http_connect, FakeRing, FakeMemcache from swift.proxy import server as proxy_server +from swift.common.request_helpers import get_sys_meta_prefix FakeResponse_status_int = 201 @@ -88,7 +90,7 @@ class TestFuncs(unittest.TestCase): def test_GETorHEAD_base(self): base = Controller(self.app) - req = Request.blank('/a/c/o/with/slashes') + req = Request.blank('/v1/a/c/o/with/slashes') with patch('swift.proxy.controllers.base.' 'http_connect', fake_http_connect(200)): resp = base.GETorHEAD_base(req, 'object', FakeRing(), 'part', @@ -96,14 +98,14 @@ class TestFuncs(unittest.TestCase): self.assertTrue('swift.object/a/c/o/with/slashes' in resp.environ) self.assertEqual( resp.environ['swift.object/a/c/o/with/slashes']['status'], 200) - req = Request.blank('/a/c/o') + req = Request.blank('/v1/a/c/o') with patch('swift.proxy.controllers.base.' 'http_connect', fake_http_connect(200)): resp = base.GETorHEAD_base(req, 'object', FakeRing(), 'part', '/a/c/o') self.assertTrue('swift.object/a/c/o' in resp.environ) self.assertEqual(resp.environ['swift.object/a/c/o']['status'], 200) - req = Request.blank('/a/c') + req = Request.blank('/v1/a/c') with patch('swift.proxy.controllers.base.' 'http_connect', fake_http_connect(200)): resp = base.GETorHEAD_base(req, 'container', FakeRing(), 'part', @@ -111,7 +113,7 @@ class TestFuncs(unittest.TestCase): self.assertTrue('swift.container/a/c' in resp.environ) self.assertEqual(resp.environ['swift.container/a/c']['status'], 200) - req = Request.blank('/a') + req = Request.blank('/v1/a') with patch('swift.proxy.controllers.base.' 'http_connect', fake_http_connect(200)): resp = base.GETorHEAD_base(req, 'account', FakeRing(), 'part', @@ -250,7 +252,9 @@ class TestFuncs(unittest.TestCase): def test_get_container_info_cache(self): cached = {'status': 404, 'bytes': 3333, - 'object_count': 10} + 'object_count': 10, + # simplejson sometimes hands back strings, sometimes unicodes + 'versions': u"\u1F4A9"} req = Request.blank("/v1/account/cont", environ={'swift.cache': FakeCache(cached)}) with patch('swift.proxy.controllers.base.' @@ -259,6 +263,7 @@ class TestFuncs(unittest.TestCase): self.assertEquals(resp['bytes'], 3333) self.assertEquals(resp['object_count'], 10) self.assertEquals(resp['status'], 404) + self.assertEquals(resp['versions'], "\xe1\xbd\x8a\x39") def test_get_container_info_env(self): cache_key = get_container_memcache_key("account", "cont") @@ -361,6 +366,15 @@ class TestFuncs(unittest.TestCase): self.assertEquals(resp['meta']['whatevs'], 14) self.assertEquals(resp['meta']['somethingelse'], 0) + def test_headers_to_container_info_sys_meta(self): + prefix = get_sys_meta_prefix('container') + headers = {'%sWhatevs' % prefix: 14, + '%ssomethingelse' % prefix: 0} + resp = headers_to_container_info(headers.items(), 200) + self.assertEquals(len(resp['sysmeta']), 2) + self.assertEquals(resp['sysmeta']['whatevs'], 14) + self.assertEquals(resp['sysmeta']['somethingelse'], 0) + def test_headers_to_container_info_values(self): headers = { 'x-container-read': 'readvalue', @@ -392,6 +406,15 @@ class TestFuncs(unittest.TestCase): self.assertEquals(resp['meta']['whatevs'], 14) self.assertEquals(resp['meta']['somethingelse'], 0) + def test_headers_to_account_info_sys_meta(self): + prefix = get_sys_meta_prefix('account') + headers = {'%sWhatevs' % prefix: 14, + '%ssomethingelse' % prefix: 0} + resp = headers_to_account_info(headers.items(), 200) + self.assertEquals(len(resp['sysmeta']), 2) + self.assertEquals(resp['sysmeta']['whatevs'], 14) + self.assertEquals(resp['sysmeta']['somethingelse'], 0) + def test_headers_to_account_info_values(self): headers = { 'x-account-object-count': '10', @@ -433,3 +456,79 @@ class TestFuncs(unittest.TestCase): self.assertEquals( resp, headers_to_object_info(headers.items(), 200)) + + def test_have_quorum(self): + base = Controller(self.app) + # just throw a bunch of test cases at it + self.assertEqual(base.have_quorum([201, 404], 3), False) + self.assertEqual(base.have_quorum([201, 201], 4), False) + self.assertEqual(base.have_quorum([201, 201, 404, 404], 4), False) + self.assertEqual(base.have_quorum([201, 503, 503, 201], 4), False) + self.assertEqual(base.have_quorum([201, 201], 3), True) + self.assertEqual(base.have_quorum([404, 404], 3), True) + self.assertEqual(base.have_quorum([201, 201], 2), True) + self.assertEqual(base.have_quorum([404, 404], 2), True) + self.assertEqual(base.have_quorum([201, 404, 201, 201], 4), True) + + def test_range_fast_forward(self): + req = Request.blank('/') + handler = GetOrHeadHandler(None, req, None, None, None, None, {}) + handler.fast_forward(50) + self.assertEquals(handler.backend_headers['Range'], 'bytes=50-') + + handler = GetOrHeadHandler(None, req, None, None, None, None, + {'Range': 'bytes=23-50'}) + handler.fast_forward(20) + self.assertEquals(handler.backend_headers['Range'], 'bytes=43-50') + self.assertRaises(HTTPException, + handler.fast_forward, 80) + + handler = GetOrHeadHandler(None, req, None, None, None, None, + {'Range': 'bytes=23-'}) + handler.fast_forward(20) + self.assertEquals(handler.backend_headers['Range'], 'bytes=43-') + + handler = GetOrHeadHandler(None, req, None, None, None, None, + {'Range': 'bytes=-100'}) + handler.fast_forward(20) + self.assertEquals(handler.backend_headers['Range'], 'bytes=-80') + + def test_transfer_headers_with_sysmeta(self): + base = Controller(self.app) + good_hdrs = {'x-base-sysmeta-foo': 'ok', + 'X-Base-sysmeta-Bar': 'also ok'} + bad_hdrs = {'x-base-sysmeta-': 'too short'} + hdrs = dict(good_hdrs) + hdrs.update(bad_hdrs) + dst_hdrs = HeaderKeyDict() + base.transfer_headers(hdrs, dst_hdrs) + self.assertEqual(HeaderKeyDict(good_hdrs), dst_hdrs) + + def test_generate_request_headers(self): + base = Controller(self.app) + src_headers = {'x-remove-base-meta-owner': 'x', + 'x-base-meta-size': '151M', + 'new-owner': 'Kun'} + req = Request.blank('/v1/a/c/o', headers=src_headers) + dst_headers = base.generate_request_headers(req, transfer=True) + expected_headers = {'x-base-meta-owner': '', + 'x-base-meta-size': '151M'} + for k, v in expected_headers.iteritems(): + self.assertTrue(k in dst_headers) + self.assertEqual(v, dst_headers[k]) + self.assertFalse('new-owner' in dst_headers) + + def test_generate_request_headers_with_sysmeta(self): + base = Controller(self.app) + good_hdrs = {'x-base-sysmeta-foo': 'ok', + 'X-Base-sysmeta-Bar': 'also ok'} + bad_hdrs = {'x-base-sysmeta-': 'too short'} + hdrs = dict(good_hdrs) + hdrs.update(bad_hdrs) + req = Request.blank('/v1/a/c/o', headers=hdrs) + dst_headers = base.generate_request_headers(req, transfer=True) + for k, v in good_hdrs.iteritems(): + self.assertTrue(k.lower() in dst_headers) + self.assertEqual(v, dst_headers[k.lower()]) + for k, v in bad_hdrs.iteritems(): + self.assertFalse(k.lower() in dst_headers) diff --git a/test/unit/proxy/controllers/test_container.py b/test/unit/proxy/controllers/test_container.py index 63e6b0e..7c8ecf7 100644 --- a/test/unit/proxy/controllers/test_container.py +++ b/test/unit/proxy/controllers/test_container.py @@ -20,6 +20,7 @@ from swift.common.swob import Request from swift.proxy import server as proxy_server from swift.proxy.controllers.base import headers_to_container_info from test.unit import fake_http_connect, FakeRing, FakeMemcache +from swift.common.request_helpers import get_sys_meta_prefix class TestContainerController(unittest.TestCase): @@ -33,7 +34,7 @@ class TestContainerController(unittest.TestCase): controller = proxy_server.ContainerController(self.app, 'a', 'c') with mock.patch('swift.proxy.controllers.base.http_connect', fake_http_connect(200, 200, body='')): - req = Request.blank('/a/c', {'PATH_INFO': '/a/c'}) + req = Request.blank('/v1/a/c', {'PATH_INFO': '/v1/a/c'}) resp = controller.HEAD(req) self.assertEqual(2, resp.status_int // 100) self.assertTrue("swift.container/a/c" in resp.environ) @@ -46,7 +47,7 @@ class TestContainerController(unittest.TestCase): 'x-container-sync-key': 'value', 'x-container-sync-to': 'value'} controller = proxy_server.ContainerController(self.app, 'a', 'c') - req = Request.blank('/a/c') + req = Request.blank('/v1/a/c') with mock.patch('swift.proxy.controllers.base.http_connect', fake_http_connect(200, 200, headers=owner_headers)): resp = controller.HEAD(req) @@ -54,7 +55,7 @@ class TestContainerController(unittest.TestCase): for key in owner_headers: self.assertTrue(key not in resp.headers) - req = Request.blank('/a/c', environ={'swift_owner': True}) + req = Request.blank('/v1/a/c', environ={'swift_owner': True}) with mock.patch('swift.proxy.controllers.base.http_connect', fake_http_connect(200, 200, headers=owner_headers)): resp = controller.HEAD(req) @@ -62,6 +63,61 @@ class TestContainerController(unittest.TestCase): for key in owner_headers: self.assertTrue(key in resp.headers) + def _make_callback_func(self, context): + def callback(ipaddr, port, device, partition, method, path, + headers=None, query_string=None, ssl=False): + context['method'] = method + context['path'] = path + context['headers'] = headers or {} + return callback + + def test_sys_meta_headers_PUT(self): + # check that headers in sys meta namespace make it through + # the container controller + sys_meta_key = '%stest' % get_sys_meta_prefix('container') + sys_meta_key = sys_meta_key.title() + user_meta_key = 'X-Container-Meta-Test' + controller = proxy_server.ContainerController(self.app, 'a', 'c') + + context = {} + callback = self._make_callback_func(context) + hdrs_in = {sys_meta_key: 'foo', + user_meta_key: 'bar', + 'x-timestamp': '1.0'} + req = Request.blank('/v1/a/c', headers=hdrs_in) + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200, 200, give_connect=callback)): + controller.PUT(req) + self.assertEqual(context['method'], 'PUT') + self.assertTrue(sys_meta_key in context['headers']) + self.assertEqual(context['headers'][sys_meta_key], 'foo') + self.assertTrue(user_meta_key in context['headers']) + self.assertEqual(context['headers'][user_meta_key], 'bar') + self.assertNotEqual(context['headers']['x-timestamp'], '1.0') + + def test_sys_meta_headers_POST(self): + # check that headers in sys meta namespace make it through + # the container controller + sys_meta_key = '%stest' % get_sys_meta_prefix('container') + sys_meta_key = sys_meta_key.title() + user_meta_key = 'X-Container-Meta-Test' + controller = proxy_server.ContainerController(self.app, 'a', 'c') + context = {} + callback = self._make_callback_func(context) + hdrs_in = {sys_meta_key: 'foo', + user_meta_key: 'bar', + 'x-timestamp': '1.0'} + req = Request.blank('/v1/a/c', headers=hdrs_in) + with mock.patch('swift.proxy.controllers.base.http_connect', + fake_http_connect(200, 200, give_connect=callback)): + controller.POST(req) + self.assertEqual(context['method'], 'POST') + self.assertTrue(sys_meta_key in context['headers']) + self.assertEqual(context['headers'][sys_meta_key], 'foo') + self.assertTrue(user_meta_key in context['headers']) + self.assertEqual(context['headers'][user_meta_key], 'bar') + self.assertNotEqual(context['headers']['x-timestamp'], '1.0') + if __name__ == '__main__': unittest.main() diff --git a/test/unit/proxy/controllers/test_info.py b/test/unit/proxy/controllers/test_info.py new file mode 100644 index 0000000..f33beba --- /dev/null +++ b/test/unit/proxy/controllers/test_info.py @@ -0,0 +1,293 @@ +# 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 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.utils import json +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() |