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