1#!/usr/bin/python 2# Copyright (c) 2012 The Native Client Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6"""Tests of directory storage adapter.""" 7 8import os 9import posixpath 10import unittest 11 12import directory_storage 13import fake_storage 14import gsd_storage 15import hashing_tools 16import hashing_tools_test 17import working_directory 18 19 20class TestDirectoryStorage(unittest.TestCase): 21 22 def setUp(self): 23 storage = fake_storage.FakeStorage() 24 self._dir_storage = directory_storage.DirectoryStorageAdapter(storage) 25 26 def test_WriteRead(self): 27 # Check that a directory can be written and then read back. 28 with working_directory.TemporaryWorkingDirectory() as work_dir: 29 temp1 = os.path.join(work_dir, 'temp1') 30 temp2 = os.path.join(work_dir, 'temp2') 31 hashing_tools_test.GenerateTestTree('write_read', temp1) 32 self._dir_storage.PutDirectory(temp1, 'foo') 33 self._dir_storage.GetDirectory('foo', temp2) 34 self.assertEqual(hashing_tools.StableHashPath(temp1), 35 hashing_tools.StableHashPath(temp2)) 36 37 def test_InputUntouched(self): 38 # Check that PutDirectory doesn't alter its inputs. 39 with working_directory.TemporaryWorkingDirectory() as work_dir: 40 temp1 = os.path.join(work_dir, 'temp1') 41 hashing_tools_test.GenerateTestTree('input_untouched', temp1) 42 h1 = hashing_tools.StableHashPath(temp1) 43 self._dir_storage.PutDirectory(temp1, 'hello') 44 h2 = hashing_tools.StableHashPath(temp1) 45 self.assertEqual(h1, h2) 46 47 def test_URLsPropagate(self): 48 # Check that consistent non-None URLs come from get and put. 49 with working_directory.TemporaryWorkingDirectory() as work_dir: 50 temp1 = os.path.join(work_dir, 'temp1') 51 temp2 = os.path.join(work_dir, 'temp2') 52 hashing_tools_test.GenerateTestTree('url_propagate', temp1) 53 url1 = self._dir_storage.PutDirectory(temp1, 'me').url 54 url2 = self._dir_storage.GetDirectory('me', temp2).url 55 self.assertEqual(url1, url2) 56 self.assertNotEqual(None, url1) 57 58 def test_DirectoryName(self): 59 with working_directory.TemporaryWorkingDirectory() as work_dir: 60 temp1 = os.path.join(work_dir, 'temp1') 61 temp2 = os.path.join(work_dir, 'temp2') 62 hashing_tools_test.GenerateTestTree('directory_name', temp1) 63 key_name = 'name_test' 64 key_location = posixpath.join('name_parent', key_name) 65 result1 = self._dir_storage.PutDirectory(temp1, key_location) 66 result2 = self._dir_storage.GetDirectory(key_location, temp2) 67 self.assertNotEqual(None, result1) 68 self.assertNotEqual(None, result2) 69 self.assertEqual(key_name, result1.name) 70 self.assertEqual(key_name, result2.name) 71 72 def test_HashStable(self): 73 with working_directory.TemporaryWorkingDirectory() as work_dir: 74 temp1 = os.path.join(work_dir, 'temp1') 75 temp2 = os.path.join(work_dir, 'temp2') 76 hashing_tools_test.GenerateTestTree('stable_hash', temp1) 77 result1 = self._dir_storage.PutDirectory(temp1, 'hash_test') 78 result2 = self._dir_storage.GetDirectory('hash_test', temp2) 79 self.assertNotEqual(None, result1) 80 self.assertNotEqual(None, result2) 81 self.assertEqual(result1.hash, result2.hash) 82 83 def test_CustomHasher(self): 84 hash_value = 'test_hash_value' 85 custom_hasher = lambda filename : hash_value 86 with working_directory.TemporaryWorkingDirectory() as work_dir: 87 temp = os.path.join(work_dir, 'temp') 88 hashing_tools_test.GenerateTestTree('custom_hasher', temp) 89 result = self._dir_storage.PutDirectory( 90 temp, 91 'custom_hasher', 92 hasher=custom_hasher 93 ) 94 self.assertNotEqual(None, result) 95 self.assertEqual(result.hash, hash_value) 96 97 def test_BadWrite(self): 98 def call(cmd): 99 return 1 100 storage = gsd_storage.GSDStorage( 101 gsutil=['mygsutil'], 102 write_bucket='mybucket', 103 read_buckets=[], 104 call=call) 105 dir_storage = directory_storage.DirectoryStorageAdapter(storage) 106 # Check that storage exceptions come thru on failure. 107 with working_directory.TemporaryWorkingDirectory() as work_dir: 108 temp1 = os.path.join(work_dir, 'temp1') 109 hashing_tools_test.GenerateTestTree('bad_write', temp1) 110 self.assertRaises(gsd_storage.GSDStorageError, 111 dir_storage.PutDirectory, temp1, 'bad') 112 113 def test_BadRead(self): 114 # Check that storage exceptions come thru on failure. 115 with working_directory.TemporaryWorkingDirectory() as work_dir: 116 temp1 = os.path.join(work_dir, 'temp1') 117 self.assertEqual(None, self._dir_storage.GetDirectory('foo', temp1)) 118 119 120if __name__ == '__main__': 121 unittest.main() 122