src/chrome/common/extensions/docs/server2/subversion_file_system_test.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2012 The Chromium 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 import json
   7 import os
   8 import unittest
   9
  10 from extensions_paths import SERVER2
  11 from fake_url_fetcher import FakeUrlFetcher
  12 from file_system import FileSystemError, StatInfo
  13 from subversion_file_system import SubversionFileSystem
  14 from test_util import ReadFile, Server2Path
  15
  16
  17 _SHARED_FILE_SYSTEM_TEST_DATA = Server2Path('test_data', 'file_system')
  18 _SUBVERSION_FILE_SYSTEM_TEST_DATA = Server2Path(
  19     'test_data', 'subversion_file_system')
  20
  21
  22 def _CreateSubversionFileSystem(path):
  23   fetcher = FakeUrlFetcher(path)
  24   return SubversionFileSystem(fetcher, fetcher, path), fetcher
  25
  26
  27 class SubversionFileSystemTest(unittest.TestCase):
  28   def testReadFiles(self):
  29     expected = {
  30       'test1.txt': 'test1\n',
  31       'test2.txt': 'test2\n',
  32       'test3.txt': 'test3\n',
  33     }
  34     file_system, fetcher = _CreateSubversionFileSystem(
  35         _SHARED_FILE_SYSTEM_TEST_DATA)
  36     read_future = file_system.Read(['test1.txt', 'test2.txt', 'test3.txt'])
  37     self.assertTrue(*fetcher.CheckAndReset(async_count=3))
  38     self.assertEqual(expected, read_future.Get())
  39     self.assertTrue(*fetcher.CheckAndReset(async_resolve_count=3))
  40
  41   def testListDir(self):
  42     expected = ['dir/'] + ['file%d.html' % i for i in range(7)]
  43     file_system, fetcher = _CreateSubversionFileSystem(
  44         _SHARED_FILE_SYSTEM_TEST_DATA)
  45     list_future = file_system.ReadSingle('list/')
  46     self.assertTrue(*fetcher.CheckAndReset(async_count=1))
  47     self.assertEqual(expected, sorted(list_future.Get()))
  48     self.assertTrue(*fetcher.CheckAndReset(async_resolve_count=1))
  49
  50   def testListSubDir(self):
  51     expected = ['empty.txt'] + ['file%d.html' % i for i in range(3)]
  52     file_system, fetcher = _CreateSubversionFileSystem(
  53         _SHARED_FILE_SYSTEM_TEST_DATA)
  54     list_future = file_system.ReadSingle('list/dir/')
  55     self.assertTrue(*fetcher.CheckAndReset(async_count=1))
  56     self.assertEqual(expected, sorted(list_future.Get()))
  57     self.assertTrue(*fetcher.CheckAndReset(async_resolve_count=1))
  58
  59   def testDirStat(self):
  60     file_system, fetcher = _CreateSubversionFileSystem(
  61         _SHARED_FILE_SYSTEM_TEST_DATA)
  62     stat_info = file_system.Stat('stat/')
  63     self.assertTrue(*fetcher.CheckAndReset(async_count=1,
  64                                            async_resolve_count=1))
  65     expected = StatInfo(
  66       '151113',
  67       child_versions=json.loads(ReadFile(
  68           SERVER2, 'test_data', 'file_system', 'stat_result.json')))
  69     self.assertEqual(expected, stat_info)
  70
  71   def testFileStat(self):
  72     file_system, fetcher = _CreateSubversionFileSystem(
  73         _SHARED_FILE_SYSTEM_TEST_DATA)
  74     stat_info = file_system.Stat('stat/extension_api.h')
  75     self.assertTrue(*fetcher.CheckAndReset(async_count=1,
  76                                            async_resolve_count=1))
  77     self.assertEqual(StatInfo('146163'), stat_info)
  78
  79   def testRevisions(self):
  80     # This is a super hacky test. Record the path that was fetched then exit the
  81     # test. Compare.
  82     class ValueErrorFetcher(object):
  83       def __init__(self):
  84         self.last_fetched = None
  85
  86       def FetchAsync(self, path):
  87         class ThrowsValueError(object):
  88           def Get(self): raise ValueError()
  89         self.last_fetched = path
  90         return ThrowsValueError()
  91
  92       def Fetch(self, path, **kwargs):
  93         self.last_fetched = path
  94         raise ValueError()
  95
  96     file_fetcher = ValueErrorFetcher()
  97     stat_fetcher = ValueErrorFetcher()
  98     svn_path = 'svn:'
  99
 100     svn_file_system = SubversionFileSystem(file_fetcher,
 101                                            stat_fetcher,
 102                                            svn_path,
 103                                            revision=42)
 104
 105     self.assertRaises(FileSystemError,
 106                       svn_file_system.ReadSingle('dir/file').Get)
 107     self.assertEqual('dir/file?p=42', file_fetcher.last_fetched)
 108     # Stat() will always stat directories.
 109     self.assertRaises(FileSystemError, svn_file_system.Stat, 'dir/file')
 110     self.assertEqual('dir?pathrev=42', stat_fetcher.last_fetched)
 111
 112     self.assertRaises(FileSystemError,
 113                       svn_file_system.ReadSingle('dir/').Get)
 114     self.assertEqual('dir/?p=42', file_fetcher.last_fetched)
 115     self.assertRaises(FileSystemError, svn_file_system.Stat, 'dir/')
 116     self.assertEqual('dir?pathrev=42', stat_fetcher.last_fetched)
 117
 118   def testDirectoryVersionOnDeletion(self):
 119     '''Tests the case when the most recent operation on a directory is the
 120     deletion of a file. Here it is not enough to take the maximum version of all
 121     files in the directory, as we used to, for obvious reasons.
 122     '''
 123     file_system, _ = _CreateSubversionFileSystem(
 124         _SUBVERSION_FILE_SYSTEM_TEST_DATA)
 125     dir_stat = file_system.Stat('docs_public_extensions_214898/')
 126     self.assertEqual('214692', dir_stat.version)
 127
 128   def testEmptyDirectory(self):
 129     file_system, _ = _CreateSubversionFileSystem(
 130         _SUBVERSION_FILE_SYSTEM_TEST_DATA)
 131     dir_stat = file_system.Stat('api_icons_214898/')
 132     self.assertEqual('193838', dir_stat.version)
 133     self.assertEqual({}, dir_stat.child_versions)
 134
 135     def testSkipNotFound(self):
 136       file_system, _ = _CreateSubversionFileSystem(
 137           _SUBVERSION_FILE_SYSTEM_TEST_DATA)
 138       self.assertEqual({}, file_system.Read(('fakefile',),
 139                                             skip_not_found=True).Get())
 140
 141
 142 if __name__ == '__main__':
 143   unittest.main()