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