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 6 import tempfile 7 import unittest 8 9 import common 10 from autotest_lib.client.bin import utils 11 from autotest_lib.client.common_lib import error 12 from autotest_lib.site_utils import lxc 13 from autotest_lib.site_utils.lxc import unittest_setup 14 from autotest_lib.site_utils.lxc import utils as lxc_utils 15 16 17 class 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 200 if __name__ == '__main__': 201 unittest.main() 202
