[fio.git] / t / fiotestlib.py

#!/usr/bin/env python3
"""
fiotestlib.py

This library contains FioTest objects that provide convenient means to run
different sorts of fio tests.

It also contains a test runner that runs an array of dictionary objects
describing fio tests.
"""

import os
import sys
import json
import locale
import logging
import platform
import traceback
import subprocess
from pathlib import Path


class FioTest():
    """Base for all fio tests."""

    def __init__(self, exe_path, parameters, success):
        self.exe_path = exe_path
        self.parameters = parameters
        self.success = success
        self.output = {}
        self.artifact_root = None
        self.testnum = None
        self.test_dir = None
        self.passed = True
        self.failure_reason = ''
        self.command_file = None
        self.stdout_file = None
        self.stderr_file = None
        self.exitcode_file = None

    def setup(self, artifact_root, testnum):
        """Setup instance variables for test."""

        self.artifact_root = artifact_root
        self.testnum = testnum
        self.test_dir = os.path.join(artifact_root, f"{testnum:04d}")
        if not os.path.exists(self.test_dir):
            os.mkdir(self.test_dir)

        self.command_file = os.path.join( self.test_dir,
                                         f"{os.path.basename(self.exe_path)}.command")
        self.stdout_file = os.path.join( self.test_dir,
                                        f"{os.path.basename(self.exe_path)}.stdout")
        self.stderr_file = os.path.join( self.test_dir,
                                        f"{os.path.basename(self.exe_path)}.stderr")
        self.exitcode_file = os.path.join( self.test_dir,
                                          f"{os.path.basename(self.exe_path)}.exitcode")

    def run(self):
        """Run the test."""

        raise NotImplementedError()

    def check_result(self):
        """Check test results."""

        raise NotImplementedError()


class FioExeTest(FioTest):
    """Test consists of an executable binary or script"""

    def __init__(self, exe_path, parameters, success):
        """Construct a FioExeTest which is a FioTest consisting of an
        executable binary or script.

        exe_path:       location of executable binary or script
        parameters:     list of parameters for executable
        success:        Definition of test success
        """

        FioTest.__init__(self, exe_path, parameters, success)

    def run(self):
        """Execute the binary or script described by this instance."""

        command = [self.exe_path] + self.parameters
        command_file = open(self.command_file, "w+",
                            encoding=locale.getpreferredencoding())
        command_file.write(f"{command}\n")
        command_file.close()

        stdout_file = open(self.stdout_file, "w+",
                           encoding=locale.getpreferredencoding())
        stderr_file = open(self.stderr_file, "w+",
                           encoding=locale.getpreferredencoding())
        exitcode_file = open(self.exitcode_file, "w+",
                             encoding=locale.getpreferredencoding())
        try:
            proc = None
            # Avoid using subprocess.run() here because when a timeout occurs,
            # fio will be stopped with SIGKILL. This does not give fio a
            # chance to clean up and means that child processes may continue
            # running and submitting IO.
            proc = subprocess.Popen(command,
                                    stdout=stdout_file,
                                    stderr=stderr_file,
                                    cwd=self.test_dir,
                                    universal_newlines=True)
            proc.communicate(timeout=self.success['timeout'])
            exitcode_file.write(f'{proc.returncode}\n')
            logging.debug("Test %d: return code: %d", self.testnum, proc.returncode)
            self.output['proc'] = proc
        except subprocess.TimeoutExpired:
            proc.terminate()
            proc.communicate()
            assert proc.poll()
            self.output['failure'] = 'timeout'
        except Exception:
            if proc:
                if not proc.poll():
                    proc.terminate()
                    proc.communicate()
            self.output['failure'] = 'exception'
            self.output['exc_info'] = sys.exc_info()
        finally:
            stdout_file.close()
            stderr_file.close()
            exitcode_file.close()

    def check_result(self):
        """Check results of test run."""

        if 'proc' not in self.output:
            if self.output['failure'] == 'timeout':
                self.failure_reason = f"{self.failure_reason} timeout,"
            else:
                assert self.output['failure'] == 'exception'
                self.failure_reason = '{0} exception: {1}, {2}'.format(
                    self.failure_reason, self.output['exc_info'][0],
                    self.output['exc_info'][1])

            self.passed = False
            return

        if 'zero_return' in self.success:
            if self.success['zero_return']:
                if self.output['proc'].returncode != 0:
                    self.passed = False
                    self.failure_reason = f"{self.failure_reason} non-zero return code,"
            else:
                if self.output['proc'].returncode == 0:
                    self.failure_reason = f"{self.failure_reason} zero return code,"
                    self.passed = False

        stderr_size = os.path.getsize(self.stderr_file)
        if 'stderr_empty' in self.success:
            if self.success['stderr_empty']:
                if stderr_size != 0:
                    self.failure_reason = f"{self.failure_reason} stderr not empty,"
                    self.passed = False
            else:
                if stderr_size == 0:
                    self.failure_reason = f"{self.failure_reason} stderr empty,"
                    self.passed = False


