| #!/usr/bin/env python |
| # Copyright 2013 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import functools |
| import os |
| |
| from appengine_wrappers import GetAppVersion |
| from compiled_file_system import CompiledFileSystem |
| from copy import deepcopy |
| from file_system import FileNotFoundError, FileSystem, StatInfo |
| from future import Gettable, Future |
| from mock_file_system import MockFileSystem |
| from object_store_creator import ObjectStoreCreator |
| from test_file_system import TestFileSystem |
| from test_object_store import TestObjectStore |
| import unittest |
| |
| _TEST_DATA = { |
| '404.html': '404.html contents', |
| 'apps': { |
| 'a11y.html': 'a11y.html contents', |
| 'about_apps.html': 'about_apps.html contents', |
| 'fakedir': { |
| 'file.html': 'file.html contents' |
| }, |
| 'deepdir': { |
| 'deepfile.html': 'deepfile.html contents', |
| 'deeper': { |
| 'deepest.html': 'deepest.html contents', |
| }, |
| } |
| }, |
| 'extensions': { |
| 'activeTab.html': 'activeTab.html contents', |
| 'alarms.html': 'alarms.html contents' |
| } |
| } |
| |
| identity = lambda _, x: x |
| |
| def _GetTestCompiledFsCreator(): |
| '''Returns a function which creates CompiledFileSystem views of |
| TestFileSystems backed by _TEST_DATA. |
| ''' |
| return functools.partial( |
| CompiledFileSystem.Factory( |
| ObjectStoreCreator(start_empty=False, |
| store_type=TestObjectStore, |
| disable_wrappers=True), |
| ).Create, |
| TestFileSystem(deepcopy(_TEST_DATA))) |
| |
| class CompiledFileSystemTest(unittest.TestCase): |
| def testPopulateNamespace(self): |
| def CheckNamespace(expected_file, expected_list, fs): |
| self.assertEqual(expected_file, fs._file_object_store.namespace) |
| self.assertEqual(expected_list, fs._list_object_store.namespace) |
| compiled_fs_creator = _GetTestCompiledFsCreator() |
| f = lambda x: x |
| CheckNamespace( |
| 'class=CompiledFileSystem&' |
| 'category=CompiledFileSystemTest/TestFileSystem/file&' |
| 'app_version=%s' % GetAppVersion(), |
| 'class=CompiledFileSystem&' |
| 'category=CompiledFileSystemTest/TestFileSystem/list&' |
| 'app_version=%s' % GetAppVersion(), |
| compiled_fs_creator(f, CompiledFileSystemTest)) |
| CheckNamespace( |
| 'class=CompiledFileSystem&' |
| 'category=CompiledFileSystemTest/TestFileSystem/foo/file&' |
| 'app_version=%s' % GetAppVersion(), |
| 'class=CompiledFileSystem&' |
| 'category=CompiledFileSystemTest/TestFileSystem/foo/list&' |
| 'app_version=%s' % GetAppVersion(), |
| compiled_fs_creator(f, CompiledFileSystemTest, category='foo')) |
| |
| def testPopulateFromFile(self): |
| def Sleepy(key, val): |
| return '%s%s' % ('Z' * len(key), 'z' * len(val)) |
| compiled_fs = _GetTestCompiledFsCreator()(Sleepy, CompiledFileSystemTest) |
| self.assertEqual('ZZZZZZZZzzzzzzzzzzzzzzzzz', |
| compiled_fs.GetFromFile('404.html').Get()) |
| self.assertEqual('ZZZZZZZZZZZZZZzzzzzzzzzzzzzzzzzz', |
| compiled_fs.GetFromFile('apps/a11y.html').Get()) |
| self.assertEqual('ZZZZZZZZZZZZZZZZZZZZZZZzzzzzzzzzzzzzzzzzz', |
| compiled_fs.GetFromFile('/apps/fakedir/file.html').Get()) |
| |
| def testPopulateFromFileListing(self): |
| def strip_ext(path, files): |
| return [os.path.splitext(f)[0] for f in files] |
| compiled_fs = _GetTestCompiledFsCreator()(strip_ext, CompiledFileSystemTest) |
| expected_top_listing = [ |
| '404', |
| 'apps/a11y', |
| 'apps/about_apps', |
| 'apps/deepdir/deeper/deepest', |
| 'apps/deepdir/deepfile', |
| 'apps/fakedir/file', |
| 'extensions/activeTab', |
| 'extensions/alarms' |
| ] |
| self.assertEqual(expected_top_listing, |
| sorted(compiled_fs.GetFromFileListing('/').Get())) |
| self.assertEqual(expected_top_listing, |
| sorted(compiled_fs.GetFromFileListing('').Get())) |
| expected_apps_listing = [ |
| 'a11y', |
| 'about_apps', |
| 'deepdir/deeper/deepest', |
| 'deepdir/deepfile', |
| 'fakedir/file', |
| ] |
| self.assertEqual(expected_apps_listing, |
| sorted(compiled_fs.GetFromFileListing('/apps/').Get())) |
| self.assertEqual(expected_apps_listing, |
| sorted(compiled_fs.GetFromFileListing('apps/').Get())) |
| self.assertEqual(['file',], |
| compiled_fs.GetFromFileListing('/apps/fakedir/').Get()) |
| self.assertEqual(['file',], |
| compiled_fs.GetFromFileListing('apps/fakedir/').Get()) |
| self.assertEqual(['deeper/deepest', 'deepfile'], |
| sorted(compiled_fs.GetFromFileListing( |
| '/apps/deepdir/').Get())) |
| self.assertEqual(['deeper/deepest', 'deepfile'], |
| sorted(compiled_fs.GetFromFileListing( |
| 'apps/deepdir/').Get())) |
| self.assertEqual(['deepest'], |
| compiled_fs.GetFromFileListing( |
| '/apps/deepdir/deeper/').Get()) |
| self.assertEqual(['deepest'], |
| compiled_fs.GetFromFileListing( |
| 'apps/deepdir/deeper/').Get()) |
| |
| def testCaching(self): |
| compiled_fs = _GetTestCompiledFsCreator()(identity, CompiledFileSystemTest) |
| self.assertEqual('404.html contents', |
| compiled_fs.GetFromFile('404.html').Get()) |
| self.assertEqual(set(('file.html',)), |
| set(compiled_fs.GetFromFileListing('apps/fakedir').Get())) |
| |
| compiled_fs._file_system._obj['404.html'] = 'boom' |
| compiled_fs._file_system._obj['apps']['fakedir']['boom.html'] = 'blam' |
| self.assertEqual('404.html contents', |
| compiled_fs.GetFromFile('404.html').Get()) |
| self.assertEqual(set(('file.html',)), |
| set(compiled_fs.GetFromFileListing('apps/fakedir').Get())) |
| |
| compiled_fs._file_system.IncrementStat() |
| self.assertEqual('boom', compiled_fs.GetFromFile('404.html').Get()) |
| self.assertEqual(set(('file.html', 'boom.html')), |
| set(compiled_fs.GetFromFileListing('apps/fakedir').Get())) |
| |
| def testFailures(self): |
| compiled_fs = _GetTestCompiledFsCreator()(identity, CompiledFileSystemTest) |
| self.assertRaises(FileNotFoundError, |
| compiled_fs.GetFromFile('405.html').Get) |
| # TODO(kalman): would be nice to test this fails since apps/ is a dir. |
| compiled_fs.GetFromFile('apps/') |
| #self.assertRaises(SomeError, compiled_fs.GetFromFile, 'apps/') |
| self.assertRaises(FileNotFoundError, |
| compiled_fs.GetFromFileListing('nodir/').Get) |
| # TODO(kalman): likewise, not a FileNotFoundError. |
| self.assertRaises(FileNotFoundError, |
| compiled_fs.GetFromFileListing('404.html').Get) |
| |
| def testCorrectFutureBehaviour(self): |
| # Tests that the underlying FileSystem's Read Future has had Get() called |
| # on it before the Future is resolved, but the underlying Future isn't |
| # resolved until Get is. |
| mock_fs = MockFileSystem(TestFileSystem(_TEST_DATA)) |
| compiled_fs = CompiledFileSystem.Factory( |
| ObjectStoreCreator.ForTest()).Create( |
| mock_fs, lambda path, contents: contents, type(self)) |
| |
| self.assertTrue(*mock_fs.CheckAndReset()) |
| future = compiled_fs.GetFromFile('404.html') |
| self.assertTrue(*mock_fs.CheckAndReset(stat_count=1, read_count=1)) |
| future.Get() |
| self.assertTrue(*mock_fs.CheckAndReset(read_resolve_count=1)) |
| |
| future = compiled_fs.GetFromFileListing('apps/') |
| # Current behaviour is to have read=2 and read_resolve=1 because the first |
| # level is read eagerly, then all of the second is read (in parallel). If |
| # it weren't eager (and it may be worth experimenting with that) then it'd |
| # be read=1 and read_resolve=0. |
| self.assertTrue(*mock_fs.CheckAndReset(stat_count=1, |
| read_count=2, |
| read_resolve_count=1)) |
| future.Get() |
| # It's doing 1 more level 'deeper' (already read 'fakedir' and 'deepdir' |
| # though not resolved), so that's 1 more read/resolve + the resolve from |
| # the first read. |
| self.assertTrue(*mock_fs.CheckAndReset(read_count=1, read_resolve_count=2)) |
| |
| # Even though the directory is 1 layer deep the caller has no way of |
| # determining that ahead of time (though perhaps the API could give some |
| # kind of clue, if we really cared). |
| future = compiled_fs.GetFromFileListing('extensions/') |
| self.assertTrue(*mock_fs.CheckAndReset(stat_count=1, |
| read_count=1, |
| read_resolve_count=1)) |
| future.Get() |
| self.assertTrue(*mock_fs.CheckAndReset()) |
| |
| # Similar configuration to the 'apps/' case but deeper. |
| future = compiled_fs.GetFromFileListing('/') |
| self.assertTrue(*mock_fs.CheckAndReset(stat_count=1, |
| read_count=2, |
| read_resolve_count=1)) |
| future.Get() |
| self.assertTrue(*mock_fs.CheckAndReset(read_count=2, read_resolve_count=3)) |
| |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
