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