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