#!/usr/bin/python2 # 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 os import shutil import tempfile 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 BaseImage from autotest_lib.site_utils.lxc import constants from autotest_lib.site_utils.lxc import unittest_setup from autotest_lib.site_utils.lxc import utils as lxc_utils test_dir = None # A reference to an existing base container that can be copied for tests that # need a base container. This is an optimization. reference_container = None # The reference container can either be a reference to an existing container, or # to a container that was downloaded by this test. If the latter, then it needs # to be cleaned up when the tests are complete. cleanup_ref_container = False class BaseImageTests(lxc_utils.LXCTests): """Unit tests to verify the BaseImage class.""" def testCreate_existing(self): """Verifies that BaseImage works with existing base containers.""" with TestBaseContainer() as control: manager = BaseImage(control.container_path, control.name) self.assertIsNotNone(manager.base_container) self.assertEquals(control.container_path, manager.base_container.container_path) self.assertEquals(control.name, manager.base_container.name) try: manager.base_container.refresh_status() except error.ContainerError: self.fail('Base container was not valid.\n%s' % error.format_error()) def testCleanup_noClones(self): """Verifies that cleanup cleans up the base image.""" base = lxc.Container.clone(src=reference_container, new_name=constants.BASE, new_path=test_dir, snapshot=True) manager = BaseImage(base.container_path, base.name) # Precondition: ensure base exists and is a valid container. base.refresh_status() manager.cleanup() # Verify that the base container was cleaned up. self.assertFalse(lxc_utils.path_exists( os.path.join(base.container_path, base.name))) def testCleanup_withClones(self): """Verifies that cleanup cleans up the base image. Ensure that it works even when clones of the base image exist. """ # Do not snapshot, as snapshots of snapshots behave differently than # snapshots of full container clones. BaseImage cleanup code assumes # that the base container is not a snapshot. base = lxc.Container.clone(src=reference_container, new_name=constants.BASE, new_path=test_dir, snapshot=False) manager = BaseImage(base.container_path, base.name) clones = [] for i in range(3): clones.append(lxc.Container.clone(src=base, new_name='clone_%d' % i, snapshot=True)) # Precondition: all containers are valid. base.refresh_status() for container in clones: container.refresh_status() manager.cleanup() # Verify that all containers were cleaned up self.assertFalse(lxc_utils.path_exists( os.path.join(base.container_path, base.name))) for container in clones: if constants.SUPPORT_SNAPSHOT_CLONE: # Snapshot clones should get deleted along with the base # container. self.assertFalse(lxc_utils.path_exists( os.path.join(container.container_path, container.name))) else: # If snapshot clones aren't supported (e.g. on moblab), the # clones should not be affected by the destruction of the base # container. try: container.refresh_status() except error.ContainerError: self.fail(error.format_error()) class BaseImageSetupTests(lxc_utils.LXCTests): """Unit tests to verify the setup of specific images. Some images differ in layout from others. These tests test specific images to make sure the setup code works for all of them. """ def setUp(self): self.manager = BaseImage(container_path=test_dir) def tearDown(self): self.manager.cleanup() def testSetupBase05(self): """Verifies that setup works for moblab base container. Verifies that the code for installing the rootfs location into the lxc config, is working correctly. """ # Set up the bucket, then start the base container, and verify it works. self.manager.setup('base_05') container = self.manager.base_container container.start(wait_for_network=False) self.assertTrue(container.is_running()) @unittest.skipIf(constants.IS_MOBLAB, "Moblab does not support the regular base container.") def testSetupBase09(self): """Verifies that setup works for base container. Verifies that the code for installing the rootfs location into the lxc config, is working correctly. """ self.manager.setup('base_09') container = self.manager.base_container container.start(wait_for_network=False) self.assertTrue(container.is_running()) @contextmanager def TestBaseContainer(name=constants.BASE): """Context manager for creating a scoped base container for testing. @param name: (optional) Name of the base container. If this is not provided, the default base container name is used. """ container = lxc.Container.clone(src=reference_container, new_name=name, new_path=test_dir, snapshot=True, cleanup=False) try: yield container finally: if not unittest_setup.config.skip_cleanup: container.destroy() def setUpModule(): """Module setup for base image unittests. Sets up a test directory along with a reference container that is used by tests that need an existing base container. """ global test_dir global reference_container global cleanup_ref_container test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH, prefix='base_container_manager_unittest_') # Unfortunately, aside from duping the BaseImage code completely, there # isn't an easy way to download and configure a base container. So even # though this is the BaseImage unittest, we use a BaseImage to set it up. bcm = BaseImage() if bcm.base_container is None: bcm.setup() cleanup_ref_container = True reference_container = bcm.base_container def tearDownModule(): """Deletes the test dir and reference container.""" if not unittest_setup.config.skip_cleanup: if cleanup_ref_container: reference_container.destroy() shutil.rmtree(test_dir) if __name__ == '__main__': unittest.main()