class FioJobTest(FioExeTest):
    """Test consists of a fio job"""

    def __init__(self, fio_path, fio_job, success, fio_pre_job=None,
                 fio_pre_success=None, output_format="normal"):
        """Construct a FioJobTest which is a FioExeTest consisting of a
        single fio job file with an optional setup step.

        fio_path:           location of fio executable
        fio_job:            location of fio job file
        success:            Definition of test success
        fio_pre_job:        fio job for preconditioning
        fio_pre_success:    Definition of test success for fio precon job
        output_format:      normal (default), json, jsonplus, or terse
        """

        self.fio_job = fio_job
        self.fio_pre_job = fio_pre_job
        self.fio_pre_success = fio_pre_success if fio_pre_success else success
        self.output_format = output_format
        self.precon_failed = False
        self.json_data = None
        self.fio_output = f"{os.path.basename(self.fio_job)}.output"
        self.fio_args = [
            "--max-jobs=16",
            f"--output-format={self.output_format}",
            f"--output={self.fio_output}",
            self.fio_job,
            ]
        FioExeTest.__init__(self, fio_path, self.fio_args, success)

    def setup(self, artifact_root, testnum):
        """Setup instance variables for fio job test."""

        super().setup(artifact_root, testnum)

        self.command_file = os.path.join(self.test_dir,
                                         f"{os.path.basename(self.fio_job)}.command")
        self.stdout_file = os.path.join(self.test_dir,
                                        f"{os.path.basename(self.fio_job)}.stdout")
        self.stderr_file = os.path.join(self.test_dir,
                                        f"{os.path.basename(self.fio_job)}.stderr")
        self.exitcode_file = os.path.join(self.test_dir,
                                          f"{os.path.basename(self.fio_job)}.exitcode")

    def run_pre_job(self):
        """Run fio job precondition step."""

        precon = FioJobTest(self.exe_path, self.fio_pre_job,
                            self.fio_pre_success,
                            output_format=self.output_format)
        precon.setup(self.artifact_root, self.testnum)
        precon.run()
        precon.check_result()
        self.precon_failed = not precon.passed
        self.failure_reason = precon.failure_reason

    def run(self):
        """Run fio job test."""

        if self.fio_pre_job:
            self.run_pre_job()

        if not self.precon_failed:
            super().run()
        else:
            logging.debug("Test %d: precondition step failed", self.testnum)

    @classmethod
    def get_file(cls, filename):
        """Safely read a file."""
        file_data = ''
        success = True

        try:
            with open(filename, "r", encoding=locale.getpreferredencoding()) as output_file:
                file_data = output_file.read()
        except OSError:
            success = False

        return file_data, success

    def get_file_fail(self, filename):
        """Safely read a file and fail the test upon error."""
        file_data = None

        try:
            with open(filename, "r", encoding=locale.getpreferredencoding()) as output_file:
                file_data = output_file.read()
        except OSError:
            self.failure_reason += f" unable to read file {filename}"
            self.passed = False

        return file_data

    def check_result(self):
        """Check fio job results."""

        if self.precon_failed:
            self.passed = False
            self.failure_reason = f"{self.failure_reason} precondition step failed,"
            return

        super().check_result()

        if not self.passed:
            return

        if 'json' not in self.output_format:
            return

        file_data = self.get_file_fail(os.path.join(self.test_dir, self.fio_output))
        if not file_data:
            return

        #
        # Sometimes fio informational messages are included at the top of the
        # JSON output, especially under Windows. Try to decode output as JSON
        # data, skipping everything until the first {
        #
        lines = file_data.splitlines()
        file_data = '\n'.join(lines[lines.index("{"):])
        try:
            self.json_data = json.loads(file_data)
        except json.JSONDecodeError:
            self.failure_reason = f"{self.failure_reason} unable to decode JSON data,"
            self.passed = False


