1 # Copyright 2015 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 cPickle 6 import datetime 7 import unittest 8 9 import webapp2 10 import webtest 11 12 from google.appengine.ext import ndb 13 14 from dashboard import layered_cache 15 from dashboard import testing_common 16 17 18 class LayeredCacheTest(testing_common.TestCase): 19 20 def setUp(self): 21 super(LayeredCacheTest, self).setUp() 22 app = webapp2.WSGIApplication([( 23 '/delete_expired_entities', 24 layered_cache.DeleteExpiredEntitiesHandler)]) 25 self.testapp = webtest.TestApp(app) 26 self.UnsetCurrentUser() 27 testing_common.SetIsInternalUser('internal (at] chromium.org', True) 28 testing_common.SetIsInternalUser('foo (at] chromium.org', False) 29 30 def testSetAndGet(self): 31 self.SetCurrentUser('internal (at] chromium.org') 32 layered_cache.Set('str', 'Hello, World!') 33 layered_cache.Set('dict', {'hello': [1, 2, 3]}) 34 self.assertEqual( 35 'Hello, World!', 36 cPickle.loads( 37 ndb.Key('CachedPickledString', 'internal_only__str').get().value)) 38 self.assertIsNone( 39 ndb.Key('CachedPickledString', 'externally_visible__str').get()) 40 self.assertEqual('Hello, World!', layered_cache.Get('str')) 41 self.SetCurrentUser('foo (at] chromium.org') 42 self.assertIsNone(layered_cache.Get('str')) 43 self.SetCurrentUser('internal (at] chromium.org') 44 self.assertEqual( 45 {'hello': [1, 2, 3]}, 46 cPickle.loads( 47 ndb.Key('CachedPickledString', 'internal_only__dict').get().value)) 48 self.assertIsNone( 49 ndb.Key('CachedPickledString', 'externally_visible__dict').get()) 50 self.assertEqual({'hello': [1, 2, 3]}, layered_cache.Get('dict')) 51 self.SetCurrentUser('foo (at] chromium.org') 52 self.assertIsNone(layered_cache.Get('dict')) 53 54 def testGetAndSet_External(self): 55 layered_cache.SetExternal('str', 'Hello, World!') 56 layered_cache.SetExternal('dict', {'hello': [1, 2, 3]}) 57 self.assertEqual( 58 'Hello, World!', 59 cPickle.loads( 60 ndb.Key('CachedPickledString', 61 'externally_visible__str').get().value)) 62 self.assertEqual( 63 None, 64 ndb.Key('CachedPickledString', 'internal_only__str').get()) 65 self.assertEqual('Hello, World!', layered_cache.GetExternal('str')) 66 self.assertEqual( 67 {'hello': [1, 2, 3]}, 68 cPickle.loads( 69 ndb.Key('CachedPickledString', 70 'externally_visible__dict').get().value)) 71 self.assertEqual( 72 None, 73 ndb.Key('CachedPickledString', 'internal_only__dict').get()) 74 self.assertEqual({'hello': [1, 2, 3]}, layered_cache.GetExternal('dict')) 75 76 def testDelete(self): 77 self.SetCurrentUser('internal (at] chromium.org') 78 layered_cache.Set('hello', 'secret') 79 self.SetCurrentUser('foo (at] chromium.org') 80 layered_cache.Set('hello', 'not secret') 81 layered_cache.Delete('hello') 82 self.SetCurrentUser('internal (at] chromium.org') 83 self.assertIsNone(layered_cache.Get('hello')) 84 self.SetCurrentUser('foo (at] chromium.org') 85 self.assertIsNone(layered_cache.Get('hello')) 86 87 def testExpireTime(self): 88 self.SetCurrentUser('internal (at] chromium.org') 89 layered_cache.Set('str1', 'Hello, World!', days_to_keep=10) 90 key_internal = ndb.Key('CachedPickledString', 'internal_only__str1') 91 key_external = ndb.Key('CachedPickledString', 'externally_visible__str1') 92 self.assertEqual('Hello, World!', cPickle.loads(key_internal.get().value)) 93 self.assertIsNone(key_external.get()) 94 self.assertEqual('Hello, World!', layered_cache.Get('str1')) 95 96 # The expire date should be 10 days after the current date. 97 actual_date = key_internal.get().expire_time 98 expected_date = datetime.datetime.now() + datetime.timedelta(days=10) 99 self.assertEqual(actual_date.date(), expected_date.date()) 100 101 # When current user is external, the external version is returned by Get. 102 self.SetCurrentUser('foo (at] chromium.org') 103 self.assertIsNone(layered_cache.Get('str1')) 104 105 def testDeleteAllExpiredEntities(self): 106 self.SetCurrentUser('internal (at] chromium.org') 107 layered_cache.Set('expired_str1', 'apple', days_to_keep=-10) 108 layered_cache.Set('expired_str2', 'bat', days_to_keep=-1) 109 layered_cache.Set('expired_str3', 'cat', days_to_keep=10) 110 layered_cache.Set('expired_str4', 'dog', days_to_keep=0) 111 layered_cache.Set('expired_str5', 'egg') 112 self.assertEqual('apple', layered_cache.Get('expired_str1')) 113 self.assertEqual('bat', layered_cache.Get('expired_str2')) 114 self.assertEqual('cat', layered_cache.Get('expired_str3')) 115 self.assertEqual('dog', layered_cache.Get('expired_str4')) 116 self.assertEqual('egg', layered_cache.Get('expired_str5')) 117 layered_cache.DeleteAllExpiredEntities() 118 self.assertIsNone(layered_cache.Get('expired_str1')) 119 self.assertIsNone(layered_cache.Get('expired_str2')) 120 self.assertEqual('cat', layered_cache.Get('expired_str3')) 121 self.assertEqual('dog', layered_cache.Get('expired_str4')) 122 self.assertEqual('egg', layered_cache.Get('expired_str5')) 123 124 def testGet_DeleteExpiredEntities(self): 125 self.SetCurrentUser('internal (at] chromium.org') 126 layered_cache.Set('expired_str1', 'apple', days_to_keep=-10) 127 layered_cache.Set('expired_str2', 'bat', days_to_keep=-1) 128 layered_cache.Set('expired_str3', 'cat', days_to_keep=10) 129 layered_cache.Set('expired_str4', 'dog', days_to_keep=0) 130 layered_cache.Set('expired_str5', 'egg') 131 self.assertEqual('apple', layered_cache.Get('expired_str1')) 132 self.assertEqual('bat', layered_cache.Get('expired_str2')) 133 self.assertEqual('cat', layered_cache.Get('expired_str3')) 134 self.assertEqual('dog', layered_cache.Get('expired_str4')) 135 self.assertEqual('egg', layered_cache.Get('expired_str5')) 136 self.testapp.get('/delete_expired_entities') 137 self.assertIsNone(layered_cache.Get('expired_str1')) 138 self.assertIsNone(layered_cache.Get('expired_str2')) 139 self.assertEqual('cat', layered_cache.Get('expired_str3')) 140 self.assertEqual('dog', layered_cache.Get('expired_str4')) 141 self.assertEqual('egg', layered_cache.Get('expired_str5')) 142 143 144 if __name__ == '__main__': 145 unittest.main() 146