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