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 argparse 7 import logging 8 import os 9 import tempfile 10 import shutil 11 import sys 12 import unittest 13 from contextlib import contextmanager 14 15 import common 16 from autotest_lib.client.common_lib import error 17 from autotest_lib.site_utils import lxc 18 from autotest_lib.site_utils.lxc import constants 19 from autotest_lib.site_utils.lxc import unittest_http 20 from autotest_lib.site_utils.lxc import unittest_logging 21 from autotest_lib.site_utils.lxc import utils as lxc_utils 22 from autotest_lib.site_utils.lxc.unittest_container_bucket \ 23 import FastContainerBucket 24 25 options = None 26 27 class ContainerTests(unittest.TestCase): 28 """Unit tests for the Container class.""" 29 30 @classmethod 31 def setUpClass(cls): 32 cls.test_dir = tempfile.mkdtemp(dir=lxc.DEFAULT_CONTAINER_PATH, 33 prefix='container_unittest_') 34 cls.shared_host_path = os.path.join(cls.test_dir, 'host') 35 36 # Use a container bucket just to download and set up the base image. 37 cls.bucket = FastContainerBucket(cls.test_dir, cls.shared_host_path) 38 39 if cls.bucket.base_container is None: 40 logging.debug('Base container not found - reinitializing') 41 cls.bucket.setup_base() 42 else: 43 logging.debug('base container found') 44 cls.base_container = cls.bucket.base_container 45 assert(cls.base_container is not None) 46 47 48 @classmethod 49 def tearDownClass(cls): 50 cls.base_container = None 51 if not options.skip_cleanup: 52 cls.bucket.destroy_all() 53 shutil.rmtree(cls.test_dir) 54 55 def tearDown(self): 56 # Ensure host dirs from each test are completely destroyed. 57 for host_dir in os.listdir(self.shared_host_path): 58 host_dir = os.path.realpath(os.path.join(self.shared_host_path, 59 host_dir)) 60 lxc_utils.cleanup_host_mount(host_dir); 61 62 63 def testInit(self): 64 """Verifies that containers initialize correctly.""" 65 # Make a container that just points to the base container. 66 container = lxc.Container.createFromExistingDir( 67 self.base_container.container_path, 68 self.base_container.name) 69 self.assertFalse(container.is_running()) 70 71 72 def testInitInvalid(self): 73 """Verifies that invalid containers can still be instantiated, 74 if not used. 75 """ 76 with tempfile.NamedTemporaryFile(dir=self.test_dir) as tmpfile: 77 name = os.path.basename(tmpfile.name) 78 container = lxc.Container.createFromExistingDir(self.test_dir, name) 79 with self.assertRaises(error.ContainerError): 80 container.refresh_status() 81 82 83 def testDefaultHostname(self): 84 """Verifies that the zygote starts up with a default hostname that is 85 the lxc container name.""" 86 test_name = 'testHostname' 87 with self.createContainer(name=test_name) as container: 88 container.start(wait_for_network=True) 89 hostname = container.attach_run('hostname').stdout.strip() 90 self.assertEqual(test_name, hostname) 91 92 93 @unittest.skip('Setting the container hostname using lxc.utsname does not' 94 'work on goobuntu.') 95 def testSetHostnameNotRunning(self): 96 """Verifies that the hostname can be set on a stopped container.""" 97 with self.createContainer() as container: 98 expected_hostname = 'my-new-hostname' 99 container.set_hostname(expected_hostname) 100 container.start(wait_for_network=True) 101 hostname = container.attach_run('hostname').stdout.strip() 102 self.assertEqual(expected_hostname, hostname) 103 104 105 def testClone(self): 106 """Verifies that cloning a container works as expected.""" 107 clone = lxc.Container.clone(src=self.base_container, 108 new_name="testClone", 109 snapshot=True) 110 try: 111 # Throws an exception if the container is not valid. 112 clone.refresh_status() 113 finally: 114 clone.destroy() 115 116 117 def testCloneWithoutCleanup(self): 118 """Verifies that cloning a container to an existing name will fail as 119 expected. 120 """ 121 lxc.Container.clone(src=self.base_container, 122 new_name="testCloneWithoutCleanup", 123 snapshot=True) 124 with self.assertRaises(error.ContainerError): 125 lxc.Container.clone(src=self.base_container, 126 new_name="testCloneWithoutCleanup", 127 snapshot=True) 128 129 130 def testCloneWithCleanup(self): 131 """Verifies that cloning a container with cleanup works properly.""" 132 clone0 = lxc.Container.clone(src=self.base_container, 133 new_name="testClone", 134 snapshot=True) 135 clone0.start(wait_for_network=False) 136 tmpfile = clone0.attach_run('mktemp').stdout 137 # Verify that our tmpfile exists 138 clone0.attach_run('test -f %s' % tmpfile) 139 140 # Clone another container in place of the existing container. 141 clone1 = lxc.Container.clone(src=self.base_container, 142 new_name="testClone", 143 snapshot=True, 144 cleanup=True) 145 with self.assertRaises(error.CmdError): 146 clone1.attach_run('test -f %s' % tmpfile) 147 148 149 def testInstallSsp(self): 150 """Verifies that installing the ssp in the container works.""" 151 # Hard-coded path to some golden data for this test. 152 test_ssp = os.path.join( 153 common.autotest_dir, 154 'site_utils', 'lxc', 'test', 'test_ssp.tar.bz2') 155 # Create a container, install the self-served ssp, then check that it is 156 # installed into the container correctly. 157 with self.createContainer() as container: 158 with unittest_http.serve_locally(test_ssp) as url: 159 container.install_ssp(url) 160 container.start(wait_for_network=False) 161 162 # The test ssp just contains a couple of text files, in known 163 # locations. Verify the location and content of those files in the 164 # container. 165 cat = lambda path: container.attach_run('cat %s' % path).stdout 166 test0 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, 167 'test.0')) 168 test1 = cat(os.path.join(constants.CONTAINER_AUTOTEST_DIR, 169 'dir0', 'test.1')) 170 self.assertEquals('the five boxing wizards jumped quickly', 171 test0) 172 self.assertEquals('the quick brown fox jumps over the lazy dog', 173 test1) 174 175 176 def testInstallControlFile(self): 177 """Verifies that installing a control file in the container works.""" 178 _unused, tmpfile = tempfile.mkstemp() 179 with self.createContainer() as container: 180 container.install_control_file(tmpfile) 181 container.start(wait_for_network=False) 182 # Verify that the file is found in the container. 183 container.attach_run( 184 'test -f %s' % os.path.join(lxc.CONTROL_TEMP_PATH, 185 os.path.basename(tmpfile))) 186 187 188 @contextmanager 189 def createContainer(self, name=None): 190 """Creates a container from the base container, for testing. 191 Use this to ensure that containers get properly cleaned up after each 192 test. 193 194 @param name: An optional name for the new container. 195 """ 196 if name is None: 197 name = self.id().split('.')[-1] 198 container = self.bucket.create_from_base(name) 199 try: 200 yield container 201 finally: 202 container.destroy() 203 204 205 def parse_options(): 206 """Parse command line inputs. 207 """ 208 parser = argparse.ArgumentParser() 209 parser.add_argument('-v', '--verbose', action='store_true', 210 help='Print out ALL entries.') 211 parser.add_argument('--skip_cleanup', action='store_true', 212 help='Skip deleting test containers.') 213 args, argv = parser.parse_known_args() 214 215 # Hack: python unittest also processes args. Construct an argv to pass to 216 # it, that filters out the options it won't recognize. 217 if args.verbose: 218 argv.insert(0, '-v') 219 argv.insert(0, sys.argv[0]) 220 221 return args, argv 222 223 224 if __name__ == '__main__': 225 options, unittest_argv = parse_options() 226 227 log_level=(logging.DEBUG if options.verbose else logging.INFO) 228 unittest_logging.setup(log_level) 229 230 unittest.main(argv=unittest_argv) 231
