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