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