This commit is contained in:
Kovid Goyal 2019-12-05 20:05:57 +05:30
parent 99bc1a8e65
commit 0f852ee6f3
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C

View File

@ -314,8 +314,14 @@ def find_tests():
self.ae(instance['key'], key) self.ae(instance['key'], key)
open(os.path.join(temp_path, instance['path'], 'sentinel'), 'wb').write(b'test') open(os.path.join(temp_path, instance['path'], 'sentinel'), 'wb').write(b'test')
def set_data(x):
if not isinstance(x, bytes):
x = x.encode('utf-8')
with open(book_src, 'wb') as f:
f.write(x)
book_src = os.path.join(self.tdir, 'book.epub') book_src = os.path.join(self.tdir, 'book.epub')
open(book_src, 'wb').write(b'a') set_data('a')
path = prepare_book(book_src, convert_func=convert_mock) path = prepare_book(book_src, convert_func=convert_mock)
self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'test') self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'test')
@ -324,7 +330,7 @@ def find_tests():
self.ae(path, second_path) self.ae(path, second_path)
# Test that changing the book updates the cache # Test that changing the book updates the cache
open(book_src, 'wb').write(b'bc') set_data('bc')
third_path = prepare_book(book_src, convert_func=convert_mock) third_path = prepare_book(book_src, convert_func=convert_mock)
self.assertNotEqual(path, third_path) self.assertNotEqual(path, third_path)
@ -335,21 +341,21 @@ def find_tests():
self.assertNotEqual(third_path, fourth_path) self.assertNotEqual(third_path, fourth_path)
# Test cache expiry # Test cache expiry
open(book_src, 'wb').write(b'bcd') set_data('bcd')
prepare_book(book_src, convert_func=convert_mock, max_age=-1000) prepare_book(book_src, convert_func=convert_mock, max_age=-1000)
self.ae([], os.listdir(os.path.join(book_cache_dir(), 'f'))) self.ae([], os.listdir(os.path.join(book_cache_dir(), 'f')))
# Test modifying a book and opening it repeatedly leaves only # Test modifying a book and opening it repeatedly leaves only
# a single entry for it in the cache # a single entry for it in the cache
prepare_book(book_src, convert_func=convert_mock, force_expire=True) opath = prepare_book(book_src, convert_func=convert_mock, force_expire=True)
finished_entries = os.listdir(os.path.join(book_cache_dir(), 'f')) finished_entries = os.listdir(os.path.join(book_cache_dir(), 'f'))
self.ae(len(finished_entries), 1) self.ae(len(finished_entries), 1)
open(book_src, 'wb').write(b'bcde') set_data('bcde' * 4096)
prepare_book(book_src, convert_func=convert_mock, force_expire=True) npath = prepare_book(book_src, convert_func=convert_mock, force_expire=True)
new_finished_entries = os.listdir(os.path.join(book_cache_dir(), 'f')) new_finished_entries = os.listdir(os.path.join(book_cache_dir(), 'f'))
self.ae(len(new_finished_entries), 1) self.ae(len(new_finished_entries), 1)
self.assertNotEqual(finished_entries, new_finished_entries) self.assertNotEqual(opath, npath)
open(book_src, 'wb').write(b'bcdef') set_data('bcdef')
prepare_book(book_src, convert_func=convert_mock, max_age=-1000, force_expire=True) prepare_book(book_src, convert_func=convert_mock, max_age=-1000, force_expire=True)
self.ae([], os.listdir(os.path.join(book_cache_dir(), 'f'))) self.ae([], os.listdir(os.path.join(book_cache_dir(), 'f')))
with cache_lock() as f: with cache_lock() as f:
@ -358,11 +364,11 @@ def find_tests():
# Test updating cached book # Test updating cached book
book_src = os.path.join(self.tdir, 'book2.epub') book_src = os.path.join(self.tdir, 'book2.epub')
open(book_src, 'wb').write(b'bb') set_data('bb')
path = prepare_book(book_src, convert_func=convert_mock) path = prepare_book(book_src, convert_func=convert_mock)
self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'test') self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'test')
bs = os.stat(book_src) bs = os.stat(book_src)
open(book_src, 'wb').write(b'cde') set_data('cde')
update_book(book_src, bs, name_data_map={'sentinel': b'updated'}) update_book(book_src, bs, name_data_map={'sentinel': b'updated'})
self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'updated') self.ae(open(os.path.join(path, 'sentinel'), 'rb').read(), b'updated')
self.ae(1, len(os.listdir(os.path.join(book_cache_dir(), 'f')))) self.ae(1, len(os.listdir(os.path.join(book_cache_dir(), 'f'))))