1 #!/usr/bin/python 2 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 6 import __builtin__ 7 import mox 8 import os 9 import unittest 10 import time 11 from StringIO import StringIO 12 13 import utils 14 15 16 class TestUtils(mox.MoxTestBase): 17 """Test utility functions.""" 18 19 20 def test_load_servo_interface_mapping(self): 21 """Test servo-interface mapping file can be loaded.""" 22 self.mox.StubOutWithMock(__builtin__, 'open') 23 fake_content = ( 24 'chromeos1-rack5-host10-servo, chromeos1-poe-switch1, fa42\n' 25 'chromeos1-rack5-host11-servo, chromeos1-poe-switch1, fa43\n' 26 ', chromeos2-poe-switch8, fa43\n' 27 'chromeos2-rack5-host11-servo, chromeos2-poe-switch8, fa44\n') 28 fake_file = self.mox.CreateMockAnything() 29 fake_file.__enter__().AndReturn(StringIO(fake_content)) 30 fake_file.__exit__(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) 31 open('fake_file.csv').AndReturn(fake_file) 32 expect = {'chromeos1-rack5-host10-servo': 33 ('chromeos1-poe-switch1', 'fa42'), 34 'chromeos1-rack5-host11-servo': 35 ('chromeos1-poe-switch1', 'fa43'), 36 'chromeos2-rack5-host11-servo': 37 ('chromeos2-poe-switch8', 'fa44')} 38 self.mox.ReplayAll() 39 self.assertEqual( 40 utils.load_servo_interface_mapping('fake_file.csv'), expect) 41 self.mox.VerifyAll() 42 43 44 def _reload_helper(self, do_reload): 45 """Helper class for mapping file reloading tests.""" 46 self.mox.StubOutWithMock(utils, 'load_servo_interface_mapping') 47 self.mox.StubOutWithMock(os.path, 'getmtime') 48 check_point = 1369783561.8525634 49 if do_reload: 50 last_modified = check_point + 10.0 51 servo_interface = {'fake_servo': ('fake_switch', 'fake_if')} 52 utils.load_servo_interface_mapping('fake_file').AndReturn( 53 servo_interface) 54 else: 55 last_modified = check_point 56 os.path.getmtime(mox.IgnoreArg()).AndReturn(last_modified) 57 self.mox.ReplayAll() 58 result = utils.reload_servo_interface_mapping_if_necessary( 59 check_point, mapping_file='fake_file') 60 if do_reload: 61 self.assertEqual(result, (last_modified, servo_interface)) 62 else: 63 self.assertIsNone(result) 64 self.mox.VerifyAll() 65 66 67 def test_reload_servo_interface_mapping_necessary(self): 68 """Test that mapping file is reloaded when it is modified.""" 69 self._reload_helper(True) 70 71 72 def test_reload_servo_interface_mapping_not_necessary(self): 73 """Test that mapping file is not reloaded when it is not modified.""" 74 self._reload_helper(False) 75 76 77 def test_LRU_cache(self): 78 """Test LRUCache.""" 79 p1 = utils.PowerUnitInfo( 80 'host1', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 81 'rpm1', 'hydra1') 82 p2 = utils.PowerUnitInfo( 83 'host2', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 84 'rpm2', 'hydra2') 85 p3 = utils.PowerUnitInfo( 86 'host3', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 87 'rpm3', 'hydra3') 88 # Initialize an LRU with size 2, items never expire. 89 cache = utils.LRUCache(2, expiration_secs=None) 90 # Add two items, LRU should be full now 91 cache['host1'] = p1 92 cache['host2'] = p2 93 self.assertEqual(len(cache.cache), 2) 94 # Visit host2 and add one more item 95 # host1 should be removed from cache 96 _ = cache['host2'] 97 cache['host3'] = p3 98 self.assertEqual(len(cache.cache), 2) 99 self.assertTrue('host1' not in cache) 100 self.assertTrue('host2' in cache) 101 self.assertTrue('host3' in cache) 102 103 104 def test_LRU_cache_expires(self): 105 """Test LRUCache expires.""" 106 self.mox.StubOutWithMock(time, 'time') 107 time.time().AndReturn(10) 108 time.time().AndReturn(25) 109 p1 = utils.PowerUnitInfo( 110 'host1', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 111 'rpm1', 'hydra1') 112 113 self.mox.ReplayAll() 114 # Initialize an LRU with size 1, items exppire after 10 secs. 115 cache = utils.LRUCache(1, expiration_secs=10) 116 # Add two items, LRU should be full now 117 cache['host1'] = p1 118 check_contains_1 = 'host1' in cache 119 check_contains_2 = 'host2' in cache 120 self.mox.VerifyAll() 121 self.assertFalse(check_contains_1) 122 self.assertFalse(check_contains_2) 123 124 125 def test_LRU_cache_full_with_expries(self): 126 """Test timestamp is removed properly when cache is full.""" 127 self.mox.StubOutWithMock(time, 'time') 128 time.time().AndReturn(10) 129 time.time().AndReturn(25) 130 p1 = utils.PowerUnitInfo( 131 'host1', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 132 'rpm1', 'hydra1') 133 p2 = utils.PowerUnitInfo( 134 'host2', utils.PowerUnitInfo.POWERUNIT_TYPES.RPM, 135 'rpm2', 'hydra2') 136 self.mox.ReplayAll() 137 # Initialize an LRU with size 1, items expire after 10 secs. 138 cache = utils.LRUCache(1, expiration_secs=10) 139 # Add two items, LRU should be full now 140 cache['host1'] = p1 141 cache['host2'] = p2 142 self.mox.VerifyAll() 143 self.assertEqual(len(cache.timestamps), 1) 144 self.assertEqual(len(cache.cache), 1) 145 self.assertTrue('host2' in cache.timestamps) 146 self.assertTrue('host2' in cache.cache) 147 148 149 if __name__ == '__main__': 150 unittest.main() 151
