1 # Copyright (c) 2012 The Chromium Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import os 6 import posixpath 7 8 from future import Future 9 from path_util import AssertIsDirectory, IsDirectory 10 11 12 class _Response(object): 13 def __init__(self, content=''): 14 self.content = content 15 self.headers = {'Content-Type': 'none'} 16 self.status_code = 200 17 18 19 class FakeUrlFetcher(object): 20 def __init__(self, base_path): 21 self._base_path = base_path 22 # Mock capabilities. Perhaps this class should be MockUrlFetcher. 23 self._sync_count = 0 24 self._async_count = 0 25 self._async_resolve_count = 0 26 27 def _ReadFile(self, filename): 28 # Fake DownloadError, the error that appengine usually raises. 29 class DownloadError(Exception): pass 30 try: 31 with open(os.path.join(self._base_path, filename), 'r') as f: 32 return f.read() 33 except IOError as e: 34 raise DownloadError(e) 35 36 def _ListDir(self, directory): 37 # In some tests, we need to test listing a directory from the HTML returned 38 # from SVN. This reads an HTML file that has the directories HTML. 39 if not os.path.isdir(os.path.join(self._base_path, directory)): 40 return self._ReadFile(directory[:-1]) 41 files = os.listdir(os.path.join(self._base_path, directory)) 42 html = '<html><title>Revision: 00000</title>\n' 43 for filename in files: 44 if filename.startswith('.'): 45 continue 46 if os.path.isdir(os.path.join(self._base_path, directory, filename)): 47 html += '<a>' + filename + '/</a>\n' 48 else: 49 html += '<a>' + filename + '</a>\n' 50 html += '</html>' 51 return html 52 53 def FetchAsync(self, url): 54 self._async_count += 1 55 url = url.rsplit('?', 1)[0] 56 def resolve(): 57 self._async_resolve_count += 1 58 return self._DoFetch(url) 59 return Future(callback=resolve) 60 61 def Fetch(self, url): 62 self._sync_count += 1 63 return self._DoFetch(url) 64 65 def _DoFetch(self, url): 66 url = url.rsplit('?', 1)[0] 67 result = _Response() 68 if IsDirectory(url): 69 result.content = self._ListDir(url) 70 else: 71 result.content = self._ReadFile(url) 72 return result 73 74 def CheckAndReset(self, sync_count=0, async_count=0, async_resolve_count=0): 75 '''Returns a tuple (success, error). Use in tests like: 76 self.assertTrue(*fetcher.CheckAndReset(...)) 77 ''' 78 errors = [] 79 for desc, expected, actual in ( 80 ('sync_count', sync_count, self._sync_count), 81 ('async_count', async_count, self._async_count), 82 ('async_resolve_count', async_resolve_count, 83 self._async_resolve_count)): 84 if actual != expected: 85 errors.append('%s: expected %s got %s' % (desc, expected, actual)) 86 try: 87 return (len(errors) == 0, ', '.join(errors)) 88 finally: 89 self.Reset() 90 91 def Reset(self): 92 self._sync_count = 0 93 self._async_count = 0 94 self._async_resolve_count = 0 95 96 97 class FakeURLFSFetcher(object): 98 '''Use a file_system to resolve fake fetches. Mimics the interface of Google 99 Appengine's urlfetch. 100 ''' 101 102 def __init__(self, file_system, base_path): 103 AssertIsDirectory(base_path) 104 self._base_path = base_path 105 self._file_system = file_system 106 107 def FetchAsync(self, url, **kwargs): 108 return Future(value=self.Fetch(url)) 109 110 def Fetch(self, url, **kwargs): 111 return _Response(self._file_system.ReadSingle( 112 posixpath.join(self._base_path, url)).Get()) 113 114 def UpdateFS(self, file_system, base_path=None): 115 '''Replace the underlying FileSystem used to reslove URLs. 116 ''' 117 self._file_system = file_system 118 self._base_path = base_path or self._base_path 119 120 121 class MockURLFetcher(object): 122 def __init__(self, fetcher): 123 self._fetcher = fetcher 124 self.Reset() 125 126 def Fetch(self, url, **kwargs): 127 self._fetch_count += 1 128 return self._fetcher.Fetch(url, **kwargs) 129 130 def FetchAsync(self, url, **kwargs): 131 self._fetch_async_count += 1 132 future = self._fetcher.FetchAsync(url, **kwargs) 133 def resolve(): 134 self._fetch_resolve_count += 1 135 return future.Get() 136 return Future(callback=resolve) 137 138 def CheckAndReset(self, 139 fetch_count=0, 140 fetch_async_count=0, 141 fetch_resolve_count=0): 142 errors = [] 143 for desc, expected, actual in ( 144 ('fetch_count', fetch_count, self._fetch_count), 145 ('fetch_async_count', fetch_async_count, self._fetch_async_count), 146 ('fetch_resolve_count', fetch_resolve_count, 147 self._fetch_resolve_count)): 148 if actual != expected: 149 errors.append('%s: expected %s got %s' % (desc, expected, actual)) 150 try: 151 return (len(errors) == 0, ', '.join(errors)) 152 finally: 153 self.Reset() 154 155 def Reset(self): 156 self._fetch_count = 0 157 self._fetch_async_count = 0 158 self._fetch_resolve_count = 0 159
