• Home
  • History
  • Annotate
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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