# 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 mock import tempfile import time from test import safe_repr from test.unit import MockTrue from swift.common.swob import HTTPBadRequest, Request, HTTPException from swift.common.http import HTTP_REQUEST_ENTITY_TOO_LARGE, \ HTTP_BAD_REQUEST, HTTP_LENGTH_REQUIRED from swift.common import constraints, utils class TestConstraints(unittest.TestCase): def assertIn(self, member, container, msg=None): """Copied from 2.7""" if member not in container: standardMsg = '%s not found in %s' % (safe_repr(member), safe_repr(container)) self.fail(self._formatMessage(msg, standardMsg)) def test_check_metadata_empty(self): headers = {} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) def test_check_metadata_good(self): headers = {'X-Object-Meta-Name': 'Value'} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) def test_check_metadata_empty_name(self): headers = {'X-Object-Meta-': 'Value'} self.assert_(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), HTTPBadRequest) def test_check_metadata_name_length(self): name = 'a' * constraints.MAX_META_NAME_LENGTH headers = {'X-Object-Meta-%s' % name: 'v'} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) name = 'a' * (constraints.MAX_META_NAME_LENGTH + 1) headers = {'X-Object-Meta-%s' % name: 'v'} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object').status_int, HTTP_BAD_REQUEST) self.assertIn( ('X-Object-Meta-%s' % name).lower(), constraints.check_metadata(Request.blank( '/', headers=headers), 'object').body.lower()) def test_check_metadata_value_length(self): value = 'a' * constraints.MAX_META_VALUE_LENGTH headers = {'X-Object-Meta-Name': value} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) value = 'a' * (constraints.MAX_META_VALUE_LENGTH + 1) headers = {'X-Object-Meta-Name': value} self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object').status_int, HTTP_BAD_REQUEST) self.assertIn( 'x-object-meta-name', constraints.check_metadata(Request.blank( '/', headers=headers), 'object').body.lower()) self.assertIn( str(constraints.MAX_META_VALUE_LENGTH), constraints.check_metadata(Request.blank( '/', headers=headers), 'object').body) def test_check_metadata_count(self): headers = {} for x in xrange(constraints.MAX_META_COUNT): headers['X-Object-Meta-%d' % x] = 'v' self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) headers['X-Object-Meta-Too-Many'] = 'v' self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object').status_int, HTTP_BAD_REQUEST) def test_check_metadata_size(self): headers = {} size = 0 chunk = constraints.MAX_META_NAME_LENGTH + \ constraints.MAX_META_VALUE_LENGTH x = 0 while size + chunk < constraints.MAX_META_OVERALL_SIZE: headers['X-Object-Meta-%04d%s' % (x, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \ 'v' * constraints.MAX_META_VALUE_LENGTH size += chunk x += 1 self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object'), None) # add two more headers in case adding just one falls exactly on the # limit (eg one header adds 1024 and the limit is 2048) headers['X-Object-Meta-%04d%s' % (x, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \ 'v' * constraints.MAX_META_VALUE_LENGTH headers['X-Object-Meta-%04d%s' % (x + 1, 'a' * (constraints.MAX_META_NAME_LENGTH - 4))] = \ 'v' * constraints.MAX_META_VALUE_LENGTH self.assertEquals(constraints.check_metadata(Request.blank( '/', headers=headers), 'object').status_int, HTTP_BAD_REQUEST) def test_check_object_creation_content_length(self): headers = {'Content-Length': str(constraints.MAX_FILE_SIZE), 'Content-Type': 'text/plain'} self.assertEquals(constraints.check_object_creation(Request.blank( '/', headers=headers), 'object_name'), None) headers = {'Content-Length': str(constraints.MAX_FILE_SIZE + 1), 'Content-Type': 'text/plain'} self.assertEquals(constraints.check_object_creation( Request.blank('/', headers=headers), 'object_name').status_int, HTTP_REQUEST_ENTITY_TOO_LARGE) headers = {'Transfer-Encoding': 'chunked', 'Content-Type': 'text/plain'} self.assertEquals(constraints.check_object_creation(Request.blank( '/', headers=headers), 'object_name'), None) headers = {'Content-Type': 'text/plain'} self.assertEquals(constraints.check_object_creation( Request.blank('/', headers=headers), 'object_name').status_int, HTTP_LENGTH_REQUIRED) def test_check_object_creation_name_length(self): headers = {'Transfer-Encoding': 'chunked', 'Content-Type': 'text/plain'} name = 'o' * constraints.MAX_OBJECT_NAME_LENGTH self.assertEquals(constraints.check_object_creation(Request.blank( '/', headers=headers), name), None) name = 'o' * (constraints.MAX_OBJECT_NAME_LENGTH + 1) self.assertEquals(constraints.check_object_creation( Request.blank('/', headers=headers), name).status_int, HTTP_BAD_REQUEST) def test_check_object_creation_content_type(self): headers = {'Transfer-Encoding': 'chunked', 'Content-Type': 'text/plain'} self.assertEquals(constraints.check_object_creation(Request.blank( '/', headers=headers), 'object_name'), None) headers = {'Transfer-Encoding': 'chunked'} self.assertEquals(constraints.check_object_creation( Request.blank('/', headers=headers), 'object_name').status_int, HTTP_BAD_REQUEST) def test_check_object_creation_bad_content_type(self): headers = {'Transfer-Encoding': 'chunked', 'Content-Type': '\xff\xff'} resp = constraints.check_object_creation( Request.blank('/', headers=headers), 'object_name') self.assertEquals(resp.status_int, HTTP_BAD_REQUEST) self.assert_('Content-Type' in resp.body) def test_check_mount(self): self.assertFalse(constraints.check_mount('', '')) with mock.patch("swift.common.utils.ismount", MockTrue()): self.assertTrue(constraints.check_mount('/srv', '1')) self.assertTrue(constraints.check_mount('/srv', 'foo-bar')) self.assertTrue(constraints.check_mount( '/srv', '003ed03c-242a-4b2f-bee9-395f801d1699')) self.assertFalse(constraints.check_mount('/srv', 'foo bar')) self.assertFalse(constraints.check_mount('/srv', 'foo/bar')) self.assertFalse(constraints.check_mount('/srv', 'foo?bar')) def test_check_float(self): self.assertFalse(constraints.check_float('')) self.assertTrue(constraints.check_float('0')) def test_valid_timestamp(self): self.assertRaises(HTTPException, constraints.valid_timestamp, Request.blank('/')) self.assertRaises(HTTPException, constraints.valid_timestamp, Request.blank('/', headers={ 'X-Timestamp': 'asdf'})) timestamp = utils.Timestamp(time.time()) req = Request.blank('/', headers={'X-Timestamp': timestamp.internal}) self.assertEqual(timestamp, constraints.valid_timestamp(req)) req = Request.blank('/', headers={'X-Timestamp': timestamp.normal}) self.assertEqual(timestamp, constraints.valid_timestamp(req)) def test_check_utf8(self): unicode_sample = u'\uc77c\uc601' valid_utf8_str = unicode_sample.encode('utf-8') invalid_utf8_str = unicode_sample.encode('utf-8')[::-1] unicode_with_null = u'abc\u0000def' utf8_with_null = unicode_with_null.encode('utf-8') for false_argument in [None, '', invalid_utf8_str, unicode_with_null, utf8_with_null]: self.assertFalse(constraints.check_utf8(false_argument)) for true_argument in ['this is ascii and utf-8, too', unicode_sample, valid_utf8_str]: self.assertTrue(constraints.check_utf8(true_argument)) def test_validate_bad_meta(self): req = Request.blank( '/v/a/c/o', headers={'x-object-meta-hello': 'ab' * constraints.MAX_HEADER_SIZE}) self.assertEquals(constraints.check_metadata(req, 'object').status_int, HTTP_BAD_REQUEST) self.assertIn('x-object-meta-hello', constraints.check_metadata(req, 'object').body.lower()) def test_validate_constraints(self): c = constraints self.assertTrue(c.MAX_META_OVERALL_SIZE > c.MAX_META_NAME_LENGTH) self.assertTrue(c.MAX_META_OVERALL_SIZE > c.MAX_META_VALUE_LENGTH) self.assertTrue(c.MAX_HEADER_SIZE > c.MAX_META_NAME_LENGTH) self.assertTrue(c.MAX_HEADER_SIZE > c.MAX_META_VALUE_LENGTH) def test_validate_copy_from(self): req = Request.blank( '/v/a/c/o', headers={'x-copy-from': 'c/o2'}) src_cont, src_obj = constraints.check_copy_from_header(req) self.assertEqual(src_cont, 'c') self.assertEqual(src_obj, 'o2') req = Request.blank( '/v/a/c/o', headers={'x-copy-from': 'c/subdir/o2'}) src_cont, src_obj = constraints.check_copy_from_header(req) self.assertEqual(src_cont, 'c') self.assertEqual(src_obj, 'subdir/o2') req = Request.blank( '/v/a/c/o', headers={'x-copy-from': '/c/o2'}) src_cont, src_obj = constraints.check_copy_from_header(req) self.assertEqual(src_cont, 'c') self.assertEqual(src_obj, 'o2') def test_validate_bad_copy_from(self): req = Request.blank( '/v/a/c/o', headers={'x-copy-from': 'bad_object'}) self.assertRaises(HTTPException, constraints.check_copy_from_header, req) class TestConstraintsConfig(unittest.TestCase): def test_default_constraints(self): for key in constraints.DEFAULT_CONSTRAINTS: # if there is local over-rides in swift.conf we just continue on if key in constraints.OVERRIDE_CONSTRAINTS: continue # module level attrs (that aren't in OVERRIDE) should have the # same value as the DEFAULT map module_level_value = getattr(constraints, key.upper()) self.assertEquals(constraints.DEFAULT_CONSTRAINTS[key], module_level_value) def test_effective_constraints(self): for key in constraints.DEFAULT_CONSTRAINTS: # module level attrs should always mirror the same value as the # EFFECTIVE map module_level_value = getattr(constraints, key.upper()) self.assertEquals(constraints.EFFECTIVE_CONSTRAINTS[key], module_level_value) # if there are local over-rides in swift.conf those should be # reflected in the EFFECTIVE, otherwise we expect the DEFAULTs self.assertEquals(constraints.EFFECTIVE_CONSTRAINTS[key], constraints.OVERRIDE_CONSTRAINTS.get( key, constraints.DEFAULT_CONSTRAINTS[key])) def test_override_constraints(self): try: with tempfile.NamedTemporaryFile() as f: f.write('[swift-constraints]\n') # set everything to 1 for key in constraints.DEFAULT_CONSTRAINTS: f.write('%s = 1\n' % key) f.flush() with mock.patch.object(utils, 'SWIFT_CONF_FILE', f.name): constraints.reload_constraints() for key in constraints.DEFAULT_CONSTRAINTS: # module level attrs should all be 1 module_level_value = getattr(constraints, key.upper()) self.assertEquals(module_level_value, 1) # all keys should be in OVERRIDE self.assertEquals(constraints.OVERRIDE_CONSTRAINTS[key], module_level_value) # module level attrs should always mirror the same value as # the EFFECTIVE map self.assertEquals(constraints.EFFECTIVE_CONSTRAINTS[key], module_level_value) finally: constraints.reload_constraints() def test_reload_reset(self): try: with tempfile.NamedTemporaryFile() as f: f.write('[swift-constraints]\n') # set everything to 1 for key in constraints.DEFAULT_CONSTRAINTS: f.write('%s = 1\n' % key) f.flush() with mock.patch.object(utils, 'SWIFT_CONF_FILE', f.name): constraints.reload_constraints() self.assertTrue(constraints.SWIFT_CONSTRAINTS_LOADED) self.assertEquals(sorted(constraints.DEFAULT_CONSTRAINTS.keys()), sorted(constraints.OVERRIDE_CONSTRAINTS.keys())) # file is now deleted... with mock.patch.object(utils, 'SWIFT_CONF_FILE', f.name): constraints.reload_constraints() # no constraints have been loaded from non-existent swift.conf self.assertFalse(constraints.SWIFT_CONSTRAINTS_LOADED) # no constraints are in OVERRIDE self.assertEquals([], constraints.OVERRIDE_CONSTRAINTS.keys()) # the EFFECTIVE constraints mirror DEFAULT self.assertEquals(constraints.EFFECTIVE_CONSTRAINTS, constraints.DEFAULT_CONSTRAINTS) finally: constraints.reload_constraints() if __name__ == '__main__': unittest.main()