140 lines
5.9 KiB
Python
140 lines
5.9 KiB
Python
# Copyright (c) 2010-2013 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.
|
|
|
|
"""Tests for swift.common.ondisk"""
|
|
|
|
from __future__ import with_statement
|
|
from test.unit import temptree
|
|
|
|
import os
|
|
|
|
import unittest
|
|
|
|
from swift.common import ondisk
|
|
|
|
|
|
class TestOndisk(unittest.TestCase):
|
|
"""Tests for swift.common.ondisk"""
|
|
|
|
def setUp(self):
|
|
ondisk.HASH_PATH_SUFFIX = 'endcap'
|
|
ondisk.HASH_PATH_PREFIX = 'startcap'
|
|
|
|
def test_normalize_timestamp(self):
|
|
# Test swift.common.ondisk.normalize_timestamp
|
|
self.assertEquals(ondisk.normalize_timestamp('1253327593.48174'),
|
|
"1253327593.48174")
|
|
self.assertEquals(ondisk.normalize_timestamp(1253327593.48174),
|
|
"1253327593.48174")
|
|
self.assertEquals(ondisk.normalize_timestamp('1253327593.48'),
|
|
"1253327593.48000")
|
|
self.assertEquals(ondisk.normalize_timestamp(1253327593.48),
|
|
"1253327593.48000")
|
|
self.assertEquals(ondisk.normalize_timestamp('253327593.48'),
|
|
"0253327593.48000")
|
|
self.assertEquals(ondisk.normalize_timestamp(253327593.48),
|
|
"0253327593.48000")
|
|
self.assertEquals(ondisk.normalize_timestamp('1253327593'),
|
|
"1253327593.00000")
|
|
self.assertEquals(ondisk.normalize_timestamp(1253327593),
|
|
"1253327593.00000")
|
|
self.assertRaises(ValueError, ondisk.normalize_timestamp, '')
|
|
self.assertRaises(ValueError, ondisk.normalize_timestamp, 'abc')
|
|
|
|
def test_validate_device_partition(self):
|
|
# Test swift.common.ondisk.validate_device_partition
|
|
ondisk.validate_device_partition('foo', 'bar')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, '', '')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, '', 'foo')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, 'foo', '')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, 'foo/bar', 'foo')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, 'foo', 'foo/bar')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, '.', 'foo')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, '..', 'foo')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, 'foo', '.')
|
|
self.assertRaises(ValueError,
|
|
ondisk.validate_device_partition, 'foo', '..')
|
|
try:
|
|
ondisk.validate_device_partition('o\nn e', 'foo')
|
|
except ValueError as err:
|
|
self.assertEquals(str(err), 'Invalid device: o%0An%20e')
|
|
try:
|
|
ondisk.validate_device_partition('foo', 'o\nn e')
|
|
except ValueError as err:
|
|
self.assertEquals(str(err), 'Invalid partition: o%0An%20e')
|
|
|
|
def test_storage_directory(self):
|
|
self.assertEquals(ondisk.storage_directory('objects', '1', 'ABCDEF'),
|
|
'objects/1/DEF/ABCDEF')
|
|
|
|
def test_hash_path(self):
|
|
_prefix = ondisk.HASH_PATH_PREFIX
|
|
ondisk.HASH_PATH_PREFIX = ''
|
|
# Yes, these tests are deliberately very fragile. We want to make sure
|
|
# that if someones changes the results hash_path produces, they know it
|
|
try:
|
|
self.assertEquals(ondisk.hash_path('a'),
|
|
'1c84525acb02107ea475dcd3d09c2c58')
|
|
self.assertEquals(ondisk.hash_path('a', 'c'),
|
|
'33379ecb053aa5c9e356c68997cbb59e')
|
|
self.assertEquals(ondisk.hash_path('a', 'c', 'o'),
|
|
'06fbf0b514e5199dfc4e00f42eb5ea83')
|
|
self.assertEquals(
|
|
ondisk.hash_path('a', 'c', 'o', raw_digest=False),
|
|
'06fbf0b514e5199dfc4e00f42eb5ea83')
|
|
self.assertEquals(ondisk.hash_path('a', 'c', 'o', raw_digest=True),
|
|
'\x06\xfb\xf0\xb5\x14\xe5\x19\x9d\xfcN'
|
|
'\x00\xf4.\xb5\xea\x83')
|
|
self.assertRaises(ValueError, ondisk.hash_path, 'a', object='o')
|
|
ondisk.HASH_PATH_PREFIX = 'abcdef'
|
|
self.assertEquals(
|
|
ondisk.hash_path('a', 'c', 'o', raw_digest=False),
|
|
'363f9b535bfb7d17a43a46a358afca0e')
|
|
finally:
|
|
ondisk.HASH_PATH_PREFIX = _prefix
|
|
|
|
|
|
class TestAuditLocationGenerator(unittest.TestCase):
|
|
def test_non_dir_contents(self):
|
|
with temptree([]) as tmpdir:
|
|
data = os.path.join(tmpdir, "drive", "data")
|
|
os.makedirs(data)
|
|
with open(os.path.join(data, "partition1"), "w"):
|
|
pass
|
|
partition = os.path.join(data, "partition2")
|
|
os.makedirs(partition)
|
|
with open(os.path.join(partition, "suffix1"), "w"):
|
|
pass
|
|
suffix = os.path.join(partition, "suffix2")
|
|
os.makedirs(suffix)
|
|
with open(os.path.join(suffix, "hash1"), "w"):
|
|
pass
|
|
locations = ondisk.audit_location_generator(
|
|
tmpdir, "data", mount_check=False
|
|
)
|
|
self.assertEqual(list(locations), [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|