tests: add prefixes to the bare mkdtemp calls
[qemu.git] / tests / acceptance / avocado_qemu / __init__.py
1 # Test class and utilities for functional tests
2 #
3 # Copyright (c) 2018 Red Hat, Inc.
4 #
5 # Author:
6 # Cleber Rosa <crosa@redhat.com>
7 #
8 # This work is licensed under the terms of the GNU GPL, version 2 or
9 # later. See the COPYING file in the top-level directory.
10
11 import logging
12 import os
13 import sys
14 import uuid
15 import tempfile
16
17 import avocado
18
19 #: The QEMU build root directory. It may also be the source directory
20 #: if building from the source dir, but it's safer to use BUILD_DIR for
21 #: that purpose. Be aware that if this code is moved outside of a source
22 #: and build tree, it will not be accurate.
23 BUILD_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
24
25 if os.path.islink(os.path.dirname(os.path.dirname(__file__))):
26 # The link to the acceptance tests dir in the source code directory
27 lnk = os.path.dirname(os.path.dirname(__file__))
28 #: The QEMU root source directory
29 SOURCE_DIR = os.path.dirname(os.path.dirname(os.readlink(lnk)))
30 else:
31 SOURCE_DIR = BUILD_DIR
32
33 sys.path.append(os.path.join(SOURCE_DIR, 'python'))
34
35 from qemu.machine import QEMUMachine
36
37 def is_readable_executable_file(path):
38 return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
39
40
41 def pick_default_qemu_bin(arch=None):
42 """
43 Picks the path of a QEMU binary, starting either in the current working
44 directory or in the source tree root directory.
45
46 :param arch: the arch to use when looking for a QEMU binary (the target
47 will match the arch given). If None (the default), arch
48 will be the current host system arch (as given by
49 :func:`os.uname`).
50 :type arch: str
51 :returns: the path to the default QEMU binary or None if one could not
52 be found
53 :rtype: str or None
54 """
55 if arch is None:
56 arch = os.uname()[4]
57 # qemu binary path does not match arch for powerpc, handle it
58 if 'ppc64le' in arch:
59 arch = 'ppc64'
60 qemu_bin_relative_path = "./qemu-system-%s" % arch
61 if is_readable_executable_file(qemu_bin_relative_path):
62 return qemu_bin_relative_path
63
64 qemu_bin_from_bld_dir_path = os.path.join(BUILD_DIR,
65 qemu_bin_relative_path)
66 if is_readable_executable_file(qemu_bin_from_bld_dir_path):
67 return qemu_bin_from_bld_dir_path
68
69
70 def _console_interaction(test, success_message, failure_message,
71 send_string, keep_sending=False, vm=None):
72 assert not keep_sending or send_string
73 if vm is None:
74 vm = test.vm
75 console = vm.console_socket.makefile()
76 console_logger = logging.getLogger('console')
77 while True:
78 if send_string:
79 vm.console_socket.sendall(send_string.encode())
80 if not keep_sending:
81 send_string = None # send only once
82 msg = console.readline().strip()
83 if not msg:
84 continue
85 console_logger.debug(msg)
86 if success_message in msg:
87 break
88 if failure_message and failure_message in msg:
89 console.close()
90 fail = 'Failure message found in console: %s' % failure_message
91 test.fail(fail)
92
93 def interrupt_interactive_console_until_pattern(test, success_message,
94 failure_message=None,
95 interrupt_string='\r'):
96 """
97 Keep sending a string to interrupt a console prompt, while logging the
98 console output. Typical use case is to break a boot loader prompt, such:
99
100 Press a key within 5 seconds to interrupt boot process.
101 5
102 4
103 3
104 2
105 1
106 Booting default image...
107
108 :param test: an Avocado test containing a VM that will have its console
109 read and probed for a success or failure message
110 :type test: :class:`avocado_qemu.Test`
111 :param success_message: if this message appears, test succeeds
112 :param failure_message: if this message appears, test fails
113 :param interrupt_string: a string to send to the console before trying
114 to read a new line
115 """
116 _console_interaction(test, success_message, failure_message,
117 interrupt_string, True)
118
119 def wait_for_console_pattern(test, success_message, failure_message=None,
120 vm=None):
121 """
122 Waits for messages to appear on the console, while logging the content
123
124 :param test: an Avocado test containing a VM that will have its console
125 read and probed for a success or failure message
126 :type test: :class:`avocado_qemu.Test`
127 :param success_message: if this message appears, test succeeds
128 :param failure_message: if this message appears, test fails
129 """
130 _console_interaction(test, success_message, failure_message, None, vm=vm)
131
132 def exec_command_and_wait_for_pattern(test, command,
133 success_message, failure_message=None):
134 """
135 Send a command to a console (appending CRLF characters), then wait
136 for success_message to appear on the console, while logging the.
137 content. Mark the test as failed if failure_message is found instead.
138
139 :param test: an Avocado test containing a VM that will have its console
140 read and probed for a success or failure message
141 :type test: :class:`avocado_qemu.Test`
142 :param command: the command to send
143 :param success_message: if this message appears, test succeeds
144 :param failure_message: if this message appears, test fails
145 """
146 _console_interaction(test, success_message, failure_message, command + '\r')
147
148 class Test(avocado.Test):
149 def _get_unique_tag_val(self, tag_name):
150 """
151 Gets a tag value, if unique for a key
152 """
153 vals = self.tags.get(tag_name, [])
154 if len(vals) == 1:
155 return vals.pop()
156 return None
157
158 def setUp(self):
159 self._vms = {}
160
161 self.arch = self.params.get('arch',
162 default=self._get_unique_tag_val('arch'))
163
164 self.machine = self.params.get('machine',
165 default=self._get_unique_tag_val('machine'))
166
167 default_qemu_bin = pick_default_qemu_bin(arch=self.arch)
168 self.qemu_bin = self.params.get('qemu_bin',
169 default=default_qemu_bin)
170 if self.qemu_bin is None:
171 self.cancel("No QEMU binary defined or found in the build tree")
172
173 def _new_vm(self, *args):
174 sd = tempfile.mkdtemp(prefix="avo_qemu_sock_")
175 vm = QEMUMachine(self.qemu_bin, sock_dir=sd)
176 if args:
177 vm.add_args(*args)
178 return vm
179
180 @property
181 def vm(self):
182 return self.get_vm(name='default')
183
184 def get_vm(self, *args, name=None):
185 if not name:
186 name = str(uuid.uuid4())
187 if self._vms.get(name) is None:
188 self._vms[name] = self._new_vm(*args)
189 if self.machine is not None:
190 self._vms[name].set_machine(self.machine)
191 return self._vms[name]
192
193 def tearDown(self):
194 for vm in self._vms.values():
195 vm.shutdown()
196
197 def fetch_asset(self, name,
198 asset_hash=None, algorithm=None,
199 locations=None, expire=None,
200 find_only=False, cancel_on_missing=True):
201 return super(Test, self).fetch_asset(name,
202 asset_hash=asset_hash,
203 algorithm=algorithm,
204 locations=locations,
205 expire=expire,
206 find_only=find_only,
207 cancel_on_missing=cancel_on_missing)