• 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 tempfile
7import unittest
8
9import common
10from autotest_lib.client.bin import utils
11from autotest_lib.client.common_lib import error
12from autotest_lib.site_utils import lxc
13from autotest_lib.site_utils.lxc import unittest_setup
14from autotest_lib.site_utils.lxc import utils as lxc_utils
15
16
17class ContainerFactoryTests(lxc_utils.LXCTests):
18    """Unit tests for the ContainerFactory class."""
19
20    @classmethod
21    def setUpClass(cls):
22        super(ContainerFactoryTests, cls).setUpClass()
23        cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH,
24                                        prefix='container_factory_unittest_')
25
26        # Check if a base container exists on this machine and download one if
27        # necessary.
28        image = lxc.BaseImage()
29        try:
30            cls.base_container = image.get()
31            cls.cleanup_base_container = False
32        except error.ContainerError:
33            image.setup()
34            cls.base_container = image.get()
35            cls.cleanup_base_container = True
36        assert(cls.base_container is not None)
37
38
39    @classmethod
40    def tearDownClass(cls):
41        cls.base_container = None
42        if not unittest_setup.config.skip_cleanup:
43            if cls.cleanup_base_container:
44                lxc.BaseImage().cleanup()
45            utils.run('sudo rm -r %s' % cls.test_dir)
46
47
48    def setUp(self):
49        # Create a separate dir for each test, so they are hermetic.
50        self.test_dir = tempfile.mkdtemp(dir=ContainerFactoryTests.test_dir)
51        self.test_factory = lxc.ContainerFactory(
52                base_container=self.base_container,
53                lxc_path=self.test_dir)
54
55
56    def testCreateContainer(self):
57        """Tests basic container creation."""
58        container = self.test_factory.create_container()
59
60        try:
61            container.refresh_status()
62        except:
63            self.fail('Invalid container:\n%s' % error.format_error())
64
65
66    def testCreateContainer_noId(self):
67        """Tests container creation with default IDs."""
68        container = self.test_factory.create_container()
69        self.assertIsNone(container.id)
70
71
72    def testCreateContainer_withId(self):
73        """Tests container creation with given IDs. """
74        id0 = lxc.ContainerId(1, 2, 3)
75        container = self.test_factory.create_container(id0)
76        self.assertEquals(id0, container.id)
77
78
79    def testContainerName(self):
80        """Tests that created containers have the right name."""
81        id0 = lxc.ContainerId(1, 2, 3)
82        id1 = lxc.ContainerId(42, 41, 40)
83
84        container0 = self.test_factory.create_container(id0)
85        container1 = self.test_factory.create_container(id1)
86
87        self.assertEqual(str(id0), container0.name)
88        self.assertEqual(str(id1), container1.name)
89
90
91    def testContainerPath(self):
92        """Tests that created containers have the right LXC path."""
93        dir0 = tempfile.mkdtemp(dir=self.test_dir)
94        dir1 = tempfile.mkdtemp(dir=self.test_dir)
95
96        container0 = self.test_factory.create_container(lxc_path=dir0)
97        container1 = self.test_factory.create_container(lxc_path=dir1)
98
99        self.assertEqual(dir0, container0.container_path);
100        self.assertEqual(dir1, container1.container_path);
101
102
103    def testCreateContainer_alreadyExists(self):
104        """Tests that container ID conflicts raise errors as expected."""
105        id0 = lxc.ContainerId(1, 2, 3)
106
107        self.test_factory.create_container(id0)
108        with self.assertRaises(error.ContainerError):
109            self.test_factory.create_container(id0)
110
111
112    def testCreateContainer_forceReset(self):
113        """Tests that force-resetting containers works."""
114        factory = lxc.ContainerFactory(base_container=self.base_container,
115                                       lxc_path=self.test_dir,
116                                       force_cleanup=True)
117
118        id0 = lxc.ContainerId(1, 2, 3)
119        container0 = factory.create_container(id0)
120        container0.start(wait_for_network=False)
121
122        # Create a file in the original container.
123        tmpfile = container0.attach_run('mktemp').stdout
124        exists = 'test -e %s' % tmpfile
125        try:
126            container0.attach_run(exists)
127        except error.CmdError as e:
128            self.fail(e)
129
130        # Create a new container in place of the original, then verify that the
131        # file is no longer there.
132        container1 = factory.create_container(id0)
133        container1.start(wait_for_network=False)
134        with self.assertRaises(error.CmdError):
135            container1.attach_run(exists)
136
137
138    def testCreateContainer_subclass(self):
139        """Tests that the factory produces objects of the requested class."""
140        container = self.test_factory.create_container()
141        # Don't use isinstance, we want to check the exact type.
142        self.assertTrue(type(container) is lxc.Container)
143
144        class _TestContainer(lxc.Container):
145            """A test Container subclass"""
146            pass
147
148        test_factory = lxc.ContainerFactory(base_container=self.base_container,
149                                            container_class=_TestContainer,
150                                            lxc_path=self.test_dir)
151        test_container = test_factory.create_container()
152        self.assertTrue(type(test_container) is _TestContainer)
153
154
155    def testCreateContainer_snapshotFails(self):
156        """Tests the scenario where snapshotting fails.
157
158        Verifies that the factory is still able to produce a Container when
159        cloning fails.
160        """
161        class MockContainerClass(object):
162            """A mock object to simulate the container class.
163
164            This mock has a clone method that simulates a failure when clone is
165            called with snapshot=True.  Clone calls are recorded so they can be
166            verified later.
167            """
168            def __init__(self):
169                """Initializes the mock."""
170                self.clone_count = 0
171                self.clone_kwargs = []
172
173
174            def clone(self, *args, **kwargs):
175                """Mocks the Container.clone class method. """
176                # Record the particulars of this call.
177                self.clone_count += 1
178                self.clone_kwargs.append(kwargs)
179                # Simulate failure if a snapshot is requested, otherwise create
180                # and return the clone.
181                if kwargs['snapshot']:
182                    raise error.CmdError('fake error', None)
183                else:
184                    return lxc.Container.clone(*args, **kwargs)
185
186        mock = MockContainerClass()
187        factory = lxc.ContainerFactory(base_container=self.base_container,
188                                       container_class=mock,
189                                       snapshot=True,
190                                       lxc_path=self.test_dir)
191
192        factory.create_container()
193        # The factory should have made 2 calls to mock.clone - the first with
194        # snapshot=True, then the second with snapshot=False.
195        self.assertEquals(2, mock.clone_count)
196        self.assertTrue(mock.clone_kwargs[0]['snapshot'])
197        self.assertFalse(mock.clone_kwargs[1]['snapshot'])
198
199
200if __name__ == '__main__':
201    unittest.main()
202