def run_fio_tests(test_list, test_env, args):
    """
    Run tests as specified in test_list.
    """

    passed = 0
    failed = 0
    skipped = 0

    for config in test_list:
        if (args.skip and config['test_id'] in args.skip) or \
           (args.run_only and config['test_id'] not in args.run_only):
            skipped = skipped + 1
            print(f"Test {config['test_id']} SKIPPED (User request)")
            continue

        if issubclass(config['test_class'], FioJobTest):
            if config['pre_job']:
                fio_pre_job = os.path.join(test_env['fio_root'], 't', 'jobs',
                                           config['pre_job'])
            else:
                fio_pre_job = None
            if config['pre_success']:
                fio_pre_success = config['pre_success']
            else:
                fio_pre_success = None
            if 'output_format' in config:
                output_format = config['output_format']
            else:
                output_format = 'normal'
            test = config['test_class'](
                test_env['fio_path'],
                os.path.join(test_env['fio_root'], 't', 'jobs', config['job']),
                config['success'],
                fio_pre_job=fio_pre_job,
                fio_pre_success=fio_pre_success,
                output_format=output_format)
            desc = config['job']
        elif issubclass(config['test_class'], FioExeTest):
            exe_path = os.path.join(test_env['fio_root'], config['exe'])
            if config['parameters']:
                parameters = [p.format(fio_path=test_env['fio_path'], nvmecdev=args.nvmecdev)
                              for p in config['parameters']]
            else:
                parameters = []
            if Path(exe_path).suffix == '.py' and platform.system() == "Windows":
                parameters.insert(0, exe_path)
                exe_path = "python.exe"
            if config['test_id'] in test_env['pass_through']:
                parameters += test_env['pass_through'][config['test_id']].split()
            test = config['test_class'](exe_path, parameters,
                                        config['success'])
            desc = config['exe']
        else:
            print(f"Test {config['test_id']} FAILED: unable to process test config")
            failed = failed + 1
            continue

        if not args.skip_req:
            reqs_met = True
            for req in config['requirements']:
                reqs_met, reason = req()
                logging.debug("Test %d: Requirement '%s' met? %s", config['test_id'], reason,
                              reqs_met)
                if not reqs_met:
                    break
            if not reqs_met:
                print(f"Test {config['test_id']} SKIPPED ({reason}) {desc}")
                skipped = skipped + 1
                continue

        try:
            test.setup(test_env['artifact_root'], config['test_id'])
            test.run()
            test.check_result()
        except KeyboardInterrupt:
            break
        except Exception as e:
            test.passed = False
            test.failure_reason += str(e)
            logging.debug("Test %d exception:\n%s\n", config['test_id'], traceback.format_exc())
        if test.passed:
            result = "PASSED"
            passed = passed + 1
        else:
            result = f"FAILED: {test.failure_reason}"
            failed = failed + 1
            contents, _ = FioJobTest.get_file(test.stderr_file)
            logging.debug("Test %d: stderr:\n%s", config['test_id'], contents)
            contents, _ = FioJobTest.get_file(test.stdout_file)
            logging.debug("Test %d: stdout:\n%s", config['test_id'], contents)
        print(f"Test {config['test_id']} {result} {desc}")

    print(f"{passed} test(s) passed, {failed} failed, {skipped} skipped")

    return passed, failed, skipped
