site_utils/lxc/container_factory_unittest.py

#!/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 tempfile
import unittest

import common
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.site_utils import lxc
from autotest_lib.site_utils.lxc import unittest_setup
from autotest_lib.site_utils.lxc import utils as lxc_utils


class ContainerFactoryTests(lxc_utils.LXCTests):
    """Unit tests for the ContainerFactory class."""

    @classmethod
    def setUpClass(cls):
        super(ContainerFactoryTests, cls).setUpClass()
        cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH,
                                        prefix='container_factory_unittest_')

        # Check if a base container exists on this machine and download one if
        # necessary.
        image = lxc.BaseImage()
        try:
            cls.base_container = image.get()
            cls.cleanup_base_container = False
        except error.ContainerError:
            image.setup()
            cls.base_container = image.get()
            cls.cleanup_base_container = True
        assert(cls.base_container is not None)


    @classmethod
    def tearDownClass(cls):
        cls.base_container = None
        if not unittest_setup.config.skip_cleanup:
            if cls.cleanup_base_container:
                lxc.BaseImage().cleanup()
            utils.run('sudo rm -r %s' % cls.test_dir)


    def setUp(self):
        # Create a separate dir for each test, so they are hermetic.
        self.test_dir = tempfile.mkdtemp(dir=ContainerFactoryTests.test_dir)
        self.test_factory = lxc.ContainerFactory(
                base_container=self.base_container,
                lxc_path=self.test_dir)


    def testCreateContainer(self):
        """Tests basic container creation."""
        container = self.test_factory.create_container()

        try:
            container.refresh_status()
        except:
            self.fail('Invalid container:\n%s' % error.format_error())


    def testCreateContainer_noId(self):
        """Tests container creation with default IDs."""
        container = self.test_factory.create_container()
        self.assertIsNone(container.id)


    def testCreateContainer_withId(self):
        """Tests container creation with given IDs. """
        id0 = lxc.ContainerId(1, 2, 3)
        container = self.test_factory.create_container(id0)
        self.assertEquals(id0, container.id)


    def testContainerName(self):
        """Tests that created containers have the right name."""
        id0 = lxc.ContainerId(1, 2, 3)
        id1 = lxc.ContainerId(42, 41, 40)

        container0 = self.test_factory.create_container(id0)
        container1 = self.test_factory.create_container(id1)

        self.assertEqual(str(id0), container0.name)
        self.assertEqual(str(id1), container1.name)


    def testContainerPath(self):
        """Tests that created containers have the right LXC path."""
        dir0 = tempfile.mkdtemp(dir=self.test_dir)
        dir1 = tempfile.mkdtemp(dir=self.test_dir)

        container0 = self.test_factory.create_container(lxc_path=dir0)
        container1 = self.test_factory.create_container(lxc_path=dir1)

        self.assertEqual(dir0, container0.container_path);
        self.assertEqual(dir1, container1.container_path);


    def testCreateContainer_alreadyExists(self):
        """Tests that container ID conflicts raise errors as expected."""
        id0 = lxc.ContainerId(1, 2, 3)

        self.test_factory.create_container(id0)
        with self.assertRaises(error.ContainerError):
            self.test_factory.create_container(id0)


    def testCreateContainer_forceReset(self):
        """Tests that force-resetting containers works."""
        factory = lxc.ContainerFactory(base_container=self.base_container,
                                       lxc_path=self.test_dir,
                                       force_cleanup=True)

        id0 = lxc.ContainerId(1, 2, 3)
        container0 = factory.create_container(id0)
        container0.start(wait_for_network=False)

        # Create a file in the original container.
        tmpfile = container0.attach_run('mktemp').stdout
        exists = 'test -e %s' % tmpfile
        try:
            container0.attach_run(exists)
        except error.CmdError as e:
            self.fail(e)

        # Create a new container in place of the original, then verify that the
        # file is no longer there.
        container1 = factory.create_container(id0)
        container1.start(wait_for_network=False)
        with self.assertRaises(error.CmdError):
            container1.attach_run(exists)


    def testCreateContainer_subclass(self):
        """Tests that the factory produces objects of the requested class."""
        container = self.test_factory.create_container()
        # Don't use isinstance, we want to check the exact type.
        self.assertTrue(type(container) is lxc.Container)

        class _TestContainer(lxc.Container):
            """A test Container subclass"""
            pass

        test_factory = lxc.ContainerFactory(base_container=self.base_container,
                                            container_class=_TestContainer,
                                            lxc_path=self.test_dir)
        test_container = test_factory.create_container()
        self.assertTrue(type(test_container) is _TestContainer)


    def testCreateContainer_snapshotFails(self):
        """Tests the scenario where snapshotting fails.

        Verifies that the factory is still able to produce a Container when
        cloning fails.
        """
        class MockContainerClass(object):
            """A mock object to simulate the container class.

            This mock has a clone method that simulates a failure when clone is
            called with snapshot=True.  Clone calls are recorded so they can be
            verified later.
            """
            def __init__(self):
                """Initializes the mock."""
                self.clone_count = 0
                self.clone_kwargs = []


            def clone(self, *args, **kwargs):
                """Mocks the Container.clone class method. """
                # Record the particulars of this call.
                self.clone_count += 1
                self.clone_kwargs.append(kwargs)
                # Simulate failure if a snapshot is requested, otherwise create
                # and return the clone.
                if kwargs['snapshot']:
                    raise error.CmdError('fake error', None)
                else:
                    return lxc.Container.clone(*args, **kwargs)

        mock = MockContainerClass()
        factory = lxc.ContainerFactory(base_container=self.base_container,
                                       container_class=mock,
                                       snapshot=True,
                                       lxc_path=self.test_dir)

        factory.create_container()
        # The factory should have made 2 calls to mock.clone - the first with
        # snapshot=True, then the second with snapshot=False.
        self.assertEquals(2, mock.clone_count)
        self.assertTrue(mock.clone_kwargs[0]['snapshot'])
        self.assertFalse(mock.clone_kwargs[1]['snapshot'])


if __name__ == '__main__':
    unittest.main()