import traceback
import subprocess
from pathlib import Path
-from fiotestcommon import get_file
+from fiotestcommon import get_file, SUCCESS_DEFAULT
class FioTest():
"""Base for all fio tests."""
- def __init__(self, exe_path, success):
- self.paths = {'exe': exe_path}
+ def __init__(self, exe_path, success, testnum, artifact_root):
self.success = success
+ self.testnum = testnum
self.output = {}
- self.testnum = None
self.passed = True
self.failure_reason = ''
- self.filenames = {}
self.parameters = None
-
- def setup(self, artifact_root, testnum, parameters):
+ self.paths = {
+ 'exe': exe_path,
+ 'artifacts': artifact_root,
+ 'test_dir': os.path.join(artifact_root, \
+ f"{testnum:04d}"),
+ }
+ self.filenames = {
+ 'cmd': os.path.join(self.paths['test_dir'], \
+ f"{os.path.basename(self.paths['exe'])}.command"),
+ 'stdout': os.path.join(self.paths['test_dir'], \
+ f"{os.path.basename(self.paths['exe'])}.stdout"),
+ 'stderr': os.path.join(self.paths['test_dir'], \
+ f"{os.path.basename(self.paths['exe'])}.stderr"),
+ 'exitcode': os.path.join(self.paths['test_dir'], \
+ f"{os.path.basename(self.paths['exe'])}.exitcode"),
+ }
+
+ def setup(self, parameters):
"""Setup instance variables for test."""
- self.testnum = testnum
self.parameters = parameters
- self.paths['artifacts'] = artifact_root
- self.paths['test_dir'] = os.path.join(artifact_root, f"{testnum:04d}")
if not os.path.exists(self.paths['test_dir']):
os.mkdir(self.paths['test_dir'])
- self.filenames['cmd'] = os.path.join(self.paths['test_dir'],
- f"{os.path.basename(self.paths['exe'])}.command")
- self.filenames['stdout'] = os.path.join(self.paths['test_dir'],
- f"{os.path.basename(self.paths['exe'])}.stdout")
- self.filenames['stderr'] = os.path.join(self.paths['test_dir'],
- f"{os.path.basename(self.paths['exe'])}.stderr")
- self.filenames['exitcode'] = os.path.join(self.paths['test_dir'],
- f"{os.path.basename(self.paths['exe'])}.exitcode")
-
def run(self):
"""Run the test."""
class FioExeTest(FioTest):
"""Test consists of an executable binary or script"""
- def __init__(self, exe_path, success):
- """Construct a FioExeTest which is a FioTest consisting of an
- executable binary or script.
-
- exe_path: location of executable binary or script
- success: Definition of test success
- """
-
- FioTest.__init__(self, exe_path, success)
-
def run(self):
"""Execute the binary or script described by this instance."""
command = [self.paths['exe']] + self.parameters
with open(self.filenames['cmd'], "w+",
encoding=locale.getpreferredencoding()) as command_file:
- command_file.write(f"{command}\n")
+ command_file.write(" \\\n ".join(command))
try:
with open(self.filenames['stdout'], "w+",
class FioJobFileTest(FioExeTest):
"""Test consists of a fio job with options in a job file."""
- def __init__(self, fio_path, fio_job, success, fio_pre_job=None,
- fio_pre_success=None, output_format="normal"):
+ def __init__(self, fio_path, fio_job, success, testnum, artifact_root,
+ fio_pre_job=None, fio_pre_success=None,
+ output_format="normal"):
"""Construct a FioJobFileTest 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
+ testnum: test ID
+ artifact_root: root directory for artifacts
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.precon_failed = False
self.json_data = None
- FioExeTest.__init__(self, fio_path, success)
+ super().__init__(fio_path, success, testnum, artifact_root)
- def setup(self, artifact_root, testnum, parameters=None):
+ def setup(self, parameters=None):
"""Setup instance variables for fio job test."""
self.filenames['fio_output'] = f"{os.path.basename(self.fio_job)}.output"
self.fio_job,
]
- super().setup(artifact_root, testnum, fio_args)
+ super().setup(fio_args)
# Update the filenames from the default
self.filenames['cmd'] = os.path.join(self.paths['test_dir'],
precon = FioJobFileTest(self.paths['exe'], self.fio_pre_job,
self.fio_pre_success,
+ self.testnum,
+ self.paths['artifacts'],
output_format=self.output_format)
- precon.setup(self.paths['artifacts'], self.testnum)
+ precon.setup()
precon.run()
precon.check_result()
self.precon_failed = not precon.passed
self.passed = False
+class FioJobCmdTest(FioExeTest):
+ """This runs a fio job with options specified on the command line."""
+
+ def __init__(self, fio_path, success, testnum, artifact_root, fio_opts, basename=None):
+
+ self.basename = basename if basename else os.path.basename(fio_path)
+ self.fio_opts = fio_opts
+ self.json_data = None
+ self.iops_log_lines = None
+
+ super().__init__(fio_path, success, testnum, artifact_root)
+
+ filename_stub = os.path.join(self.paths['test_dir'], f"{self.basename}{self.testnum:03d}")
+ self.filenames['cmd'] = f"{filename_stub}.command"
+ self.filenames['stdout'] = f"{filename_stub}.stdout"
+ self.filenames['stderr'] = f"{filename_stub}.stderr"
+ self.filenames['output'] = os.path.abspath(f"{filename_stub}.output")
+ self.filenames['exitcode'] = f"{filename_stub}.exitcode"
+ self.filenames['iopslog'] = os.path.abspath(f"{filename_stub}")
+
+ def run(self):
+ super().run()
+
+ if 'output-format' in self.fio_opts and 'json' in \
+ self.fio_opts['output-format']:
+ if not self.get_json():
+ print('Unable to decode JSON data')
+ self.passed = False
+
+ if any('--write_iops_log=' in param for param in self.parameters):
+ self.get_iops_log()
+
+ def get_iops_log(self):
+ """Read IOPS log from the first job."""
+
+ log_filename = self.filenames['iopslog'] + "_iops.1.log"
+ with open(log_filename, 'r', encoding=locale.getpreferredencoding()) as iops_file:
+ self.iops_log_lines = iops_file.read()
+
+ def get_json(self):
+ """Convert fio JSON output into a python JSON object"""
+
+ filename = self.filenames['output']
+ with open(filename, 'r', encoding=locale.getpreferredencoding()) as file:
+ file_data = file.read()
+
+ #
+ # Sometimes fio informational messages are included at the top of the
+ # JSON output, especially under Windows. Try to decode output as JSON
+ # data, lopping off up to the first four lines
+ #
+ lines = file_data.splitlines()
+ for i in range(5):
+ file_data = '\n'.join(lines[i:])
+ try:
+ self.json_data = json.loads(file_data)
+ except json.JSONDecodeError:
+ continue
+ else:
+ return True
+
+ return False
+
+ @staticmethod
+ def check_empty(job):
+ """
+ Make sure JSON data is empty.
+
+ Some data structures should be empty. This function makes sure that they are.
+
+ job JSON object that we need to check for emptiness
+ """
+
+ return job['total_ios'] == 0 and \
+ job['slat_ns']['N'] == 0 and \
+ job['clat_ns']['N'] == 0 and \
+ job['lat_ns']['N'] == 0
+
+ def check_all_ddirs(self, ddir_nonzero, job):
+ """
+ Iterate over the data directions and check whether each is
+ appropriately empty or not.
+ """
+
+ retval = True
+ ddirlist = ['read', 'write', 'trim']
+
+ for ddir in ddirlist:
+ if ddir in ddir_nonzero:
+ if self.check_empty(job[ddir]):
+ print(f"Unexpected zero {ddir} data found in output")
+ retval = False
+ else:
+ if not self.check_empty(job[ddir]):
+ print(f"Unexpected {ddir} data found in output")
+ retval = False
+
+ return retval
+
+
def run_fio_tests(test_list, test_env, args):
"""
Run tests as specified in test_list.
test_env['fio_path'],
os.path.join(test_env['fio_root'], 't', 'jobs', config['job']),
config['success'],
+ config['test_id'],
+ test_env['artifact_root'],
fio_pre_job=fio_pre_job,
fio_pre_success=fio_pre_success,
output_format=output_format)
desc = config['job']
parameters = []
+ elif issubclass(config['test_class'], FioJobCmdTest):
+ if not 'success' in config:
+ config['success'] = SUCCESS_DEFAULT
+ test = config['test_class'](test_env['fio_path'],
+ config['success'],
+ config['test_id'],
+ test_env['artifact_root'],
+ config['fio_opts'],
+ test_env['basename'])
+ desc = config['test_id']
+ parameters = config
elif issubclass(config['test_class'], FioExeTest):
exe_path = os.path.join(test_env['fio_root'], config['exe'])
parameters = []
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, config['success'])
+ test = config['test_class'](
+ exe_path,
+ config['success'],
+ config['test_id'],
+ test_env['artifact_root'])
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:
+ if 'requirements' in config and not args.skip_req:
reqs_met = True
for req in config['requirements']:
reqs_met, reason = req()
continue
try:
- test.setup(test_env['artifact_root'], config['test_id'], parameters)
+ test.setup(parameters)
test.run()
test.check_result()
except KeyboardInterrupt:
projects / fio.git / blobdiff