• Home
  • 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 argparse
7import logging
8import os
9import tempfile
10import shutil
11import sys
12import unittest
13from contextlib import contextmanager
14
15import common
16from autotest_lib.client.common_lib import error
17from autotest_lib.site_utils import lxc
18from autotest_lib.site_utils.lxc import constants
19from autotest_lib.site_utils.lxc import unittest_http
20from autotest_lib.site_utils.lxc import unittest_logging
21from autotest_lib.site_utils.lxc import utils as lxc_utils
22from autotest_lib.site_utils.lxc.unittest_container_bucket \
23        import FastContainerBucket
24
25options = None
26
27class ContainerTests(unittest.TestCase):
28    """Unit tests for the Container class."""
29
30    @classmethod
31    def setUpClass(cls):
32        cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH,
33                                        prefix='container_unittest_')
34        cls.shared_host_path = os.path.join(cls.test_dir, 'host')
35
36        # Use a container bucket just to download and set up the base image.
37        cls.bucket = FastContainerBucket(cls.test_dir, cls.shared_host_path)
38
39        if cls.bucket.base_container is None:
40            logging.debug('Base container not found - reinitializing')
41            cls.bucket.setup_base()
42        else:
43            logging.debug('base container found')
44        cls.base_container = cls.bucket.base_container
45        assert(cls.base_container is not None)
46
47
48    @classmethod
49    def tearDownClass(cls):
50        cls.base_container = None
51        if not options.skip_cleanup:
52            cls.bucket.destroy_all()
53            shutil.rmtree(cls.test_dir)
54
55    def tearDown(self):
56        # Ensure host dirs from each test are completely destroyed.
57        for host_dir in os.listdir(self.shared_host_path):
58            host_dir = os.path.realpath(os.path.join(self.shared_host_path,
59                                                     host_dir))
60            lxc_utils.cleanup_host_mount(host_dir);
61
62
63    def testInit(self):
64        """Verifies that containers initialize correctly."""
65        # Make a container that just points to the base container.
66        container = lxc.Container.createFromExistingDir(
67            self.base_container.container_path,
68            self.base_container.name)
69        self.assertFalse(container.is_running())
70
71
72    def testInitInvalid(self):
73        """Verifies that invalid containers can still be instantiated,
74        if not used.
75        """
76        with tempfile.NamedTemporaryFile(dir=self.test_dir) as tmpfile:
77            name = os.path.basename(tmpfile.name)
78            container = lxc.Container.createFromExistingDir(self.test_dir, name)
79            with self.assertRaises(error.ContainerError):
80                container.refresh_status()
81
82
83    def testDefaultHostname(self):
84        """Verifies that the zygote starts up with a default hostname that is
85        the lxc container name."""
86        test_name = 'testHostname'
87        with self.createContainer(name=test_name) as container:
88            container.start(wait_for_network=True)
89            hostname = container.attach_run('hostname').stdout.strip()
90            self.assertEqual(test_name, hostname)
91
92
93    @unittest.skip('Setting the container hostname using lxc.utsname does not'
94                   'work on goobuntu.')
95    def testSetHostnameNotRunning(self):
96        """Verifies that the hostname can be set on a stopped container."""
97        with self.createContainer() as container:
98            expected_hostname = 'my-new-hostname'
99            container.set_hostname(expected_hostname)
100            container.start(wait_for_network=True)
101            hostname = container.attach_run('hostname').stdout.strip()
102            self.assertEqual(expected_hostname, hostname)
103
104
105    def testClone(self):
106        """Verifies that cloning a container works as expected."""
107        clone = lxc.Container.clone(src=self.base_container,
108                                    new_name="testClone",
109                                    snapshot=True)
110        try:
111            # Throws an exception if the container is not valid.
112            clone.refresh_status()
113        finally:
114            clone.destroy()
115
116
117    def testCloneWithoutCleanup(self):
118        """Verifies that cloning a container to an existing name will fail as
119        expected.
120        """
121        lxc.Container.clone(src=self.base_container,
122                            new_name="testCloneWithoutCleanup",
123                            snapshot=True)
124        with self.assertRaises(error.ContainerError):
125            lxc.Container.clone(src=self.base_container,
126                                new_name="testCloneWithoutCleanup",
127                                snapshot=True)
128
129
130    def testCloneWithCleanup(self):
131        """Verifies that cloning a container with cleanup works properly."""
132        clone0 = lxc.Container.clone(src=self.base_container,
133                                     new_name="testClone",
134                                     snapshot=True)
135        clone0.start(wait_for_network=False)
136        tmpfile = clone0.attach_run('mktemp').stdout
137        # Verify that our tmpfile exists
138        clone0.attach_run('test -f %s' % tmpfile)
139
140        # Clone another container in place of the existing container.
141        clone1 = lxc.Container.clone(src=self.base_container,
142                                     new_name="testClone",
143                                     snapshot=True,
144                                     cleanup=True)
145        with self.assertRaises(error.CmdError):
146            clone1.attach_run('test -f %s' % tmpfile)
147
148
149    def testInstallSsp(self):
150        """Verifies that installing the ssp in the container works."""
151        # Hard-coded path to some golden data for this test.
152        test_ssp = os.path.join(
153                common.autotest_dir,
154                'site_utils', 'lxc', 'test', 'test_ssp.tar.bz2')
155        # Create a container, install the self-served ssp, then check that it is
156        # installed into the container correctly.
157        with self.createContainer() as container:
158            with unittest_http.serve_locally(test_ssp) as url:
159                container.install_ssp(url)
160            container.start(wait_for_network=False)
161
162            # The test ssp just contains a couple of text files, in known
163            # locations.  Verify the location and content of those files in the
164            # container.
165            cat = lambda path: container.attach_run('cat %s' % path).stdout
166            test0 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR,
167                                     'test.0'))
168            test1 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR,
169                                     'dir0', 'test.1'))
170            self.assertEquals('the five boxing wizards jumped quickly',
171                              test0)
172            self.assertEquals('the quick brown fox jumps over the lazy dog',
173                              test1)
174
175
176    def testInstallControlFile(self):
177        """Verifies that installing a control file in the container works."""
178        _unused, tmpfile = tempfile.mkstemp()
179        with self.createContainer() as container:
180            container.install_control_file(tmpfile)
181            container.start(wait_for_network=False)
182            # Verify that the file is found in the container.
183            container.attach_run(
184                'test -f %s' % os.path.join(lxc.CONTROL_TEMP_PATH,
185                                            os.path.basename(tmpfile)))
186
187
188    @contextmanager
189    def createContainer(self, name=None):
190        """Creates a container from the base container, for testing.
191        Use this to ensure that containers get properly cleaned up after each
192        test.
193
194        @param name: An optional name for the new container.
195        """
196        if name is None:
197            name = self.id().split('.')[-1]
198        container = self.bucket.create_from_base(name)
199        try:
200            yield container
201        finally:
202            container.destroy()
203
204
205def parse_options():
206    """Parse command line inputs.
207    """
208    parser = argparse.ArgumentParser()
209    parser.add_argument('-v', '--verbose', action='store_true',
210                        help='Print out ALL entries.')
211    parser.add_argument('--skip_cleanup', action='store_true',
212                        help='Skip deleting test containers.')
213    args, argv = parser.parse_known_args()
214
215    # Hack: python unittest also processes args.  Construct an argv to pass to
216    # it, that filters out the options it won't recognize.
217    if args.verbose:
218        argv.insert(0, '-v')
219    argv.insert(0, sys.argv[0])
220
221    return args, argv
222
223
224if __name__ == '__main__':
225    options, unittest_argv = parse_options()
226
227    log_level=(logging.DEBUG if options.verbose else logging.INFO)
228    unittest_logging.setup(log_level)
229
230    unittest.main(argv=unittest_argv)
231