-class FioTest(object):
- """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, "{:04d}".format(testnum))
- if not os.path.exists(self.test_dir):
- os.mkdir(self.test_dir)
-
- self.command_file = os.path.join(
- self.test_dir,
- "{0}.command".format(os.path.basename(self.exe_path)))
- self.stdout_file = os.path.join(
- self.test_dir,
- "{0}.stdout".format(os.path.basename(self.exe_path)))
- self.stderr_file = os.path.join(
- self.test_dir,
- "{0}.stderr".format(os.path.basename(self.exe_path)))
- self.exitcode_file = os.path.join(
- self.test_dir,
- "{0}.exitcode".format(os.path.basename(self.exe_path)))
-
- 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("%s\n" % command)
- 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('{0}\n'.format(proc.returncode))
- 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 = "{0} timeout,".format(self.failure_reason)
- 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 = "{0} non-zero return code,".format(self.failure_reason)
- else:
- if self.output['proc'].returncode == 0:
- self.failure_reason = "{0} zero return code,".format(self.failure_reason)
- 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 = "{0} stderr not empty,".format(self.failure_reason)
- self.passed = False
- else:
- if stderr_size == 0:
- self.failure_reason = "{0} stderr empty,".format(self.failure_reason)
- 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 = "{0}.output".format(os.path.basename(self.fio_job))
- self.fio_args = [
- "--max-jobs=16",
- "--output-format={0}".format(self.output_format),
- "--output={0}".format(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(FioJobTest, self).setup(artifact_root, testnum)
-
- self.command_file = os.path.join(
- self.test_dir,
- "{0}.command".format(os.path.basename(self.fio_job)))
- self.stdout_file = os.path.join(
- self.test_dir,
- "{0}.stdout".format(os.path.basename(self.fio_job)))
- self.stderr_file = os.path.join(
- self.test_dir,
- "{0}.stderr".format(os.path.basename(self.fio_job)))
- self.exitcode_file = os.path.join(
- self.test_dir,
- "{0}.exitcode".format(os.path.basename(self.fio_job)))
-
- 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(FioJobTest, self).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 check_result(self):
- """Check fio job results."""
-
- if self.precon_failed:
- self.passed = False
- self.failure_reason = "{0} precondition step failed,".format(self.failure_reason)
- return
-
- super(FioJobTest, self).check_result()
-
- if not self.passed:
- return
-
- if 'json' not in self.output_format:
- return
-
- file_data, success = self.get_file(os.path.join(self.test_dir, self.fio_output))
- if not success:
- self.failure_reason = "{0} unable to open output file,".format(self.failure_reason)
- self.passed = False
- 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 = "{0} unable to decode JSON data,".format(self.failure_reason)
- self.passed = False
-
-
-class FioJobTest_t0005(FioJobTest):