Commit	Line	Data
fb551941 VF	1	#!/usr/bin/env python3
	2	"""
	3	fiotestlib.py
	4
	5	This library contains FioTest objects that provide convenient means to run
	6	different sorts of fio tests.
	7
	8	It also contains a test runner that runs an array of dictionary objects
	9	describing fio tests.
	10	"""
	11
	12	import os
	13	import sys
	14	import json
	15	import locale
	16	import logging
	17	import platform
	18	import traceback
	19	import subprocess
	20	from pathlib import Path
	21
	22
	23	class FioTest():
	24	"""Base for all fio tests."""
	25
	26	def __init__(self, exe_path, parameters, success):
	27	self.exe_path = exe_path
	28	self.parameters = parameters
	29	self.success = success
	30	self.output = {}
	31	self.artifact_root = None
	32	self.testnum = None
	33	self.test_dir = None
	34	self.passed = True
	35	self.failure_reason = ''
	36	self.command_file = None
	37	self.stdout_file = None
	38	self.stderr_file = None
	39	self.exitcode_file = None
	40
	41	def setup(self, artifact_root, testnum):
	42	"""Setup instance variables for test."""
	43
	44	self.artifact_root = artifact_root
	45	self.testnum = testnum
	46	self.test_dir = os.path.join(artifact_root, f"{testnum:04d}")
	47	if not os.path.exists(self.test_dir):
	48	os.mkdir(self.test_dir)
	49
	50	self.command_file = os.path.join( self.test_dir,
	51	f"{os.path.basename(self.exe_path)}.command")
	52	self.stdout_file = os.path.join( self.test_dir,
	53	f"{os.path.basename(self.exe_path)}.stdout")
	54	self.stderr_file = os.path.join( self.test_dir,
	55	f"{os.path.basename(self.exe_path)}.stderr")
	56	self.exitcode_file = os.path.join( self.test_dir,
	57	f"{os.path.basename(self.exe_path)}.exitcode")
	58
	59	def run(self):
	60	"""Run the test."""
	61
	62	raise NotImplementedError()
	63
	64	def check_result(self):
