1#!/usr/bin/python 2 3import common 4import datetime 5import logging 6import os 7import time 8import unittest 9from autotest_lib.client.bin import test 10from autotest_lib.client.bin import utils 11from autotest_lib.client.common_lib import error 12from autotest_lib.client.common_lib.test_utils import mock 13from autotest_lib.client.tests.wb_kupdate import wb_kupdate 14 15class WbKupdateUnitTest(unittest.TestCase): 16 def setUp(self): 17 """Set up all required variables for the Unittest. 18 """ 19 self._logger = logging.getLogger() 20 self._wbkupdate_obj = WbKupdateSubclass() 21 self._god = mock.mock_god() 22 23 def test_needs_more_time(self): 24 """Tests the _needs_more_time method. 25 """ 26 self._logger.info('Testing the "_needs_more_time" method.') 27 28 # Obvious failure - since start_time < start_time + 1. 29 self.assertTrue(self._wbkupdate_obj._needs_more_time( 30 start_time=datetime.datetime.now(), 31 duration=1)) 32 33 # Check if 1 minute has elapsed since start_time. 34 self.assertFalse(self._wbkupdate_obj._needs_more_time( 35 start_time=datetime.datetime.now(), 36 duration=1, 37 _now=datetime.datetime.now() + datetime.timedelta(seconds=60))) 38 39 def test_wait_until_data_flushed_pass(self): 40 """Tests the _wait_until_data_flushed method. 41 42 This tests the "success" code path. 43 """ 44 self._logger.info('Testing the "_wait_until_data_flushed" method - ' 45 'Success code path.') 46 47 # Creating stubs for required methods. 48 self._god.stub_function(self._wbkupdate_obj, 49 "_get_disk_usage") 50 51 # Setting default return values for stub functions. 52 # Setting the initial size of the file. 53 self._wbkupdate_obj._get_disk_usage.expect_call('').and_return(10) 54 # Returning the same file size - forcing code path to enter loop. 55 self._wbkupdate_obj._get_disk_usage.expect_call('').and_return(10) 56 # Returning a greater file size - exiting the while loop. 57 self._wbkupdate_obj._get_disk_usage.expect_call('').and_return(11) 58 59 # Call the method. 60 self._wbkupdate_obj._wait_until_data_flushed(datetime.datetime.now(), 61 1) 62 63 # Ensure all stubbed methods called. 64 self._god.check_playback() 65 66 67 def test_wait_until_data_flushed_fail(self): 68 """Tests the _wait_until_data_flushed method. 69 70 This tests the "failure" code path. 71 """ 72 self._logger.info('Testing the "_wait_until_data_flushed" method - ' 73 'Failure code path.') 74 # Creating stubs for required methods. 75 self._god.stub_function(self._wbkupdate_obj, 76 "_get_disk_usage") 77 78 # Setting default return values for stub functions. 79 # Setting the initial size of the file. 80 self._wbkupdate_obj._get_disk_usage.expect_call('').and_return(10) 81 # Returning the same file size - forcing code path to enter loop. 82 self._wbkupdate_obj._get_disk_usage.expect_call('').and_return(10) 83 84 # Call the method. 85 self.assertRaises(error.TestError, 86 self._wbkupdate_obj._wait_until_data_flushed, 87 start_time=datetime.datetime.now(), 88 max_wait_time=0) 89 90 # Ensure all stubbed methods called. 91 self._god.check_playback() 92 93 94class WbKupdateSubclass(wb_kupdate.wb_kupdate): 95 """Sub-classing the wb_kupdate class. 96 """ 97 def __init__(self): 98 """Empty constructor. 99 """ 100 # Create all test defaults. 101 self.initialize() 102 103 104if __name__ == '__main__': 105 unittest.main() 106