import os
import sys
-import json
import time
import shutil
import logging
import argparse
-import platform
-import traceback
-import subprocess
-import multiprocessing
from pathlib import Path
from statsmodels.sandbox.stats.runs import runstest_1samp
-
-
-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+")
- command_file.write(f"{command}\n")
- command_file.close()
-
- stdout_file = open(self.stdout_file, "w+")
- stderr_file = open(self.stderr_file, "w+")
- exitcode_file = open(self.exitcode_file, "w+")
- 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") 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") 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
+from fiotestlib import FioExeTest, FioJobTest, run_fio_tests
+from fiotestcommon import *
class FioJobTest_t0005(FioJobTest):
self.passed = False
-class Requirements():
- """Requirements consists of multiple run environment characteristics.
- These are to determine if a particular test can be run"""
-
- _linux = False
- _libaio = False
- _io_uring = False
- _zbd = False
- _root = False
- _zoned_nullb = False
- _not_macos = False
- _not_windows = False
- _unittests = False
- _cpucount4 = False
- _nvmecdev = False
-
- def __init__(self, fio_root, args):
- Requirements._not_macos = platform.system() != "Darwin"
- Requirements._not_windows = platform.system() != "Windows"
- Requirements._linux = platform.system() == "Linux"
-
- if Requirements._linux:
- config_file = os.path.join(fio_root, "config-host.h")
- contents, success = FioJobTest.get_file(config_file)
- if not success:
- print(f"Unable to open {config_file} to check requirements")
- Requirements._zbd = True
- else:
- Requirements._zbd = "CONFIG_HAS_BLKZONED" in contents
- Requirements._libaio = "CONFIG_LIBAIO" in contents
-
- contents, success = FioJobTest.get_file("/proc/kallsyms")
- if not success:
- print("Unable to open '/proc/kallsyms' to probe for io_uring support")
- else:
- Requirements._io_uring = "io_uring_setup" in contents
-
- Requirements._root = os.geteuid() == 0
- if Requirements._zbd and Requirements._root:
- try:
- subprocess.run(["modprobe", "null_blk"],
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- if os.path.exists("/sys/module/null_blk/parameters/zoned"):
- Requirements._zoned_nullb = True
- except Exception:
- pass
-
- if platform.system() == "Windows":
- utest_exe = "unittest.exe"
- else:
- utest_exe = "unittest"
- unittest_path = os.path.join(fio_root, "unittests", utest_exe)
- Requirements._unittests = os.path.exists(unittest_path)
-
- Requirements._cpucount4 = multiprocessing.cpu_count() >= 4
- Requirements._nvmecdev = args.nvmecdev
-
- req_list = [
- Requirements.linux,
- Requirements.libaio,
- Requirements.io_uring,
- Requirements.zbd,
- Requirements.root,
- Requirements.zoned_nullb,
- Requirements.not_macos,
- Requirements.not_windows,
- Requirements.unittests,
- Requirements.cpucount4,
- Requirements.nvmecdev,
- ]
- for req in req_list:
- value, desc = req()
- logging.debug("Requirements: Requirement '%s' met? %s", desc, value)
-
- @classmethod
- def linux(cls):
- """Are we running on Linux?"""
- return Requirements._linux, "Linux required"
-
- @classmethod
- def libaio(cls):
- """Is libaio available?"""
- return Requirements._libaio, "libaio required"
-
- @classmethod
- def io_uring(cls):
- """Is io_uring available?"""
- return Requirements._io_uring, "io_uring required"
-
- @classmethod
- def zbd(cls):
- """Is ZBD support available?"""
- return Requirements._zbd, "Zoned block device support required"
-
- @classmethod
- def root(cls):
- """Are we running as root?"""
- return Requirements._root, "root required"
-
- @classmethod
- def zoned_nullb(cls):
- """Are zoned null block devices available?"""
- return Requirements._zoned_nullb, "Zoned null block device support required"
-
- @classmethod
- def not_macos(cls):
- """Are we running on a platform other than macOS?"""
- return Requirements._not_macos, "platform other than macOS required"
-
- @classmethod
- def not_windows(cls):
- """Are we running on a platform other than Windws?"""
- return Requirements._not_windows, "platform other than Windows required"
-
- @classmethod
- def unittests(cls):
- """Were unittests built?"""
- return Requirements._unittests, "Unittests support required"
-
- @classmethod
- def cpucount4(cls):
- """Do we have at least 4 CPUs?"""
- return Requirements._cpucount4, "4+ CPUs required"
-
- @classmethod
- def nvmecdev(cls):
- """Do we have an NVMe character device to test?"""
- return Requirements._nvmecdev, "NVMe character device test target required"
-
-
-SUCCESS_DEFAULT = {
- 'zero_return': True,
- 'stderr_empty': True,
- 'timeout': 600,
- }
-SUCCESS_NONZERO = {
- 'zero_return': False,
- 'stderr_empty': False,
- 'timeout': 600,
- }
-SUCCESS_STDERR = {
- 'zero_return': True,
- 'stderr_empty': False,
- 'timeout': 600,
- }
TEST_LIST = [
{
'test_id': 1,
print(f"Artifact directory is {artifact_root}")
if not args.skip_req:
- req = Requirements(fio_root, args)
-
- 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(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'](
- fio_path,
- os.path.join(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(fio_root, config['exe'])
- if config['parameters']:
- parameters = [p.format(fio_path=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 pass_through:
- parameters += 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(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")
-
+ Requirements(fio_root, args)
+
+ test_env = {
+ 'fio_path': fio_path,
+ 'fio_root': fio_root,
+ 'artifact_root': artifact_root,
+ 'pass_through': pass_through,
+ }
+ _, failed, _ = run_fio_tests(TEST_LIST, test_env, args)
sys.exit(failed)
projects / fio.git / blobdiff