Home | History | Annotate | Download | only in wb_kupdate 
      1 #!/usr/bin/python
      2 
      3 import common
      4 import datetime
      5 import logging
      6 import os
      7 import time
      8 import unittest
      9 from autotest_lib.client.bin import test
     10 from autotest_lib.client.bin import utils
     11 from autotest_lib.client.common_lib import error
     12 from autotest_lib.client.common_lib.test_utils import mock
     13 from autotest_lib.client.tests.wb_kupdate import wb_kupdate
     14 
     15 class 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 
     94 class 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 
    104 if __name__ == '__main__':
    105     unittest.main()
    106