| #!/usr/bin/python |
| # Copyright 2017 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. |
| |
| import argparse |
| import logging |
| import os |
| import tempfile |
| import shutil |
| import sys |
| import unittest |
| from contextlib import contextmanager |
| |
| import common |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.site_utils import lxc |
| from autotest_lib.site_utils.lxc import constants |
| from autotest_lib.site_utils.lxc import unittest_http |
| from autotest_lib.site_utils.lxc import unittest_logging |
| from autotest_lib.site_utils.lxc import utils as lxc_utils |
| from autotest_lib.site_utils.lxc.unittest_container_bucket \ |
| import FastContainerBucket |
| |
| options = None |
| |
| class ContainerTests(unittest.TestCase): |
| """Unit tests for the Container class.""" |
| |
| @classmethod |
| def setUpClass(cls): |
| cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH, |
| prefix='container_unittest_') |
| cls.shared_host_path = os.path.join(cls.test_dir, 'host') |
| |
| # Use a container bucket just to download and set up the base image. |
| cls.bucket = FastContainerBucket(cls.test_dir, cls.shared_host_path) |
| |
| if cls.bucket.base_container is None: |
| logging.debug('Base container not found - reinitializing') |
| cls.bucket.setup_base() |
| else: |
| logging.debug('base container found') |
| cls.base_container = cls.bucket.base_container |
| assert(cls.base_container is not None) |
| |
| |
| @classmethod |
| def tearDownClass(cls): |
| cls.base_container = None |
| if not options.skip_cleanup: |
| cls.bucket.destroy_all() |
| shutil.rmtree(cls.test_dir) |
| |
| def tearDown(self): |
| # Ensure host dirs from each test are completely destroyed. |
| for host_dir in os.listdir(self.shared_host_path): |
| host_dir = os.path.realpath(os.path.join(self.shared_host_path, |
| host_dir)) |
| lxc_utils.cleanup_host_mount(host_dir); |
| |
| |
| def testInit(self): |
| """Verifies that containers initialize correctly.""" |
| # Make a container that just points to the base container. |
| container = lxc.Container.createFromExistingDir( |
| self.base_container.container_path, |
| self.base_container.name) |
| self.assertFalse(container.is_running()) |
| |
| |
| def testInitInvalid(self): |
| """Verifies that invalid containers can still be instantiated, |
| if not used. |
| """ |
| with tempfile.NamedTemporaryFile(dir=self.test_dir) as tmpfile: |
| name = os.path.basename(tmpfile.name) |
| container = lxc.Container.createFromExistingDir(self.test_dir, name) |
| with self.assertRaises(error.ContainerError): |
| container.refresh_status() |
| |
| |
| def testDefaultHostname(self): |
| """Verifies that the zygote starts up with a default hostname that is |
| the lxc container name.""" |
| test_name = 'testHostname' |
| with self.createContainer(name=test_name) as container: |
| container.start(wait_for_network=True) |
| hostname = container.attach_run('hostname').stdout.strip() |
| self.assertEqual(test_name, hostname) |
| |
| |
| @unittest.skip('Setting the container hostname using lxc.utsname does not' |
| 'work on goobuntu.') |
| def testSetHostnameNotRunning(self): |
| """Verifies that the hostname can be set on a stopped container.""" |
| with self.createContainer() as container: |
| expected_hostname = 'my-new-hostname' |
| container.set_hostname(expected_hostname) |
| container.start(wait_for_network=True) |
| hostname = container.attach_run('hostname').stdout.strip() |
| self.assertEqual(expected_hostname, hostname) |
| |
| |
| def testClone(self): |
| """Verifies that cloning a container works as expected.""" |
| clone = lxc.Container.clone(src=self.base_container, |
| new_name="testClone", |
| snapshot=True) |
| try: |
| # Throws an exception if the container is not valid. |
| clone.refresh_status() |
| finally: |
| clone.destroy() |
| |
| |
| def testCloneWithoutCleanup(self): |
| """Verifies that cloning a container to an existing name will fail as |
| expected. |
| """ |
| lxc.Container.clone(src=self.base_container, |
| new_name="testCloneWithoutCleanup", |
| snapshot=True) |
| with self.assertRaises(error.ContainerError): |
| lxc.Container.clone(src=self.base_container, |
| new_name="testCloneWithoutCleanup", |
| snapshot=True) |
| |
| |
| def testCloneWithCleanup(self): |
| """Verifies that cloning a container with cleanup works properly.""" |
| clone0 = lxc.Container.clone(src=self.base_container, |
| new_name="testClone", |
| snapshot=True) |
| clone0.start(wait_for_network=False) |
| tmpfile = clone0.attach_run('mktemp').stdout |
| # Verify that our tmpfile exists |
| clone0.attach_run('test -f %s' % tmpfile) |
| |
| # Clone another container in place of the existing container. |
| clone1 = lxc.Container.clone(src=self.base_container, |
| new_name="testClone", |
| snapshot=True, |
| cleanup=True) |
| with self.assertRaises(error.CmdError): |
| clone1.attach_run('test -f %s' % tmpfile) |
| |
| |
| def testInstallSsp(self): |
| """Verifies that installing the ssp in the container works.""" |
| # Hard-coded path to some golden data for this test. |
| test_ssp = os.path.join( |
| common.autotest_dir, |
| 'site_utils', 'lxc', 'test', 'test_ssp.tar.bz2') |
| # Create a container, install the self-served ssp, then check that it is |
| # installed into the container correctly. |
| with self.createContainer() as container: |
| with unittest_http.serve_locally(test_ssp) as url: |
| container.install_ssp(url) |
| container.start(wait_for_network=False) |
| |
| # The test ssp just contains a couple of text files, in known |
| # locations. Verify the location and content of those files in the |
| # container. |
| cat = lambda path: container.attach_run('cat %s' % path).stdout |
| test0 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, |
| 'test.0')) |
| test1 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, |
| 'dir0', 'test.1')) |
| self.assertEquals('the five boxing wizards jumped quickly', |
| test0) |
| self.assertEquals('the quick brown fox jumps over the lazy dog', |
| test1) |
| |
| |
| def testInstallControlFile(self): |
| """Verifies that installing a control file in the container works.""" |
| _unused, tmpfile = tempfile.mkstemp() |
| with self.createContainer() as container: |
| container.install_control_file(tmpfile) |
| container.start(wait_for_network=False) |
| # Verify that the file is found in the container. |
| container.attach_run( |
| 'test -f %s' % os.path.join(lxc.CONTROL_TEMP_PATH, |
| os.path.basename(tmpfile))) |
| |
| |
| @contextmanager |
| def createContainer(self, name=None): |
| """Creates a container from the base container, for testing. |
| Use this to ensure that containers get properly cleaned up after each |
| test. |
| |
| @param name: An optional name for the new container. |
| """ |
| if name is None: |
| name = self.id().split('.')[-1] |
| container = self.bucket.create_from_base(name) |
| try: |
| yield container |
| finally: |
| container.destroy() |
| |
| |
| def parse_options(): |
| """Parse command line inputs. |
| """ |
| parser = argparse.ArgumentParser() |
| parser.add_argument('-v', '--verbose', action='store_true', |
| help='Print out ALL entries.') |
| parser.add_argument('--skip_cleanup', action='store_true', |
| help='Skip deleting test containers.') |
| args, argv = parser.parse_known_args() |
| |
| # Hack: python unittest also processes args. Construct an argv to pass to |
| # it, that filters out the options it won't recognize. |
| if args.verbose: |
| argv.insert(0, '-v') |
| argv.insert(0, sys.argv[0]) |
| |
| return args, argv |
| |
| |
| if __name__ == '__main__': |
| options, unittest_argv = parse_options() |
| |
| log_level=(logging.DEBUG if options.verbose else logging.INFO) |
| unittest_logging.setup(log_level) |
| |
| unittest.main(argv=unittest_argv) |