client/cros/rf/config_unittest.py.disabled

# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

if __name__ == '__main__':
    import os, sys
    sys.path.append(
        os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
    import common


import tempfile
import thread
import time
import unittest
import os

# Set the factory log root so that we can watch it.
log_root = os.environ['CROS_FACTORY_LOG_ROOT'] = tempfile.mkdtemp()
console_log_path = os.path.join(log_root, "console.log")

from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.rf.config import PluggableConfig


TIMEOUT_SEC = 3


class PluggableConfigTestCase(unittest.TestCase):
    def assertInRange(self, value, min, max):
        assert min <= max
        self.assertTrue(value >= min and value <= max,
                        '%s not in range [%s,%s]' % (value, min, max))

    def testPathExists(self):
        tmp = tempfile.NamedTemporaryFile()
        print >>tmp, '12345'
        tmp.flush()

        self.assertEqual(12345,
                         PluggableConfig('').Read(config_path=tmp.name))

        self.assertEqual(
            ["[INFO] Waiting for test configuration file %r...\n" % tmp.name,
             "[INFO] Read test configuration file %r\n" % tmp.name],
            open(console_log_path).readlines()[-2:])

    def testPathDoesntExist(self):
        start = time.time()
        self.assertRaises(
            utils.TimeoutError,
            PluggableConfig('').Read,
            '/file-that-does-not-exist', timeout=TIMEOUT_SEC)
        self.assertInRange(time.time() - start,
                           TIMEOUT_SEC - 1, TIMEOUT_SEC + 1)

    def testWaitForPath(self):
        tmp_fd, tmp_path = tempfile.mkstemp()
        os.close(tmp_fd)
        os.unlink(tmp_path)

        delay = TIMEOUT_SEC / 2.0
        def CreateFile():
            time.sleep(delay)
            with open(tmp_path, "w") as tmp:
                print >>tmp, '12345'

        thread.start_new_thread(CreateFile, ())

        start = time.time()
        self.assertEqual(
            12345,
            PluggableConfig('').Read(config_path=tmp_path, timeout=TIMEOUT_SEC))
        # Check that 'delay' seconds have passed, +/-1 second.
        self.assertInRange(time.time() - start,
                           delay - 1, delay + 1)


if __name__ == "__main__":
    unittest.main()