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
6import __builtin__
7import mox
8import os
9import unittest
10import time
11from StringIO import StringIO
12
13import utils
14
15
16class 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
149if __name__ == '__main__':
150    unittest.main()
151