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
6import json
7import os
8import unittest
9
10from extensions_paths import SERVER2
11from fake_url_fetcher import FakeUrlFetcher
12from file_system import FileSystemError, StatInfo
13from subversion_file_system import SubversionFileSystem
14from 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
22def _CreateSubversionFileSystem(path):
23  fetcher = FakeUrlFetcher(path)
24  return SubversionFileSystem(fetcher, fetcher, path), fetcher
25
26
27class 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
142if __name__ == '__main__':
143  unittest.main()
144