OLD | NEW |
1 #!/usr/bin/env python | 1 #!/usr/bin/env python |
2 # Copyright 2016 The LUCI Authors. All rights reserved. | 2 # Copyright 2016 The LUCI Authors. All rights reserved. |
3 # Use of this source code is governed under the Apache License, Version 2.0 | 3 # Use of this source code is governed under the Apache License, Version 2.0 |
4 # that can be found in the LICENSE file. | 4 # that can be found in the LICENSE file. |
5 | 5 |
6 import logging | 6 import logging |
7 import os | 7 import os |
8 import sys | 8 import sys |
9 import tempfile | 9 import tempfile |
10 import unittest | 10 import unittest |
11 | 11 |
12 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath( | 12 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath( |
13 __file__.decode(sys.getfilesystemencoding())))) | 13 __file__.decode(sys.getfilesystemencoding())))) |
14 sys.path.insert(0, ROOT_DIR) | 14 sys.path.insert(0, ROOT_DIR) |
15 sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party')) | 15 sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party')) |
16 | 16 |
17 from depot_tools import fix_encoding | 17 from depot_tools import fix_encoding |
18 from utils import file_path | 18 from utils import file_path |
| 19 from utils import fs |
19 import named_cache | 20 import named_cache |
20 | 21 |
21 | 22 |
22 class CacheManagerTest(unittest.TestCase): | 23 class CacheManagerTest(unittest.TestCase): |
23 def setUp(self): | 24 def setUp(self): |
24 self.tempdir = tempfile.mkdtemp(prefix=u'named_cache_test') | 25 self.tempdir = tempfile.mkdtemp(prefix=u'named_cache_test') |
25 self.manager = named_cache.CacheManager(self.tempdir) | 26 self.manager = named_cache.CacheManager(self.tempdir) |
26 | 27 |
27 def tearDown(self): | 28 def tearDown(self): |
28 try: | 29 try: |
(...skipping 56 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
85 for i in xrange(item_count): | 86 for i in xrange(item_count): |
86 self.manager.request(str(i)) | 87 self.manager.request(str(i)) |
87 self.assertEqual(len(self.manager), item_count) | 88 self.assertEqual(len(self.manager), item_count) |
88 self.manager.trim(None) | 89 self.manager.trim(None) |
89 self.assertEqual(len(self.manager), named_cache.MAX_CACHE_SIZE) | 90 self.assertEqual(len(self.manager), named_cache.MAX_CACHE_SIZE) |
90 self.assertEqual( | 91 self.assertEqual( |
91 set(map(str, xrange(10, 10 + named_cache.MAX_CACHE_SIZE))), | 92 set(map(str, xrange(10, 10 + named_cache.MAX_CACHE_SIZE))), |
92 set(os.listdir(os.path.join(self.tempdir, 'named'))), | 93 set(os.listdir(os.path.join(self.tempdir, 'named'))), |
93 ) | 94 ) |
94 | 95 |
| 96 def test_corrupted(self): |
| 97 with open(os.path.join(self.tempdir, u'state.json'), 'w') as f: |
| 98 f.write('}}}}') |
| 99 fs.makedirs(os.path.join(self.tempdir, 'a'), 0777) |
| 100 with self.manager.open(): |
| 101 self.assertFalse(os.path.isdir(self.tempdir)) |
| 102 self.manager.request('a') |
| 103 self.assertTrue(fs.islink(os.path.join(self.tempdir, 'named', 'a'))) |
| 104 |
95 | 105 |
96 if __name__ == '__main__': | 106 if __name__ == '__main__': |
97 fix_encoding.fix_encoding() | 107 fix_encoding.fix_encoding() |
98 VERBOSE = '-v' in sys.argv | 108 VERBOSE = '-v' in sys.argv |
99 logging.basicConfig(level=logging.DEBUG if VERBOSE else logging.ERROR) | 109 logging.basicConfig(level=logging.DEBUG if VERBOSE else logging.ERROR) |
100 unittest.main() | 110 unittest.main() |
OLD | NEW |