65	"""Check test results."""
66
67	raise NotImplementedError()
68
69
70	class FioExeTest(FioTest):
71	"""Test consists of an executable binary or script"""
72
73	def __init__(self, exe_path, parameters, success):
74	"""Construct a FioExeTest which is a FioTest consisting of an
75	executable binary or script.
76
77	exe_path: location of executable binary or script
78	parameters: list of parameters for executable
79	success: Definition of test success
80	"""
81
82	FioTest.__init__(self, exe_path, parameters, success)
83
84	def run(self):
85	"""Execute the binary or script described by this instance."""
86
87	command = [self.exe_path] + self.parameters
88	command_file = open(self.command_file, "w+",
89	encoding=locale.getpreferredencoding())
90	command_file.write(f"{command}\n")
91	command_file.close()
92
93	stdout_file = open(self.stdout_file, "w+",
94	encoding=locale.getpreferredencoding())
95	stderr_file = open(self.stderr_file, "w+",
96	encoding=locale.getpreferredencoding())
97	exitcode_file = open(self.exitcode_file, "w+",
98	encoding=locale.getpreferredencoding())
99	try:
100	proc = None
101	# Avoid using subprocess.run() here because when a timeout occurs,
102	# fio will be stopped with SIGKILL. This does not give fio a
103	# chance to clean up and means that child processes may continue
104	# running and submitting IO.
105	proc = subprocess.Popen(command,
106	stdout=stdout_file,
107	stderr=stderr_file,
108	cwd=self.test_dir,
109	universal_newlines=True)
110	proc.communicate(timeout=self.success['timeout'])
111	exitcode_file.write(f'{proc.returncode}\n')
112	logging.debug("Test %d: return code: %d", self.testnum, proc.returncode)
113	self.output['proc'] = proc
114	except subprocess.TimeoutExpired:
115	proc.terminate()
116	proc.communicate()
117	assert proc.poll()
118	self.output['failure'] = 'timeout'
119	except Exception:
120	if proc:
121	if not proc.poll():
122	proc.terminate()
123	proc.communicate()
124	self.output['failure'] = 'exception'
125	self.output['exc_info'] = sys.exc_info()
126	finally:
127	stdout_file.close()
128	stderr_file.close()
129	exitcode_file.close()
130
131	def check_result(self):
132	"""Check results of test run."""
133
134	if 'proc' not in self.output:
135	if self.output['failure'] == 'timeout':
136	self.failure_reason = f"{self.failure_reason} timeout,"
137	else:
138	assert self.output['failure'] == 'exception'
139	self.failure_reason = '{0} exception: {1}, {2}'.format(
140	self.failure_reason, self.output['exc_info'][0],
141	self.output['exc_info'][1])
142
143	self.passed = False
144	return
145
146	if 'zero_return' in self.success:
147	if self.success['zero_return']:
148	if self.output['proc'].returncode != 0:
149	self.passed = False
150	self.failure_reason = f"{self.failure_reason} non-zero return code,"
151	else:
152	if self.output['proc'].returncode == 0:
153	self.failure_reason = f"{self.failure_reason} zero return code,"
154	self.passed = False
155
156	stderr_size = os.path.getsize(self.stderr_file)
157	if 'stderr_empty' in self.success:
158	if self.success['stderr_empty']:
159	if stderr_size != 0:
160	self.failure_reason = f"{self.failure_reason} stderr not empty,"
161	self.passed = False
162	else:
163	if stderr_size == 0:
164	self.failure_reason = f"{self.failure_reason} stderr empty,"
165	self.passed = False
166
167
168	class FioJobTest(FioExeTest):
169	"""Test consists of a fio job"""
170
171	def __init__(self, fio_path, fio_job, success, fio_pre_job=None,
172	fio_pre_success=None, output_format="normal"):
173	"""Construct a FioJobTest which is a FioExeTest consisting of a
174	single fio job file with an optional setup step.
175
176	fio_path: location of fio executable
177	fio_job: location of fio job file
178	success: Definition of test success
179	fio_pre_job: fio job for preconditioning
180	fio_pre_success: Definition of test success for fio precon job
181	output_format: normal (default), json, jsonplus, or terse
182	"""
183
184	self.fio_job = fio_job
185	self.fio_pre_job = fio_pre_job
186	self.fio_pre_success = fio_pre_success if fio_pre_success else success
187	self.output_format = output_format
188	self.precon_failed = False
189	self.json_data = None
190	self.fio_output = f"{os.path.basename(self.fio_job)}.output"
191	self.fio_args = [
192	"--max-jobs=16",
193	f"--output-format={self.output_format}",
194	f"--output={self.fio_output}",
195	self.fio_job,
196	]
197	FioExeTest.__init__(self, fio_path, self.fio_args, success)
198
199	def setup(self, artifact_root, testnum):
200	"""Setup instance variables for fio job test."""
201
202	super().setup(artifact_root, testnum)
203
204	self.command_file = os.path.join(self.test_dir,
205	f"{os.path.basename(self.fio_job)}.command")
206	self.stdout_file = os.path.join(self.test_dir,
207	f"{os.path.basename(self.fio_job)}.stdout")
208	self.stderr_file = os.path.join(self.test_dir,
209	f"{os.path.basename(self.fio_job)}.stderr")
210	self.exitcode_file = os.path.join(self.test_dir,
211	f"{os.path.basename(self.fio_job)}.exitcode")
212
213	def run_pre_job(self):
214	"""Run fio job precondition step."""
215
216	precon = FioJobTest(self.exe_path, self.fio_pre_job,
217	self.fio_pre_success,
218	output_format=self.output_format)
219	precon.setup(self.artifact_root, self.testnum)
220	precon.run()
221	precon.check_result()
222	self.precon_failed = not precon.passed
223	self.failure_reason = precon.failure_reason
224
225	def run(self):
226	"""Run fio job test."""
227
228	if self.fio_pre_job:
229	self.run_pre_job()
230
231	if not self.precon_failed:
232	super().run()
233	else:
234	logging.debug("Test %d: precondition step failed", self.testnum)
235
236	@classmethod
237	def get_file(cls, filename):
238	"""Safely read a file."""
239	file_data = ''
240	success = True
241
242	try:
243	with open(filename, "r", encoding=locale.getpreferredencoding()) as output_file:
244	file_data = output_file.read()
245	except OSError:
246	success = False
247
248	return file_data, success
249
250	def get_file_fail(self, filename):
251	"""Safely read a file and fail the test upon error."""
252	file_data = None
253
254	try:
255	with open(filename, "r", encoding=locale.getpreferredencoding()) as output_file:
256	file_data = output_file.read()
257	except OSError:
258	self.failure_reason += f" unable to read file {filename}"
259	self.passed = False
260
261	return file_data
262
263	def check_result(self):
264	"""Check fio job results."""
265
266	if self.precon_failed:
267	self.passed = False
268	self.failure_reason = f"{self.failure_reason} precondition step failed,"
269	return
270
271	super().check_result()
272
273	if not self.passed:
274	return
275
276	if 'json' not in self.output_format:
277	return
278
279	file_data = self.get_file_fail(os.path.join(self.test_dir, self.fio_output))
280	if not file_data:
281	return
282
283	#
284	# Sometimes fio informational messages are included at the top of the
285	# JSON output, especially under Windows. Try to decode output as JSON
286	# data, skipping everything until the first {
287	#
288	lines = file_data.splitlines()
289	file_data = '\n'.join(lines[lines.index("{"):])
290	try:
291	self.json_data = json.loads(file_data)
292	except json.JSONDecodeError:
293	self.failure_reason = f"{self.failure_reason} unable to decode JSON data,"
294	self.passed = False
295
296
297	def run_fio_tests(test_list, test_env, args):
298	"""
299	Run tests as specified in test_list.
300	"""
301
302	passed = 0
303	failed = 0
304	skipped = 0
305
306	for config in test_list:
307	if (args.skip and config['test_id'] in args.skip) or \
308	(args.run_only and config['test_id'] not in args.run_only):
309	skipped = skipped + 1
310	print(f"Test {config['test_id']} SKIPPED (User request)")
311	continue
312
313	if issubclass(config['test_class'], FioJobTest):
314	if config['pre_job']:
315	fio_pre_job = os.path.join(test_env['fio_root'], 't', 'jobs',
316	config['pre_job'])
317	else:
318	fio_pre_job = None
319	if config['pre_success']:
320	fio_pre_success = config['pre_success']
321	else:
322	fio_pre_success = None
323	if 'output_format' in config:
324	output_format = config['output_format']
325	else:
326	output_format = 'normal'
327	test = config['test_class'](
328	test_env['fio_path'],
329	os.path.join(test_env['fio_root'], 't', 'jobs', config['job']),
330	config['success'],
331	fio_pre_job=fio_pre_job,
332	fio_pre_success=fio_pre_success,
333	output_format=output_format)
334	desc = config['job']
335	elif issubclass(config['test_class'], FioExeTest):
336	exe_path = os.path.join(test_env['fio_root'], config['exe'])
337	if config['parameters']:
338	parameters = [p.format(fio_path=test_env['fio_path'], nvmecdev=args.nvmecdev)
339	for p in config['parameters']]
340	else:
341	parameters = []
342	if Path(exe_path).suffix == '.py' and platform.system() == "Windows":
343	parameters.insert(0, exe_path)
344	exe_path = "python.exe"
345	if config['test_id'] in test_env['pass_through']:
346	parameters += test_env['pass_through'][config['test_id']].split()
347	test = config['test_class'](exe_path, parameters,
348	config['success'])
349	desc = config['exe']
350	else:
351	print(f"Test {config['test_id']} FAILED: unable to process test config")
352	failed = failed + 1
353	continue
354
355	if not args.skip_req:
356	reqs_met = True
357	for req in config['requirements']:
358	reqs_met, reason = req()
359	logging.debug("Test %d: Requirement '%s' met? %s", config['test_id'], reason,
360	reqs_met)
361	if not reqs_met:
362	break
363	if not reqs_met:
364	print(f"Test {config['test_id']} SKIPPED ({reason}) {desc}")
365	skipped = skipped + 1
366	continue
367
368	try:
369	test.setup(test_env['artifact_root'], config['test_id'])
370	test.run()
371	test.check_result()
372	except KeyboardInterrupt:
373	break
374	except Exception as e:
375	test.passed = False
376	test.failure_reason += str(e)
377	logging.debug("Test %d exception:\n%s\n", config['test_id'], traceback.format_exc())
378	if test.passed:
379	result = "PASSED"
380	passed = passed + 1
381	else:
382	result = f"FAILED: {test.failure_reason}"
383	failed = failed + 1
384	contents, _ = FioJobTest.get_file(test.stderr_file)
385	logging.debug("Test %d: stderr:\n%s", config['test_id'], contents)
386	contents, _ = FioJobTest.get_file(test.stdout_file)
387	logging.debug("Test %d: stdout:\n%s", config['test_id'], contents)
388	print(f"Test {config['test_id']} {result} {desc}")
389
390	print(f"{passed} test(s) passed, {failed} failed, {skipped} skipped")
391
392	return passed, failed, skipped