blob: a0bdb645be213f1b1c1a0d63a4f743300dc6a023 [file] [log] [blame]
#!/usr/bin/python
# Copyright (c) 2012 The Native Client Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tests of a local storage cache.
Tests a local disk cache wrapping access to storage with a interface like that
of gsd_storage.GSDStorage.
"""
import os
import unittest
import fake_storage
import file_tools
import local_storage_cache
import working_directory
class TestLocalStorageCache(unittest.TestCase):
def CanBeReadBothWays(self, storage, key, out_file, expected):
# Check that reading key with both GetData and GetFile yields expected.
# out_file is used for GetFile output.
self.assertEquals(expected, storage.GetData(key).decode('utf-8'))
url = storage.GetFile(key, out_file)
self.assertNotEquals(None, url)
self.assertEquals(expected, file_tools.ReadFileString(out_file))
def test_WriteRead(self):
# Check that things written with PutData can be read back.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
storage.PutData('bar', 'foo')
self.CanBeReadBothWays(
storage, 'foo', os.path.join(work_dir, 'out'), 'bar')
def test_WriteFileRead(self):
# Check that things written with PutFile can be read back.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
bar = os.path.join(work_dir, 'bar_file')
file_tools.WriteFile('bar', bar)
storage.PutFile(bar, 'foo')
self.CanBeReadBothWays(
storage, 'foo', os.path.join(work_dir, 'out'), 'bar')
def test_WriteOnlyToLocal(self):
# Check that things written hit local storage, not the network.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
storage.PutData('bar', 'foo')
self.assertEquals(None, mem_storage.GetData('foo'))
bar = os.path.join(work_dir, 'bar_file')
file_tools.WriteFile('bar', bar)
storage.PutFile(bar, 'foo')
self.assertEquals(None, mem_storage.GetData('foo'))
def test_Exists(self):
# Checks that exists works properly.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
storage.PutData('bar', 'foo')
self.assertTrue(storage.Exists('foo'))
self.assertFalse(storage.Exists('bad_foo'))
def test_BadRead(self):
# Check that reading from a non-existant key, fails.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
self.assertEquals(None, storage.GetData('foo'))
def test_HitWrappedStorage(self):
# Check that if something isn't locally cached primary storage is hit.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
mem_storage.PutData('hello', 'foo')
self.assertEquals('hello', storage.GetData('foo'))
def test_HitLocalFirst(self):
# Check that reading hits local storage first.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
storage.PutData('there', 'foo')
mem_storage.PutData('hello', 'foo')
self.assertEquals('there', storage.GetData('foo').decode('utf-8'))
def test_AcceptSlashesAndDots(self):
# Check that keys with slashes and dots are okay.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
storage.PutData('hello', 'this/is/a/cool_test.txt')
self.assertEquals(
'hello', storage.GetData('this/is/a/cool_test.txt').decode('utf-8'))
def test_InvalidKey(self):
# Check that an invalid key asserts.
with working_directory.TemporaryWorkingDirectory() as work_dir:
mem_storage = fake_storage.FakeStorage()
storage = local_storage_cache.LocalStorageCache(
cache_path=os.path.join(work_dir, 'db'),
storage=mem_storage)
bar = os.path.join(work_dir, 'bar_file')
file_tools.WriteFile('bar', bar)
self.assertRaises(KeyError, storage.PutData, 'bar', 'foo$')
self.assertRaises(KeyError, storage.GetData, 'foo^')
self.assertRaises(KeyError, storage.PutFile, bar, 'foo#')
self.assertRaises(KeyError, storage.GetFile, 'foo!', 'bar')
if __name__ == '__main__':
unittest.main()