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 os 7 import random 8 import shutil 9 import tempfile 10 import unittest 11 from contextlib import contextmanager 12 13 import common 14 from autotest_lib.client.bin import utils 15 from autotest_lib.client.common_lib import error 16 from autotest_lib.site_utils import lxc 17 from autotest_lib.site_utils.lxc import constants 18 from autotest_lib.site_utils.lxc import container as container_module 19 from autotest_lib.site_utils.lxc import unittest_http 20 from autotest_lib.site_utils.lxc import unittest_setup 21 from autotest_lib.site_utils.lxc import utils as lxc_utils 22 23 24 class ContainerTests(unittest.TestCase): 25 """Unit tests for the Container class.""" 26 27 @classmethod 28 def setUpClass(cls): 29 cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH, 30 prefix='container_unittest_') 31 32 # Check if a base container exists on this machine and download one if 33 # necessary. 34 image = lxc.BaseImage() 35 try: 36 cls.base_container = image.get() 37 cls.cleanup_base_container = False 38 except error.ContainerError: 39 image.setup() 40 cls.base_container = image.get() 41 cls.cleanup_base_container = True 42 assert(cls.base_container is not None) 43 44 45 @classmethod 46 def tearDownClass(cls): 47 cls.base_container = None 48 if not unittest_setup.config.skip_cleanup: 49 if cls.cleanup_base_container: 50 lxc.BaseImage().cleanup() 51 utils.run('sudo rm -r %s' % cls.test_dir) 52 53 54 def testInit(self): 55 """Verifies that containers initialize correctly.""" 56 # Make a container that just points to the base container. 57 container = lxc.Container.create_from_existing_dir( 58 self.base_container.container_path, 59 self.base_container.name) 60 # Calling is_running triggers an lxc-ls call, which should verify that 61 # the on-disk container is valid. 62 self.assertFalse(container.is_running()) 63 64 65 def testInitInvalid(self): 66 """Verifies that invalid containers can still be instantiated, 67 if not used. 68 """ 69 with tempfile.NamedTemporaryFile(dir=self.test_dir) as tmpfile: 70 name = os.path.basename(tmpfile.name) 71 container = lxc.Container.create_from_existing_dir(self.test_dir, 72 name) 73 with self.assertRaises(error.ContainerError): 74 container.refresh_status() 75 76 77 def testInvalidId(self): 78 """Verifies that corrupted ID files do not raise exceptions.""" 79 with self.createContainer() as container: 80 # Create a container with an empty ID file. 81 id_path = os.path.join(container.container_path, 82 container.name, 83 container_module._CONTAINER_ID_FILENAME) 84 utils.run('sudo touch %s' % id_path) 85 try: 86 # Verify that container creation doesn't raise exceptions. 87 test_container = lxc.Container.create_from_existing_dir( 88 self.test_dir, container.name) 89 self.assertIsNone(test_container.id) 90 except Exception: 91 self.fail('Unexpected exception:\n%s' % error.format_error()) 92 93 94 def testDefaultHostname(self): 95 """Verifies that the zygote starts up with a default hostname that is 96 the lxc container name.""" 97 test_name = 'testHostname' 98 with self.createContainer(name=test_name) as container: 99 container.start(wait_for_network=True) 100 hostname = container.attach_run('hostname').stdout.strip() 101 self.assertEqual(test_name, hostname) 102 103 104 def testSetHostnameRunning(self): 105 """Verifies that the hostname can be set on a running container.""" 106 with self.createContainer() as container: 107 expected_hostname = 'my-new-hostname' 108 container.start(wait_for_network=True) 109 container.set_hostname(expected_hostname) 110 hostname = container.attach_run('hostname -f').stdout.strip() 111 self.assertEqual(expected_hostname, hostname) 112 113 114 def testSetHostnameNotRunningRaisesException(self): 115 """Verifies that set_hostname on a stopped container raises an error. 116 117 The lxc.utsname config setting is unreliable (it only works if the 118 original container name is not a valid RFC-952 hostname, e.g. if it has 119 underscores). 120 121 A more reliable method exists for setting the hostname but it requires 122 the container to be running. To avoid confusion, setting the hostname 123 on a stopped container is disallowed. 124 125 This test verifies that the operation raises a ContainerError. 126 """ 127 with self.createContainer() as container: 128 with self.assertRaises(error.ContainerError): 129 # Ensure the container is not running 130 if container.is_running(): 131 raise RuntimeError('Container should not be running.') 132 container.set_hostname('foobar') 133 134 135 def testClone(self): 136 """Verifies that cloning a container works as expected.""" 137 clone = lxc.Container.clone(src=self.base_container, 138 new_name="testClone", 139 new_path=self.test_dir, 140 snapshot=True) 141 try: 142 # Throws an exception if the container is not valid. 143 clone.refresh_status() 144 finally: 145 clone.destroy() 146 147 148 def testCloneWithoutCleanup(self): 149 """Verifies that cloning a container to an existing name will fail as 150 expected. 151 """ 152 lxc.Container.clone(src=self.base_container, 153 new_name="testCloneWithoutCleanup", 154 new_path=self.test_dir, 155 snapshot=True) 156 with self.assertRaises(error.ContainerError): 157 lxc.Container.clone(src=self.base_container, 158 new_name="testCloneWithoutCleanup", 159 new_path=self.test_dir, 160 snapshot=True) 161 162 163 def testCloneWithCleanup(self): 164 """Verifies that cloning a container with cleanup works properly.""" 165 clone0 = lxc.Container.clone(src=self.base_container, 166 new_name="testClone", 167 new_path=self.test_dir, 168 snapshot=True) 169 clone0.start(wait_for_network=False) 170 tmpfile = clone0.attach_run('mktemp').stdout 171 # Verify that our tmpfile exists 172 clone0.attach_run('test -f %s' % tmpfile) 173 174 # Clone another container in place of the existing container. 175 clone1 = lxc.Container.clone(src=self.base_container, 176 new_name="testClone", 177 new_path=self.test_dir, 178 snapshot=True, 179 cleanup=True) 180 with self.assertRaises(error.CmdError): 181 clone1.attach_run('test -f %s' % tmpfile) 182 183 184 def testInstallSsp(self): 185 """Verifies that installing the ssp in the container works.""" 186 # Hard-coded path to some golden data for this test. 187 test_ssp = os.path.join( 188 common.autotest_dir, 189 'site_utils', 'lxc', 'test', 'test_ssp.tar.bz2') 190 # Create a container, install the self-served ssp, then check that it is 191 # installed into the container correctly. 192 with self.createContainer() as container: 193 with unittest_http.serve_locally(test_ssp) as url: 194 container.install_ssp(url) 195 container.start(wait_for_network=False) 196 197 # The test ssp just contains a couple of text files, in known 198 # locations. Verify the location and content of those files in the 199 # container. 200 cat = lambda path: container.attach_run('cat %s' % path).stdout 201 test0 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, 202 'test.0')) 203 test1 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, 204 'dir0', 'test.1')) 205 self.assertEquals('the five boxing wizards jumped quickly', 206 test0) 207 self.assertEquals('the quick brown fox jumps over the lazy dog', 208 test1) 209 210 211 def testInstallControlFile(self): 212 """Verifies that installing a control file in the container works.""" 213 _unused, tmpfile = tempfile.mkstemp() 214 with self.createContainer() as container: 215 container.install_control_file(tmpfile) 216 container.start(wait_for_network=False) 217 # Verify that the file is found in the container. 218 container.attach_run( 219 'test -f %s' % os.path.join(lxc.CONTROL_TEMP_PATH, 220 os.path.basename(tmpfile))) 221 222 223 def testCopyFile(self): 224 """Verifies that files are correctly copied into the container.""" 225 control_string = 'amazingly few discotheques provide jukeboxes' 226 with tempfile.NamedTemporaryFile() as tmpfile: 227 tmpfile.write(control_string) 228 tmpfile.flush() 229 230 with self.createContainer() as container: 231 dst = os.path.join(constants.CONTAINER_AUTOTEST_DIR, 232 os.path.basename(tmpfile.name)) 233 container.copy(tmpfile.name, dst) 234 container.start(wait_for_network=False) 235 # Verify the file content. 236 test_string = container.attach_run('cat %s' % dst).stdout 237 self.assertEquals(control_string, test_string) 238 239 240 def testCopyDirectory(self): 241 """Verifies that directories are correctly copied into the container.""" 242 control_string = 'pack my box with five dozen liquor jugs' 243 with lxc_utils.TempDir() as tmpdir: 244 fd, tmpfile = tempfile.mkstemp(dir=tmpdir) 245 f = os.fdopen(fd, 'w') 246 f.write(control_string) 247 f.close() 248 249 with self.createContainer() as container: 250 dst = os.path.join(constants.CONTAINER_AUTOTEST_DIR, 251 os.path.basename(tmpdir)) 252 container.copy(tmpdir, dst) 253 container.start(wait_for_network=False) 254 # Verify the file content. 255 test_file = os.path.join(dst, os.path.basename(tmpfile)) 256 test_string = container.attach_run('cat %s' % test_file).stdout 257 self.assertEquals(control_string, test_string) 258 259 260 def testMountDirectory(self): 261 """Verifies that read-write mounts work.""" 262 with lxc_utils.TempDir() as tmpdir, self.createContainer() as container: 263 dst = '/testMountDirectory/testMount' 264 container.mount_dir(tmpdir, dst, readonly=False) 265 container.start(wait_for_network=False) 266 267 # Verify that the mount point is correctly bound, and is read-write. 268 self.verifyBindMount(container, dst, tmpdir) 269 container.attach_run('test -r %s -a -w %s' % (dst, dst)) 270 271 272 def testMountDirectoryReadOnly(self): 273 """Verifies that read-only mounts work.""" 274 with lxc_utils.TempDir() as tmpdir, self.createContainer() as container: 275 dst = '/testMountDirectoryReadOnly/testMount' 276 container.mount_dir(tmpdir, dst, readonly=True) 277 container.start(wait_for_network=False) 278 279 # Verify that the mount point is correctly bound, and is read-only. 280 self.verifyBindMount(container, dst, tmpdir) 281 container.attach_run('test -r %s -a ! -w %s' % (dst, dst)) 282 283 284 def testMountDirectoryRelativePath(self): 285 """Verifies that relative-path mounts work.""" 286 with lxc_utils.TempDir() as tmpdir, self.createContainer() as container: 287 dst = 'testMountDirectoryRelativePath/testMount' 288 container.mount_dir(tmpdir, dst, readonly=True) 289 container.start(wait_for_network=False) 290 291 # Verify that the mount points is correctly bound.. 292 self.verifyBindMount(container, dst, tmpdir) 293 294 295 def testContainerIdPersistence(self): 296 """Verifies that container IDs correctly persist. 297 298 When a Container is instantiated on top of an existing container dir, 299 check that it picks up the correct ID. 300 """ 301 with self.createContainer() as container: 302 test_id = random_container_id() 303 container.id = test_id 304 305 # Set up another container and verify that its ID matches. 306 test_container = lxc.Container.create_from_existing_dir( 307 container.container_path, container.name) 308 309 self.assertEqual(test_id, test_container.id) 310 311 312 def testContainerIdIsNone_newContainer(self): 313 """Verifies that newly created/cloned containers have no ID.""" 314 with self.createContainer() as container: 315 self.assertIsNone(container.id) 316 # Set an ID, clone the container, and verify the clone has no ID. 317 container.id = random_container_id() 318 clone = lxc.Container.clone(src=container, 319 new_name=container.name + '_clone', 320 snapshot=True) 321 self.assertIsNotNone(container.id) 322 self.assertIsNone(clone.id) 323 324 325 @contextmanager 326 def createContainer(self, name=None): 327 """Creates a container from the base container, for testing. 328 Use this to ensure that containers get properly cleaned up after each 329 test. 330 331 @param name: An optional name for the new container. 332 """ 333 if name is None: 334 name = self.id().split('.')[-1] 335 container = lxc.Container.clone(src=self.base_container, 336 new_name=name, 337 new_path=self.test_dir, 338 snapshot=True) 339 try: 340 yield container 341 finally: 342 if not unittest_setup.config.skip_cleanup: 343 container.destroy() 344 345 346 def verifyBindMount(self, container, container_path, host_path): 347 """Verifies that a given path in a container is bind-mounted to a given 348 path in the host system. 349 350 @param container: The Container instance to be tested. 351 @param container_path: The path in the container to compare. 352 @param host_path: The path in the host system to compare. 353 """ 354 container_inode = (container.attach_run('ls -id %s' % container_path) 355 .stdout.split()[0]) 356 host_inode = utils.run('ls -id %s' % host_path).stdout.split()[0] 357 # Compare the container and host inodes - they should match. 358 self.assertEqual(container_inode, host_inode) 359 360 361 class ContainerIdTests(unittest.TestCase): 362 """Unit tests for the ContainerId class.""" 363 364 def setUp(self): 365 self.test_dir = tempfile.mkdtemp() 366 367 368 def tearDown(self): 369 shutil.rmtree(self.test_dir) 370 371 372 def testPickle(self): 373 """Verifies the ContainerId persistence code.""" 374 # Create a random ID, then save and load it and compare them. 375 control = random_container_id() 376 control.save(self.test_dir) 377 378 test_data = lxc.ContainerId.load(self.test_dir) 379 self.assertEqual(control, test_data) 380 381 382 def random_container_id(): 383 """Generate a random container ID for testing.""" 384 return lxc.ContainerId.create(random.randint(0, 1000)) 385 386 387 if __name__ == '__main__': 388 unittest_setup.setup() 389 unittest.main() 390
