1#
2# Copyright (c) 2017 Wind River Systems, Inc.
3#
4# SPDX-License-Identifier: MIT
5#
6
7import re
8import tempfile
9import time
10import oe.types
11from oeqa.selftest.case import OESelftestTestCase
12from oeqa.utils.commands import bitbake, runqemu, get_bb_var, runCmd
13
14class RunqemuTests(OESelftestTestCase):
15    """Runqemu test class"""
16
17    image_is_ready = False
18    deploy_dir_image = ''
19
20    def setUpLocal(self):
21        super(RunqemuTests, self).setUpLocal()
22        self.recipe = 'core-image-minimal'
23        self.machine =  'qemux86-64'
24        self.fstypes = "ext4 iso hddimg wic.vmdk wic.qcow2 wic.vdi"
25        self.cmd_common = "runqemu nographic"
26
27        kvm = oe.types.qemu_use_kvm(get_bb_var('QEMU_USE_KVM'), 'x86_64')
28        if kvm:
29            self.cmd_common += " kvm"
30
31        self.write_config(
32"""
33MACHINE = "%s"
34IMAGE_FSTYPES = "%s"
35# 10 means 1 second
36SYSLINUX_TIMEOUT = "10"
37"""
38% (self.machine, self.fstypes)
39        )
40
41        if not RunqemuTests.image_is_ready:
42            RunqemuTests.deploy_dir_image = get_bb_var('DEPLOY_DIR_IMAGE')
43            bitbake(self.recipe)
44            RunqemuTests.image_is_ready = True
45
46    def test_boot_machine(self):
47        """Test runqemu machine"""
48        cmd = "%s %s" % (self.cmd_common, self.machine)
49        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
50            with open(qemu.qemurunnerlog) as f:
51                self.assertTrue(qemu.runner.logged, "Failed: %s, %s" % (cmd, f.read()))
52
53    def test_boot_machine_ext4(self):
54        """Test runqemu machine ext4"""
55        cmd = "%s %s ext4" % (self.cmd_common, self.machine)
56        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
57            with open(qemu.qemurunnerlog) as f:
58                self.assertIn('rootfs.ext4', f.read(), "Failed: %s" % cmd)
59
60    def test_boot_machine_iso(self):
61        """Test runqemu machine iso"""
62        cmd = "%s %s iso" % (self.cmd_common, self.machine)
63        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
64            with open(qemu.qemurunnerlog) as f:
65                self.assertIn('media=cdrom', f.read(), "Failed: %s" % cmd)
66
67    def test_boot_recipe_image(self):
68        """Test runqemu recipe-image"""
69        cmd = "%s %s" % (self.cmd_common, self.recipe)
70        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
71            with open(qemu.qemurunnerlog) as f:
72                self.assertTrue(qemu.runner.logged, "Failed: %s, %s" % (cmd, f.read()))
73
74
75    def test_boot_recipe_image_vmdk(self):
76        """Test runqemu recipe-image vmdk"""
77        cmd = "%s %s wic.vmdk" % (self.cmd_common, self.recipe)
78        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
79            with open(qemu.qemurunnerlog) as f:
80                self.assertIn('format=vmdk', f.read(), "Failed: %s" % cmd)
81
82    def test_boot_recipe_image_vdi(self):
83        """Test runqemu recipe-image vdi"""
84        cmd = "%s %s wic.vdi" % (self.cmd_common, self.recipe)
85        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
86            with open(qemu.qemurunnerlog) as f:
87                self.assertIn('format=vdi', f.read(), "Failed: %s" % cmd)
88
89    def test_boot_deploy(self):
90        """Test runqemu deploy_dir_image"""
91        cmd = "%s %s" % (self.cmd_common, self.deploy_dir_image)
92        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
93            with open(qemu.qemurunnerlog) as f:
94                self.assertTrue(qemu.runner.logged, "Failed: %s, %s" % (cmd, f.read()))
95
96
97    def test_boot_deploy_hddimg(self):
98        """Test runqemu deploy_dir_image hddimg"""
99        cmd = "%s %s hddimg" % (self.cmd_common, self.deploy_dir_image)
100        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
101            with open(qemu.qemurunnerlog) as f:
102                self.assertTrue(re.search('file=.*.hddimg', f.read()), "Failed: %s, %s" % (cmd, f.read()))
103
104    def test_boot_machine_slirp(self):
105        """Test runqemu machine slirp"""
106        cmd = "%s slirp %s" % (self.cmd_common, self.machine)
107        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
108            with open(qemu.qemurunnerlog) as f:
109                self.assertIn(' -netdev user', f.read(), "Failed: %s" % cmd)
110
111    def test_boot_machine_slirp_qcow2(self):
112        """Test runqemu machine slirp qcow2"""
113        cmd = "%s slirp wic.qcow2 %s" % (self.cmd_common, self.machine)
114        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
115            with open(qemu.qemurunnerlog) as f:
116                self.assertIn('format=qcow2', f.read(), "Failed: %s" % cmd)
117
118    def test_boot_qemu_boot(self):
119        """Test runqemu /path/to/image.qemuboot.conf"""
120        qemuboot_conf = "%s-%s.qemuboot.conf" % (self.recipe, self.machine)
121        qemuboot_conf = os.path.join(self.deploy_dir_image, qemuboot_conf)
122        if not os.path.exists(qemuboot_conf):
123            self.skipTest("%s not found" % qemuboot_conf)
124        cmd = "%s %s" % (self.cmd_common, qemuboot_conf)
125        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
126            with open(qemu.qemurunnerlog) as f:
127                self.assertTrue(qemu.runner.logged, "Failed: %s, %s" % (cmd, f.read()))
128
129    def test_boot_rootfs(self):
130        """Test runqemu /path/to/rootfs.ext4"""
131        rootfs = "%s-%s.ext4" % (self.recipe, self.machine)
132        rootfs = os.path.join(self.deploy_dir_image, rootfs)
133        if not os.path.exists(rootfs):
134            self.skipTest("%s not found" % rootfs)
135        cmd = "%s %s" % (self.cmd_common, rootfs)
136        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
137            with open(qemu.qemurunnerlog) as f:
138                self.assertTrue(qemu.runner.logged, "Failed: %s, %s" % (cmd, f.read()))
139
140
141# This test was designed as a separate class to test that shutdown
142# command will shutdown qemu as expected on each qemu architecture
143# based on the MACHINE configuration inside the config file
144# (eg. local.conf).
145#
146# This was different compared to RunqemuTests, where RunqemuTests was
147# dedicated for MACHINE=qemux86-64 where it test that qemux86-64 will
148# bootup various filesystem types, including live image(iso and hddimg)
149# where live image was not supported on all qemu architecture.
150class QemuTest(OESelftestTestCase):
151
152    @classmethod
153    def setUpClass(cls):
154        super(QemuTest, cls).setUpClass()
155        cls.recipe = 'core-image-minimal'
156        cls.machine =  get_bb_var('MACHINE')
157        cls.deploy_dir_image =  get_bb_var('DEPLOY_DIR_IMAGE')
158        cls.cmd_common = "runqemu nographic"
159        cls.qemuboot_conf = "%s-%s.qemuboot.conf" % (cls.recipe, cls.machine)
160        cls.qemuboot_conf = os.path.join(cls.deploy_dir_image, cls.qemuboot_conf)
161        bitbake(cls.recipe)
162
163    def _start_qemu_shutdown_check_if_shutdown_succeeded(self, qemu, timeout):
164        qemu.run_serial("shutdown -h now")
165        # Stop thread will stop the LoggingThread instance used for logging
166        # qemu through serial console, stop thread will prevent this code
167        # from facing exception (Console connection closed unexpectedly)
168        # when qemu was shutdown by the above shutdown command
169        qemu.runner.stop_thread()
170        time_track = 0
171        try:
172            while True:
173                is_alive = qemu.check()
174                if not is_alive:
175                    return True
176                if time_track > timeout:
177                    return False
178                time.sleep(1)
179                time_track += 1
180        except SystemExit:
181            return True
182
183    def test_qemu_can_shutdown(self):
184        self.assertExists(self.qemuboot_conf)
185        cmd = "%s %s" % (self.cmd_common, self.qemuboot_conf)
186        shutdown_timeout = 120
187        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
188            qemu_shutdown_succeeded = self._start_qemu_shutdown_check_if_shutdown_succeeded(qemu, shutdown_timeout)
189            self.assertTrue(qemu_shutdown_succeeded, 'Failed: %s does not shutdown within timeout(%s)' % (self.machine, shutdown_timeout))
190
191    # Need to have portmap/rpcbind running to allow this test to work and
192    # current autobuilder setup does not have this.
193    def disabled_test_qemu_can_boot_nfs_and_shutdown(self):
194        self.assertExists(self.qemuboot_conf)
195        bitbake('meta-ide-support')
196        rootfs_tar = "%s-%s.tar.bz2" % (self.recipe, self.machine)
197        rootfs_tar = os.path.join(self.deploy_dir_image, rootfs_tar)
198        self.assertExists(rootfs_tar)
199        tmpdir = tempfile.mkdtemp(prefix='qemu_nfs')
200        tmpdir_nfs = os.path.join(tmpdir, 'nfs')
201        cmd_extract_nfs = 'runqemu-extract-sdk %s %s' % (rootfs_tar, tmpdir_nfs)
202        result = runCmd(cmd_extract_nfs)
203        self.assertEqual(0, result.status, "runqemu-extract-sdk didn't run as expected. %s" % result.output)
204        cmd = "%s nfs %s %s" % (self.cmd_common, self.qemuboot_conf, tmpdir_nfs)
205        shutdown_timeout = 120
206        with runqemu(self.recipe, ssh=False, launch_cmd=cmd) as qemu:
207            qemu_shutdown_succeeded = self._start_qemu_shutdown_check_if_shutdown_succeeded(qemu, shutdown_timeout)
208            self.assertTrue(qemu_shutdown_succeeded, 'Failed: %s does not shutdown within timeout(%s)' % (self.machine, shutdown_timeout))
209        runCmd('rm -rf %s' % tmpdir)
210