1#!/usr/bin/python 2# Copyright 2017 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 logging 7import os 8import shutil 9import subprocess 10import tempfile 11import unittest 12 13import common 14from autotest_lib.client.common_lib import error 15from autotest_lib.client.common_lib import utils 16from autotest_lib.site_utils import lxc 17from autotest_lib.site_utils.lxc import unittest_setup 18 19 20class SharedHostDirTests(unittest.TestCase): 21 """Unit tests for the ContainerBucket class.""" 22 23 def setUp(self): 24 self.tmpdir = tempfile.mkdtemp() 25 self.shared_host_path = os.path.join(self.tmpdir, 'host') 26 27 28 def tearDown(self): 29 shutil.rmtree(self.tmpdir) 30 31 32 def testHostDirCreationAndCleanup(self): 33 """Verifies that the host dir is properly created and cleaned up when 34 the container bucket is set up and destroyed. 35 """ 36 # Precondition: host path nonexistent 37 self.assertFalse(os.path.isdir(self.shared_host_path)) 38 39 host_dir = lxc.SharedHostDir(self.shared_host_path) 40 41 # Verify the host path in the host_dir. 42 self.assertEqual(os.path.realpath(host_dir.path), 43 os.path.realpath(self.shared_host_path)) 44 self.assertTrue(os.path.isdir(self.shared_host_path)) 45 46 # Clean up, verify that the path is removed. 47 host_dir.cleanup() 48 self.assertFalse(os.path.isdir(self.shared_host_path)) 49 50 51 def testHostDirMissing(self): 52 """Verifies that a missing host dir does not cause cleanup to crash. 53 """ 54 host_dir = lxc.SharedHostDir(self.shared_host_path) 55 56 # Manually destroy the host path 57 utils.run('sudo umount %(path)s && sudo rmdir %(path)s' % 58 {'path': self.shared_host_path}) 59 60 # Verify that the host path does not exist. 61 self.assertFalse(os.path.exists(self.shared_host_path)) 62 try: 63 host_dir.cleanup() 64 except: 65 self.fail('SharedHostDir.cleanup crashed.\n%s' % 66 error.format_error()) 67 68 69 def testHostDirNotMounted(self): 70 """Verifies that an unmounted host dir does not cause container bucket 71 construction to crash. 72 """ 73 # Create the shared host dir, but do not mount it. 74 os.makedirs(self.shared_host_path) 75 76 # Setup then destroy the HPM. This should not emit any exceptions. 77 try: 78 host_dir = lxc.SharedHostDir(self.shared_host_path) 79 host_dir.cleanup() 80 except: 81 self.fail('SharedHostDir crashed.\n%s' % error.format_error()) 82 83 84 def testHostDirAccess(self): 85 """Verifies that sudo is not required to write to the shared host dir. 86 """ 87 try: 88 host_dir = lxc.SharedHostDir(self.shared_host_path) 89 tempfile.NamedTemporaryFile(dir=host_dir.path) 90 except OSError: 91 self.fail('Unable to write to shared host dir.\n%s' % 92 error.format_error()) 93 finally: 94 host_dir.cleanup() 95 96 97class TimeoutTests(unittest.TestCase): 98 """Test the timeouts on the shared host dir class.""" 99 100 def setUp(self): 101 self.tmpdir = tempfile.mkdtemp() 102 self.shared_host_path = os.path.join(self.tmpdir, 'host') 103 104 105 def tearDown(self): 106 shutil.rmtree(self.tmpdir) 107 108 109 def testTimeout(self): 110 """Verifies that cleanup code correctly times out. 111 112 Cleanup can fail because of I/O caches and other similar things keeping 113 the mount active. Test that the cleanup code properly times out in 114 these scenarios. 115 """ 116 host_dir = lxc.SharedHostDir(self.shared_host_path) 117 118 # Create a process in the shared dir to force unmounting to fail. 119 p = subprocess.Popen(['sleep', '2'], cwd=self.shared_host_path) 120 121 # Cleanup should time out. 122 with self.assertRaises(error.CmdError): 123 logging.debug('attempting cleanup (should fail)') 124 # Use a short timeout so the test doesn't take forever. 125 host_dir.cleanup(timeout=1) 126 logging.debug('cleanup did not fail') 127 128 # Kill the process occupying the mount. 129 p.terminate() 130 131 # Cleanup should succeed. 132 try: 133 # Use the default timeout so this doesn't raise false errors. 134 host_dir.cleanup() 135 except error.CmdError as e: 136 self.fail('Unexpected cleanup error: %r' % e) 137 138 139if __name__ == '__main__': 140 unittest_setup.setup() 141 unittest.main() 142