t/readonly: adapt to use fiotestlib

[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
from fiotestcommon import get_file, SUCCESS_DEFAULT


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

    def __init__(self, exe_path, success, testnum, artifact_root):
        self.success = success
        self.testnum = testnum
        self.output = {}
        self.passed = True
        self.failure_reason = ''
        self.parameters = None
        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.parameters = parameters
        if not os.path.exists(self.paths['test_dir']):
            os.mkdir(self.paths['test_dir'])

    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 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(" ".join(command))

        try:
            with open(self.filenames['stdout'], "w+",
                      encoding=locale.getpreferredencoding()) as stdout_file, \
                open(self.filenames['stderr'], "w+",
                     encoding=locale.getpreferredencoding()) as stderr_file, \
                open(self.filenames['exitcode'], "w+",
                     encoding=locale.getpreferredencoding()) as exitcode_file:
                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.paths['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()

    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 = f'{self.failure_reason} exception: ' + \
                f'{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.filenames['stderr'])
        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 FioJobFileTest(FioExeTest):
    """Test consists of a fio job with options in a job file."""

    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.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

        super().__init__(fio_path, success, testnum, artifact_root)

    def setup(self, parameters=None):
        """Setup instance variables for fio job test."""

        self.filenames['fio_output'] = f"{os.path.basename(self.fio_job)}.output"
        fio_args = [
            "--max-jobs=16",
            f"--output-format={self.output_format}",
            f"--output={self.filenames['fio_output']}",
            self.fio_job,
            ]

        super().setup(fio_args)

        # Update the filenames from the default
        self.filenames['cmd'] = os.path.join(self.paths['test_dir'],
                                             f"{os.path.basename(self.fio_job)}.command")
        self.filenames['stdout'] = os.path.join(self.paths['test_dir'],
                                                f"{os.path.basename(self.fio_job)}.stdout")
        self.filenames['stderr'] = os.path.join(self.paths['test_dir'],
                                                f"{os.path.basename(self.fio_job)}.stderr")
        self.filenames['exitcode'] = os.path.join(self.paths['test_dir'],
                                                  f"{os.path.basename(self.fio_job)}.exitcode")

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

        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()
        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)

    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.paths['test_dir'],
                                                    self.filenames['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


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

        super().__init__(fio_path, success, testnum, artifact_root)

        filename_stub = f"{self.basename}{self.testnum:03d}"
        self.filenames['cmd'] = os.path.join(self.paths['test_dir'], f"{filename_stub}.command")
        self.filenames['stdout'] = os.path.join(self.paths['test_dir'], f"{filename_stub}.stdout")
        self.filenames['stderr'] = os.path.join(self.paths['test_dir'], f"{filename_stub}.stderr")
        self.filenames['output'] = os.path.abspath(os.path.join(self.paths['test_dir'],
                                     f"{filename_stub}.output"))
        self.filenames['exitcode'] = os.path.join(self.paths['test_dir'],
                                                  f"{filename_stub}.exitcode")

    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

    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.
    """

    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'], FioJobFileTest):
            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'],
                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 = []
            if config['parameters']:
                parameters = [p.format(fio_path=test_env['fio_path'], nvmecdev=args.nvmecdev)
                              for p in config['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,
                    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 'requirements' in config and 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(parameters)
            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, _ = get_file(test.filenames['stderr'])
            logging.debug("Test %d: stderr:\n%s", config['test_id'], contents)
            contents, _ = get_file(test.filenames['stdout'])
            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